1import type {
2 EnvironmentId,
3 OrchestrationV2ShellSnapshot,
4 OrchestrationV2ThreadShell,
5 ProjectId,
6 ScopedProjectRef,
7 ScopedThreadRef,
8 ThreadId,
9} from "@t3tools/contracts";
10import { isProviderNativeSubagentThread } from "@t3tools/contracts";
11import { Atom } from "effect/reactivity";
12
13import type { EnvironmentThreadShell, PendingSubagentRequest } from "./models.ts";
14import { presentThreadShell } from "./models.ts";
15import { type EnvironmentCatalogState, enabledEnvironmentIds } from "./connections.ts";
16import {
17 arrayElementsEqual,
18 parseProjectRefCollectionKey,
19 parseThreadKey,
20 projectRefCollectionKey,
21 threadKey,
22 threadRefsEqual,
23} from "./entities.ts";
24
25const EMPTY_THREADS: ReadonlyArray<OrchestrationV2ThreadShell> = Object.freeze([]);
26const EMPTY_SCOPED_THREAD_REFS: ReadonlyArray<ScopedThreadRef> = Object.freeze([]);
27const EMPTY_THREAD_INDEX: ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> = new Map();
28const EMPTY_SUBAGENT_REQUESTS: ReadonlyMap<ThreadId, PendingSubagentRequest> = new Map();
29const EMPTY_THREAD_REFS_BY_PROJECT: ReadonlyMap<
30 ProjectId,
31 ReadonlyArray<ScopedThreadRef>
32> = new Map();
33
34// An async question outlives the subagent's run, but once that run ends the
35// parent is no longer waiting on the answer, so it stops counting as input.
36// The activity run stays set when a follow-up is queued behind a blocked run;
37// older servers omit it, so fall back to the latest run's status.
38const isChildRunLive = (
39 thread: Pick<OrchestrationV2ThreadShell, "activityRunStatus" | "status">,
40) =>
41 thread.activityRunStatus !== undefined
42 ? thread.activityRunStatus !== null
43 : thread.status === "preparing" ||
44 thread.status === "starting" ||
45 thread.status === "running" ||
46 thread.status === "waiting";
47
48/** What a subagent's live run waits on the user for. Provider-native
49 subagents ask on their parent, so they are skipped. */
50function subagentRequest(thread: OrchestrationV2ThreadShell): PendingSubagentRequest | null {
51 const kind = thread.pendingRuntimeRequest?.kind;
52 if (
53 kind === undefined ||
54 kind === "auth_refresh" ||
55 thread.lineage.relationshipToParent !== "subagent" ||
56 isProviderNativeSubagentThread(thread) ||
57 !isChildRunLive(thread)
58 ) {
59 return null;
60 }
61 return kind === "user_input" ? "user_input" : "approval";
62}
63
64/**
65 * Threads above a subagent that waits on the user, by what it waits on. Every
66 * thread up to the visible one carries it, because subagent rows are hidden.
67 * An approval outranks a question.
68 */
69export function pendingSubagentRequestsByThread(
70 threads: ReadonlyArray<OrchestrationV2ThreadShell>,
71): ReadonlyMap<ThreadId, PendingSubagentRequest> {
72 const requests = new Map<ThreadId, PendingSubagentRequest>();
73 let byId: ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> | undefined;
74 for (const child of threads) {
75 const request = subagentRequest(child);
76 if (request === null) continue;
77 byId ??= new Map(threads.map((thread) => [thread.id, thread] as const));
78 const seen = new Set<ThreadId>([child.id]);
79 let current: OrchestrationV2ThreadShell | undefined = child;
80 while (current?.lineage.relationshipToParent === "subagent") {
81 const parentId: ThreadId | null = current.lineage.parentThreadId;
82 if (parentId === null || seen.has(parentId)) break;
83 seen.add(parentId);
84 if (requests.get(parentId) !== "approval") requests.set(parentId, request);
85 current = byId.get(parentId);
86 }
87 }
88 return requests.size === 0 ? EMPTY_SUBAGENT_REQUESTS : requests;
89}
90
91/** Banner title for the subagents `childThreadInputsAtom` lists. */
92export function childThreadInputsTitle(
93 children: ReadonlyArray<Pick<EnvironmentThreadShell, "hasPendingApprovals">>,
94): string {
95 const approvals = children.filter((child) => child.hasPendingApprovals).length;
96 const need = approvals === 0 ? "input" : approvals === children.length ? "approval" : "you";
97 return children.length === 1
98 ? `Subagent needs ${need}`
99 : `${children.length} subagents need ${need}`;
100}
101
102export function createEnvironmentThreadShellAtoms(input: {
103 readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;
104 readonly snapshotAtom: (
105 environmentId: EnvironmentId,
106 ) => Atom.Atom<OrchestrationV2ShellSnapshot | null>;
107}) {
108 // Point reads and aggregate lists share values without keeping an atom alive
109 // for every listed thread. Replaced source objects can be collected.
110 const scopedThreads = new WeakMap<
111 OrchestrationV2ThreadShell,
112 Map<string, EnvironmentThreadShell>
113 >();
114 const scopedThread = (
115 environmentId: EnvironmentId,
116 thread: OrchestrationV2ThreadShell,
117 subagentRequests: ReadonlyMap<ThreadId, PendingSubagentRequest>,
118 ) => {
119 let byEnvironment = scopedThreads.get(thread);
120 if (byEnvironment === undefined) {
121 byEnvironment = new Map();
122 scopedThreads.set(thread, byEnvironment);
123 }
124 const request = subagentRequests.get(thread.id) ?? null;
125 const key = request === null ? environmentId : `${environmentId}:${request}`;
126 let value = byEnvironment.get(key);
127 if (value === undefined) {
128 value = presentThreadShell(environmentId, thread, request);
129 byEnvironment.set(key, value);
130 }
131 return value;
132 };
133
134 const environmentThreadsAtom = Atom.family((environmentId: EnvironmentId) =>
135 Atom.make(
136 (get): ReadonlyArray<OrchestrationV2ThreadShell> =>
137 get(input.snapshotAtom(environmentId))?.threads ?? EMPTY_THREADS,
138 ).pipe(Atom.withLabel(`environment-threads:${environmentId}`)),
139 );
140
141 const environmentThreadIndexAtom = Atom.family((environmentId: EnvironmentId) =>
142 Atom.make((get): ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> => {
143 const threads = get(environmentThreadsAtom(environmentId));
144 if (threads.length === 0) {
145 return EMPTY_THREAD_INDEX;
146 }
147 return new Map(threads.map((thread) => [thread.id, thread] as const));
148 }).pipe(Atom.withLabel(`environment-thread-index:${environmentId}`)),
149 );
150
151 const pendingSubagentRequestsAtom = Atom.family((environmentId: EnvironmentId) => {
152 let previous = EMPTY_SUBAGENT_REQUESTS;
153 return Atom.make((get) => {
154 const next = pendingSubagentRequestsByThread(get(environmentThreadsAtom(environmentId)));
155 if (
156 next.size === previous.size &&
157 [...next].every(([threadId, request]) => previous.get(threadId) === request)
158 ) {
159 return previous;
160 }
161 previous = next;
162 return next;
163 }).pipe(Atom.withLabel(`environment-pending-subagent-requests:${environmentId}`));
164 });
165
166 const environmentThreadRefsAtom = Atom.family((environmentId: EnvironmentId) => {
167 let previous: ReadonlyArray<ScopedThreadRef> = [];
168 return Atom.make((get) => {
169 const next = get(environmentThreadsAtom(environmentId)).map((thread) => ({
170 environmentId,
171 threadId: thread.id,
172 }));
173 if (threadRefsEqual(previous, next)) {
174 return previous;
175 }
176 previous = next;
177 return next;
178 }).pipe(Atom.withLabel(`environment-thread-refs:${environmentId}`));
179 });
180
181 const environmentThreadRefsByProjectAtom = Atom.family((environmentId: EnvironmentId) => {
182 let previous: ReadonlyMap<
183 ProjectId,
184 ReadonlyArray<ScopedThreadRef>
185 > = EMPTY_THREAD_REFS_BY_PROJECT;
186 return Atom.make((get) => {
187 const grouped = new Map<ProjectId, ScopedThreadRef[]>();
188 for (const thread of get(environmentThreadsAtom(environmentId))) {
189 const refs = grouped.get(thread.projectId);
190 const ref = { environmentId, threadId: thread.id };
191 if (refs === undefined) {
192 grouped.set(thread.projectId, [ref]);
193 } else {
194 refs.push(ref);
195 }
196 }
197 if (grouped.size === 0) {
198 previous = EMPTY_THREAD_REFS_BY_PROJECT;
199 return previous;
200 }
201 const next = new Map<ProjectId, ReadonlyArray<ScopedThreadRef>>();
202 for (const [projectId, refs] of grouped) {
203 const previousRefs = previous.get(projectId);
204 next.set(
205 projectId,
206 previousRefs !== undefined && threadRefsEqual(previousRefs, refs) ? previousRefs : refs,
207 );
208 }
209 const previousProjectIds = [...previous.keys()];
210 if (
211 next.size === previous.size &&
212 [...next].every(
213 ([projectId, refs], index) =>
214 previousProjectIds[index] === projectId && previous.get(projectId) === refs,
215 )
216 ) {
217 return previous;
218 }
219 previous = next;
220 return previous;
221 }).pipe(Atom.withLabel(`environment-thread-refs-by-project:${environmentId}`));
222 });
223
224 const threadShellAtomFamily = Atom.family((key: string) => {
225 const ref = parseThreadKey(key);
226 return Atom.make((get) => {
227 const source = get(environmentThreadIndexAtom(ref.environmentId)).get(ref.threadId) ?? null;
228 return source === null
229 ? null
230 : scopedThread(
231 ref.environmentId,
232 source,
233 get(pendingSubagentRequestsAtom(ref.environmentId)),
234 );
235 }).pipe(Atom.withLabel(`environment-thread-shell:${key}`));
236 });
237
238 // Child approvals and questions are hidden with their sidebar rows. Read their
239 // shell summaries, without subscribing to every child transcript.
240 const childThreadInputsAtomFamily = Atom.family((key: string) => {
241 const ref = parseThreadKey(key);
242 let previous: ReadonlyArray<EnvironmentThreadShell> = [];
243 return Atom.make((get) => {
244 const children = new Map<ThreadId, OrchestrationV2ThreadShell[]>();
245 for (const thread of get(environmentThreadsAtom(ref.environmentId))) {
246 const parent = thread.lineage.parentThreadId;
247 if (parent === null || thread.lineage.relationshipToParent !== "subagent") continue;
248 const siblings = children.get(parent);
249 if (siblings) siblings.push(thread);
250 else children.set(parent, [thread]);
251 }
252 const subagentRequests = get(pendingSubagentRequestsAtom(ref.environmentId));
253 const seen = new Set<ThreadId>([ref.threadId]);
254 const pending = [ref.threadId];
255 const next: EnvironmentThreadShell[] = [];
256 for (const parent of pending) {
257 for (const child of children.get(parent) ?? []) {
258 if (seen.has(child.id)) continue;
259 seen.add(child.id);
260 pending.push(child.id);
261 if (subagentRequest(child) !== null) {
262 next.push(scopedThread(ref.environmentId, child, subagentRequests));
263 }
264 }
265 }
266 // Indicators read only ids, titles and the request kind, so other shell
267 // updates keep the old array.
268 if (
269 previous.length === next.length &&
270 previous.every(
271 (thread, index) =>
272 thread.id === next[index]?.id &&
273 thread.title === next[index]?.title &&
274 thread.hasPendingApprovals === next[index]?.hasPendingApprovals,
275 )
276 ) {
277 return previous;
278 }
279 previous = next;
280 return next;
281 }).pipe(Atom.withLabel(`environment-child-thread-inputs:${key}`));
282 });
283
284 const threadShellsForProjectRefsAtomFamily = Atom.family((key: string) => {
285 const projectRefs = parseProjectRefCollectionKey(key);
286 let previous: ReadonlyArray<EnvironmentThreadShell> = [];
287 return Atom.make((get) => {
288 const next: EnvironmentThreadShell[] = [];
289 const seen = new Set<string>();
290 for (const projectRef of projectRefs) {
291 const refs =
292 get(environmentThreadRefsByProjectAtom(projectRef.environmentId)).get(
293 projectRef.projectId,
294 ) ?? EMPTY_SCOPED_THREAD_REFS;
295 if (refs.length === 0) continue;
296 const threads = get(environmentThreadIndexAtom(projectRef.environmentId));
297 const subagentRequests = get(pendingSubagentRequestsAtom(projectRef.environmentId));
298 for (const ref of refs) {
299 const key = threadKey(ref);
300 if (seen.has(key)) {
301 continue;
302 }
303 seen.add(key);
304 const thread = threads.get(ref.threadId);
305 if (thread !== undefined) {
306 next.push(scopedThread(ref.environmentId, thread, subagentRequests));
307 }
308 }
309 }
310 if (arrayElementsEqual(previous, next)) {
311 return previous;
312 }
313 previous = next;
314 return previous;
315 }).pipe(Atom.withLabel(`environment-thread-shells-for-projects:${key}`));
316 });
317
318 let previousThreadRefs: ReadonlyArray<ScopedThreadRef> = [];
319 const threadRefsAtom = Atom.make((get) => {
320 const refs: ScopedThreadRef[] = [];
321 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
322 refs.push(...get(environmentThreadRefsAtom(environmentId)));
323 }
324 if (threadRefsEqual(previousThreadRefs, refs)) {
325 return previousThreadRefs;
326 }
327 previousThreadRefs = refs;
328 return refs;
329 }).pipe(Atom.withLabel("environment-thread-refs"));
330
331 let previousThreadShells: ReadonlyArray<EnvironmentThreadShell> = [];
332 const threadShellsAtom = Atom.make((get) => {
333 const next: EnvironmentThreadShell[] = [];
334 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
335 const subagentRequests = get(pendingSubagentRequestsAtom(environmentId));
336 for (const thread of get(environmentThreadsAtom(environmentId))) {
337 next.push(scopedThread(environmentId, thread, subagentRequests));
338 }
339 }
340 if (arrayElementsEqual(previousThreadShells, next)) {
341 return previousThreadShells;
342 }
343 previousThreadShells = next;
344 return previousThreadShells;
345 }).pipe(Atom.withLabel("environment-thread-shell-list"));
346
347 let previousNavigationShells: ReadonlyArray<EnvironmentThreadShell> = [];
348 const navigationThreadShellsAtom = Atom.make((get) => {
349 const next: EnvironmentThreadShell[] = [];
350 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
351 const subagentRequests = get(pendingSubagentRequestsAtom(environmentId));
352 for (const thread of get(environmentThreadsAtom(environmentId))) {
353 if (thread.archivedAt !== null || thread.lineage.relationshipToParent === "subagent")
354 continue;
355 next.push(scopedThread(environmentId, thread, subagentRequests));
356 }
357 }
358 if (arrayElementsEqual(previousNavigationShells, next)) return previousNavigationShells;
359 previousNavigationShells = next;
360 return next;
361 }).pipe(Atom.withLabel("environment-navigation-thread-shells"));
362
363 return {
364 environmentThreadsAtom,
365 environmentThreadIndexAtom,
366 environmentThreadRefsAtom,
367 environmentThreadRefsByProjectAtom,
368 threadRefsAtom,
369 threadShellsAtom,
370 navigationThreadShellsAtom,
371 threadShellsForProjectRefsAtom: (refs: ReadonlyArray<ScopedProjectRef>) =>
372 threadShellsForProjectRefsAtomFamily(projectRefCollectionKey(refs)),
373 threadShellAtom: (ref: ScopedThreadRef) => threadShellAtomFamily(threadKey(ref)),
374 childThreadInputsAtom: (ref: ScopedThreadRef) => childThreadInputsAtomFamily(threadKey(ref)),
375 };
376}