1// @effect-diagnostics globalTimers:off - Shelf placement timers run in web and native clients, outside an Effect runtime.
2import { threadRuntimeIsActive, type EnvironmentThreadShell } from "./models.ts";
3import { toSortableTimestamp } from "./threadSort.ts";
4import { effectiveSnoozed } from "./threadSettled.ts";
5
6type WorkingThreadInput = Pick<
7 EnvironmentThreadShell,
8 | "hasActionableProposedPlan"
9 | "hasPendingApprovals"
10 | "hasPendingUserInput"
11 | "pendingSubagentRequest"
12 | "interactionMode"
13 | "latestRun"
14 | "runtime"
15>;
16
17export function isThreadWorking(thread: WorkingThreadInput): boolean {
18 if (thread.hasPendingApprovals || thread.hasPendingUserInput) return false;
19 if (thread.pendingSubagentRequest != null) return false;
20 if (!threadRuntimeIsActive(thread.runtime) && thread.runtime?.status !== "idle") return false;
21 const run = thread.latestRun;
22 const runSettled =
23 run !== null &&
24 run.status !== "preparing" &&
25 run.status !== "queued" &&
26 run.status !== "starting" &&
27 run.status !== "running" &&
28 run.status !== "waiting" &&
29 thread.runtime?.activeRunId !== run.runId;
30 return !(thread.interactionMode === "plan" && thread.hasActionableProposedPlan && runSettled);
31}
32
33/**
34 * A thread its own client is launching reads as preparing until its first run lands.
35 * The server commits the thread before its run, so the list would otherwise place it twice.
36 */
37export function asLaunchingThread<T extends EnvironmentThreadShell>(thread: T): T {
38 if (thread.latestRun !== null || thread.runtime !== null) return thread;
39 return {
40 ...thread,
41 runtime: {
42 status: "preparing",
43 activeRunId: null,
44 providerInstanceId: thread.providerInstanceId,
45 providerName: null,
46 lastError: null,
47 updatedAt: thread.updatedAt,
48 },
49 };
50}
51
52type InboxThreadInput = Pick<
53 EnvironmentThreadShell,
54 "id" | "environmentId" | "createdAt" | "unsettledAt" | "latestRun"
55>;
56
57export function sortInboxThreadsByReturn<T extends InboxThreadInput>(
58 threads: readonly T[],
59 observedReturnAt?: (thread: T) => number | undefined,
60): T[] {
61 const timestamps = new Map(
62 threads.map((thread) => [
63 thread,
64 Math.max(
65 toSortableTimestamp(thread.createdAt) ?? 0,
66 toSortableTimestamp(thread.unsettledAt ?? undefined) ?? 0,
67 toSortableTimestamp(thread.latestRun?.requestedAt ?? undefined) ?? 0,
68 toSortableTimestamp(thread.latestRun?.completedAt ?? undefined) ?? 0,
69 observedReturnAt?.(thread) ?? 0,
70 ),
71 ]),
72 );
73 return sortNewestFirst(threads, timestamps);
74}
75
76type WorkingSortInput = Pick<
77 EnvironmentThreadShell,
78 "id" | "environmentId" | "createdAt" | "latestRun" | "latestUserAuthoredMessageAt"
79>;
80
81export function sortWorkingThreadsBySend<T extends WorkingSortInput>(threads: readonly T[]): T[] {
82 const timestamps = new Map(
83 threads.map((thread) => [
84 thread,
85 Math.max(
86 toSortableTimestamp(thread.createdAt) ?? 0,
87 toSortableTimestamp(
88 (thread.latestUserAuthoredMessageAt === undefined
89 ? thread.latestRun?.requestedAt
90 : thread.latestUserAuthoredMessageAt) ?? undefined,
91 ) ?? 0,
92 ),
93 ]),
94 );
95 return sortNewestFirst(threads, timestamps);
96}
97
98function sortNewestFirst<T extends Pick<EnvironmentThreadShell, "id" | "environmentId">>(
99 threads: readonly T[],
100 timestamps: ReadonlyMap<T, number>,
101): T[] {
102 return [...threads].sort(
103 (left, right) =>
104 timestamps.get(right)! - timestamps.get(left)! ||
105 left.id.localeCompare(right.id) ||
106 left.environmentId.localeCompare(right.environmentId),
107 );
108}
109
110type TrackedInboxThread = WorkingThreadInput &
111 InboxThreadInput &
112 Partial<
113 Pick<
114 EnvironmentThreadShell,
115 | "archivedAt"
116 | "deletedAt"
117 | "pinnedAt"
118 | "settledOverride"
119 | "settledAt"
120 | "snoozedAt"
121 | "snoozedUntil"
122 | "activeOrderKey"
123 | "pendingGuestMessageCount"
124 >
125 >;
126
127type ThreadKeyInput = Pick<EnvironmentThreadShell, "environmentId" | "id">;
128
129/**
130 * How long a thread that stopped working keeps reading as working. An orchestrator
131 * run can end moments before its continuation starts, so clients wait this long
132 * before treating the thread as done.
133 */
134const WORKING_EXIT_GRACE_MS = 2_000;
135
136/**
137 * A run that ended this long ago has no continuation about to start, so its thread gets no
138 * grace. A client sees this for work that ended while it was closed or disconnected; the
139 * margin is wide so a client clock that runs ahead of the server does not cancel the grace.
140 */
141const STALE_EXIT_MS = 60_000;
142
143/**
144 * Keep one at module scope so placement survives routes that unmount the list. Call
145 * `observe` when the shells or the snapshot version change, or with null while the beta
146 * is off. Its result stays fixed, so a list built from it can be cached.
147 */
148export function createInboxReturnTracker() {
149 const keyOf = (thread: ThreadKeyInput) => `${thread.environmentId}:${thread.id}`;
150 const states = new Map<
151 string,
152 { working: boolean; deadline: number | null; placement: string | null }
153 >();
154 const returns = new Map<string, number>();
155 const listeners = new Set<() => void>();
156 let timer: ReturnType<typeof setTimeout> | undefined;
157 let version = 0;
158 const now = () => {
159 // @effect-diagnostics-next-line globalDate:off - The grace deadline is wall-clock time shared by web and native clients.
160 return Date.now();
161 };
162 const schedule = () => {
163 clearTimeout(timer);
164 timer = undefined;
165 let next = Infinity;
166 for (const state of states.values()) {
167 if (state.deadline !== null) next = Math.min(next, state.deadline);
168 }
169 if (next === Infinity) return;
170 timer = setTimeout(
171 () => {
172 const at = now();
173 for (const [key, state] of states) {
174 if (state.deadline !== null && state.deadline <= at) {
175 state.working = false;
176 state.deadline = null;
177 returns.set(key, at);
178 }
179 }
180 schedule();
181 version++;
182 for (const listener of listeners) listener();
183 },
184 Math.max(0, next - now()),
185 );
186 };
187 return {
188 subscribe(listener: () => void) {
189 listeners.add(listener);
190 return () => {
191 listeners.delete(listener);
192 };
193 },
194 getSnapshot: () => version,
195 observe(threads: readonly TrackedInboxThread[] | null): {
196 readonly isWorking: (thread: WorkingThreadInput & ThreadKeyInput) => boolean;
197 readonly returnedAt: (thread: ThreadKeyInput) => number | undefined;
198 } {
199 if (threads === null) {
200 clearTimeout(timer);
201 timer = undefined;
202 states.clear();
203 returns.clear();
204 return { isWorking: isThreadWorking, returnedAt: () => undefined };
205 }
206 const present = new Set<string>();
207 const at = now();
208 // @effect-diagnostics-next-line globalDate:off - Snooze checks compare wall-clock ISO times.
209 const observedAt = new Date(at).toISOString();
210 for (const thread of threads) {
211 const key = keyOf(thread);
212 present.add(key);
213 const previous = states.get(key);
214 const actual = isThreadWorking(thread);
215 const placement =
216 actual || previous?.working
217 ? JSON.stringify([
218 thread.archivedAt,
219 thread.deletedAt,
220 thread.pinnedAt,
221 thread.settledOverride,
222 thread.settledAt,
223 thread.snoozedAt,
224 thread.snoozedUntil,
225 thread.activeOrderKey,
226 ])
227 : null;
228 const actionable =
229 thread.hasPendingApprovals ||
230 thread.hasPendingUserInput ||
231 thread.pendingSubagentRequest != null ||
232 (thread.pendingGuestMessageCount ?? 0) > 0 ||
233 (thread.interactionMode === "plan" && thread.hasActionableProposedPlan) ||
234 thread.runtime?.lastError != null ||
235 thread.latestRun?.status === "failed";
236 const moved = previous !== undefined && previous.placement !== placement;
237 const excluded =
238 thread.archivedAt != null ||
239 thread.deletedAt != null ||
240 thread.pinnedAt != null ||
241 thread.settledOverride === "settled" ||
242 effectiveSnoozed(thread, { now: observedAt });
243 // Only shelf placement is held: execution state and attention requests stay authoritative.
244 const endedAt = toSortableTimestamp(thread.latestRun?.completedAt ?? undefined);
245 const endedLongAgo = endedAt !== null && endedAt <= at - STALE_EXIT_MS;
246 const deadline =
247 !actual && previous?.working && !actionable && !moved && !excluded && !endedLongAgo
248 ? (previous.deadline ?? at + WORKING_EXIT_GRACE_MS)
249 : null;
250 const working = actual || (deadline !== null && deadline > at);
251 if (previous?.working && !working) returns.set(key, at);
252 states.set(key, { working, deadline: working ? deadline : null, placement });
253 }
254 for (const key of states.keys()) {
255 if (!present.has(key)) {
256 states.delete(key);
257 returns.delete(key);
258 }
259 }
260 schedule();
261 const observedWorking = new Map<string, boolean>();
262 states.forEach((state, key) => observedWorking.set(key, state.working));
263 const observedReturns = new Map(returns);
264 return {
265 isWorking: (thread) => observedWorking.get(keyOf(thread)) ?? isThreadWorking(thread),
266 returnedAt: (thread) => observedReturns.get(keyOf(thread)),
267 };
268 },
269 isWorking: (thread: WorkingThreadInput & ThreadKeyInput) =>
270 states.get(keyOf(thread))?.working ?? isThreadWorking(thread),
271 returnedAt: (thread: ThreadKeyInput) => returns.get(keyOf(thread)),
272 };
273}