1import type {
2 EnvironmentId,
3 OrchestrationV2ShellSnapshot,
4 ProjectId,
5 ScopedProjectRef,
6 ScopedThreadRef,
7 ThreadId,
8} from "@t3tools/contracts";
9import {
10 isListedThreadLineage,
11 isProviderNativeSubagentThread,
12 OrchestrationV2ThreadShell,
13} from "@t3tools/contracts";
14import * as Schema from "effect/Schema";
15import { Atom } from "effect/reactivity";
16
17import type { EnvironmentThreadShell, PendingSubagentRequest } from "./models.ts";
18import { presentThreadShell, threadRuntimeIsActive } from "./models.ts";
19import { hiddenThreadKeys, NO_HIDDEN_THREADS } from "./threadTagFilter.ts";
20import { type EnvironmentCatalogState, enabledEnvironmentIds } from "./connections.ts";
21import {
22 arrayElementsEqual,
23 parseProjectRefCollectionKey,
24 parseThreadKey,
25 projectRefCollectionKey,
26 threadKey,
27 threadRefsEqual,
28} from "./entities.ts";
29
30const EMPTY_THREADS: ReadonlyArray<OrchestrationV2ThreadShell> = Object.freeze([]);
31const EMPTY_SCOPED_THREAD_REFS: ReadonlyArray<ScopedThreadRef> = Object.freeze([]);
32const EMPTY_THREAD_INDEX: ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> = new Map();
33const EMPTY_SUBAGENT_REQUESTS: ReadonlyMap<ThreadId, PendingSubagentRequest> = new Map();
34const EMPTY_SIDE_CHATS_BY_PARENT: ReadonlyMap<
35 ThreadId,
36 ReadonlyArray<OrchestrationV2ThreadShell>
37> = new Map();
38const EMPTY_THREAD_SHELLS: ReadonlyArray<EnvironmentThreadShell> = Object.freeze([]);
39const EMPTY_THREAD_REFS_BY_PROJECT: ReadonlyMap<
40 ProjectId,
41 ReadonlyArray<ScopedThreadRef>
42> = new Map();
43
44// An async question outlives the subagent's run, but once that run ends the
45// parent is no longer waiting on the answer, so it stops counting as input.
46// The activity run stays set when a follow-up is queued behind a blocked run;
47// older servers omit it, so fall back to the latest run's status.
48const isChildRunLive = (
49 thread: Pick<OrchestrationV2ThreadShell, "activityRunStatus" | "status">,
50) =>
51 thread.activityRunStatus !== undefined
52 ? thread.activityRunStatus !== null
53 : thread.status === "preparing" ||
54 thread.status === "starting" ||
55 thread.status === "running" ||
56 thread.status === "waiting";
57
58/**
59 * Whether a thread shows under its parent rather than in thread lists: a
60 * subagent, a side chat, or a task a linked environment delegated here whose parent the
61 * client lists (`delegatedUnderParent`, by thread key). A task whose parent
62 * the client cannot list stays in the lists, so its work is never out of reach.
63 */
64export function isShownUnderParent(
65 ref: ScopedThreadRef,
66 lineage: OrchestrationV2ThreadShell["lineage"],
67 delegatedUnderParent: ReadonlySet<string>,
68): boolean {
69 return !isListedThreadLineage(lineage) || delegatedUnderParent.has(threadKey(ref));
70}
71
72/** Provider-native
73 subagents ask on their parent, so they are skipped. */
74function subagentRequest(thread: OrchestrationV2ThreadShell): PendingSubagentRequest | null {
75 const kind = thread.pendingRuntimeRequest?.kind;
76 if (
77 kind === undefined ||
78 kind === "auth_refresh" ||
79 (thread.lineage.relationshipToParent !== "subagent" && thread.delegatedFrom === undefined) ||
80 isProviderNativeSubagentThread(thread) ||
81 !isChildRunLive(thread)
82 ) {
83 return null;
84 }
85 return kind === "user_input" ? "user_input" : "approval";
86}
87
88/** A side chat asks its user directly, so questions roll up to the parent as well as approvals. */
89function sideChatRequest(thread: OrchestrationV2ThreadShell): PendingSubagentRequest | null {
90 const kind = thread.pendingRuntimeRequest?.kind;
91 if (kind === undefined || kind === "auth_refresh" || thread.lineage.sideChat !== true) {
92 return null;
93 }
94 return kind === "user_input" ? "user_input" : "approval";
95}
96
97/**
98 * The threads to flag for a subagent's live approval, up its
99 * subagent chain and through a side chat to its parent. `delegated` are tasks this
100 * environment's threads delegated to linked environments, with what each one, or its
101 * own subagents, asks.
102 */
103function pendingSubagentRequestsByThread(
104 threads: ReadonlyArray<OrchestrationV2ThreadShell>,
105 delegated: ReadonlyArray<{
106 readonly parentThreadId: ThreadId;
107 readonly request: PendingSubagentRequest;
108 }> = [],
109): ReadonlyMap<ThreadId, PendingSubagentRequest> {
110 const requests = new Map<ThreadId, PendingSubagentRequest>();
111 let byId: ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> | undefined;
112 const flag = (
113 firstParentId: ThreadId | null,
114 request: PendingSubagentRequest,
115 seen: Set<ThreadId>,
116 ) => {
117 // The coordinator answers child questions; only approvals need user attention.
118 if (request === "user_input") return;
119 byId ??= new Map(threads.map((thread) => [thread.id, thread] as const));
120 let parentId = firstParentId;
121 while (parentId !== null && !seen.has(parentId)) {
122 seen.add(parentId);
123 if (requests.get(parentId) !== "approval") requests.set(parentId, request);
124 const parent = byId.get(parentId);
125 parentId =
126 parent?.lineage.relationshipToParent === "subagent" || parent?.lineage.sideChat === true
127 ? parent.lineage.parentThreadId
128 : null;
129 }
130 };
131 for (const child of threads) {
132 const parentId = child.lineage.parentThreadId;
133 if (child.lineage.sideChat === true) {
134 const request = sideChatRequest(child);
135 if (parentId !== null && request !== null && requests.get(parentId) !== "approval") {
136 requests.set(parentId, request);
137 }
138 continue;
139 }
140 if (child.lineage.relationshipToParent !== "subagent") continue;
141 const request = subagentRequest(child);
142 if (request !== null) flag(parentId, request, new Set([child.id]));
143 }
144 for (const task of delegated) flag(task.parentThreadId, task.request, new Set());
145 return requests.size === 0 ? EMPTY_SUBAGENT_REQUESTS : requests;
146}
147
148const sameRequests = (
149 left: ReadonlyMap<ThreadId, PendingSubagentRequest>,
150 right: ReadonlyMap<ThreadId, PendingSubagentRequest>,
151) =>
152 left.size === right.size &&
153 [...right].every(([threadId, request]) => left.get(threadId) === request);
154
155export function childThreadInputsTitle(
156 children: ReadonlyArray<Pick<EnvironmentThreadShell, "hasPendingApprovals" | "lineage">>,
157): string {
158 const approvals = children.filter((child) => child.hasPendingApprovals).length;
159 const need = approvals === 0 ? "input" : approvals === children.length ? "approval" : "you";
160 const sideChats = children.filter((child) => child.lineage.sideChat === true).length;
161 const noun =
162 sideChats === children.length ? "side chat" : sideChats === 0 ? "subagent" : "thread";
163 return children.length === 1
164 ? `${noun === "side chat" ? "Side chat" : "Subagent"} needs ${need}`
165 : `${children.length} ${noun}s need ${need}`;
166}
167
168const sameWireThreadShell = Schema.toEquivalence(OrchestrationV2ThreadShell);
169
170// While a run streams, the server bumps the shell's timestamps, item counts and
171// latest message several times a second. Only the zero crossing of the counts
172// says anything (whether the thread has started). Once the run stops every
173// field counts again, so snooze wakes and failure times read the current value.
174function sameApartFromStreaming(
175 previous: EnvironmentThreadShell,
176 next: EnvironmentThreadShell,
177): boolean {
178 if (
179 previous.environmentId !== next.environmentId ||
180 previous.pendingSubagentRequest !== next.pendingSubagentRequest ||
181 !threadRuntimeIsActive(next.runtime)
182 ) {
183 return false;
184 }
185 const before = previous.source;
186 const after = next.source;
187 return (
188 before.itemCount > 0 === after.itemCount > 0 &&
189 before.visibleItemCount > 0 === after.visibleItemCount > 0 &&
190 sameWireThreadShell(
191 {
192 ...before,
193 updatedAt: after.updatedAt,
194 itemCount: after.itemCount,
195 visibleItemCount: after.visibleItemCount,
196 latestVisibleMessage: after.latestVisibleMessage,
197 },
198 after,
199 )
200 );
201}
202
203export function createEnvironmentThreadShellAtoms(input: {
204 readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;
205 readonly snapshotAtom: (
206 environmentId: EnvironmentId,
207 ) => Atom.Atom<OrchestrationV2ShellSnapshot | null>;
208 /** Tags whose threads display lists leave out; null while the filter is off. */
209 readonly hiddenThreadTagsAtom: (
210 environmentId: EnvironmentId,
211 ) => Atom.Atom<ReadonlyArray<string> | null>;
212}) {
213 // Point reads and aggregate lists share values without keeping an atom alive
214 // for every listed thread. Replaced source objects can be collected.
215 const scopedThreads = new WeakMap<
216 OrchestrationV2ThreadShell,
217 Map<string, EnvironmentThreadShell>
218 >();
219 const scopedThread = (
220 environmentId: EnvironmentId,
221 thread: OrchestrationV2ThreadShell,
222 subagentRequests: ReadonlyMap<ThreadId, PendingSubagentRequest>,
223 ) => {
224 let byEnvironment = scopedThreads.get(thread);
225 if (byEnvironment === undefined) {
226 byEnvironment = new Map();
227 scopedThreads.set(thread, byEnvironment);
228 }
229 const request = subagentRequests.get(thread.id) ?? null;
230 const key = request === null ? environmentId : `${environmentId}:${request}`;
231 let value = byEnvironment.get(key);
232 if (value === undefined) {
233 value = presentThreadShell(environmentId, thread, request);
234 byEnvironment.set(key, value);
235 }
236 return value;
237 };
238
239 const environmentThreadsAtom = Atom.family((environmentId: EnvironmentId) =>
240 Atom.make(
241 (get): ReadonlyArray<OrchestrationV2ThreadShell> =>
242 get(input.snapshotAtom(environmentId))?.threads ?? EMPTY_THREADS,
243 ).pipe(Atom.withLabel(`environment-threads:${environmentId}`)),
244 );
245
246 const environmentThreadIndexAtom = Atom.family((environmentId: EnvironmentId) =>
247 Atom.make((get): ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> => {
248 const threads = get(environmentThreadsAtom(environmentId));
249 if (threads.length === 0) {
250 return EMPTY_THREAD_INDEX;
251 }
252 return new Map(threads.map((thread) => [thread.id, thread] as const));
253 }).pipe(Atom.withLabel(`environment-thread-index:${environmentId}`)),
254 );
255
256 // Side chats by parent, built once per change for every parent's row. Without side chats
257 // it stays the same empty map, so those rows do no work.
258 const environmentSideChatsByParentAtom = Atom.family((environmentId: EnvironmentId) =>
259 Atom.make((get) => {
260 let byParent: Map<ThreadId, OrchestrationV2ThreadShell[]> | undefined;
261 for (const thread of get(environmentThreadsAtom(environmentId))) {
262 const parentId = thread.lineage.parentThreadId;
263 if (thread.lineage.sideChat !== true || parentId === null) continue;
264 byParent ??= new Map();
265 const siblings = byParent.get(parentId);
266 if (siblings) siblings.push(thread);
267 else byParent.set(parentId, [thread]);
268 }
269 return byParent ?? EMPTY_SIDE_CHATS_BY_PARENT;
270 }).pipe(Atom.withLabel(`environment-side-chats-by-parent:${environmentId}`)),
271 );
272
273 // What an environment's own subagents ask. Kept apart from the tasks other
274 // environments run for it, so two environments delegating to each other
275 // never read each other's totals.
276 const localSubagentRequestsAtom = Atom.family((environmentId: EnvironmentId) => {
277 let previous = EMPTY_SUBAGENT_REQUESTS;
278 return Atom.make((get) => {
279 const next = pendingSubagentRequestsByThread(get(environmentThreadsAtom(environmentId)));
280 if (sameRequests(previous, next)) return previous;
281 previous = next;
282 return next;
283 }).pipe(Atom.withLabel(`environment-local-subagent-requests:${environmentId}`));
284 });
285
286 // Flags for environments whose threads delegated tasks, counting what each task
287 // or anything it delegated on asks. Each pass settles one more hop of a chain;
288 // flags only grow, so environments delegating to each other settle too.
289 const delegatedSubagentRequestsAtom = Atom.make((get) => {
290 const tasksByDelegator = new Map<
291 EnvironmentId,
292 Array<{
293 readonly runsIn: EnvironmentId;
294 readonly thread: OrchestrationV2ThreadShell;
295 readonly parentThreadId: ThreadId;
296 }>
297 >();
298 for (const runsIn of enabledEnvironmentIds(get(input.catalogValueAtom))) {
299 for (const thread of get(environmentThreadsAtom(runsIn))) {
300 const from = thread.delegatedFrom;
301 if (from === undefined || from.environmentId === runsIn) continue;
302 const task = { runsIn, thread, parentThreadId: from.threadId };
303 const tasks = tasksByDelegator.get(from.environmentId);
304 if (tasks === undefined) tasksByDelegator.set(from.environmentId, [task]);
305 else tasks.push(task);
306 }
307 }
308 let flags = new Map<EnvironmentId, ReadonlyMap<ThreadId, PendingSubagentRequest>>();
309 const requestsOf = (environmentId: EnvironmentId) =>
310 flags.get(environmentId) ?? get(localSubagentRequestsAtom(environmentId));
311 for (let changed = tasksByDelegator.size > 0; changed;) {
312 changed = false;
313 const next = new Map(flags);
314 for (const [environmentId, tasks] of tasksByDelegator) {
315 const delegated = tasks.flatMap(({ runsIn, thread, parentThreadId }) => {
316 const own = subagentRequest(thread);
317 const nested = requestsOf(runsIn).get(thread.id);
318 const request = own === "approval" ? own : (nested ?? own);
319 return request === null ? [] : [{ parentThreadId, request }];
320 });
321 if (delegated.length === 0) continue;
322 const requests = pendingSubagentRequestsByThread(
323 get(environmentThreadsAtom(environmentId)),
324 delegated,
325 );
326 if (sameRequests(requestsOf(environmentId), requests)) continue;
327 next.set(environmentId, requests);
328 changed = true;
329 }
330 flags = next;
331 }
332 return flags;
333 }).pipe(Atom.withLabel("environment-delegated-subagent-requests"));
334
335 const pendingSubagentRequestsAtom = Atom.family((environmentId: EnvironmentId) => {
336 let previous = EMPTY_SUBAGENT_REQUESTS;
337 return Atom.make((get) => {
338 const next =
339 get(delegatedSubagentRequestsAtom).get(environmentId) ??
340 get(localSubagentRequestsAtom(environmentId));
341 if (sameRequests(previous, next)) return previous;
342 previous = next;
343 return next;
344 }).pipe(Atom.withLabel(`environment-pending-subagent-requests:${environmentId}`));
345 });
346
347 const environmentThreadRefsAtom = Atom.family((environmentId: EnvironmentId) => {
348 let previous: ReadonlyArray<ScopedThreadRef> = [];
349 return Atom.make((get) => {
350 const next = get(environmentThreadsAtom(environmentId)).map((thread) => ({
351 environmentId,
352 threadId: thread.id,
353 }));
354 if (threadRefsEqual(previous, next)) {
355 return previous;
356 }
357 previous = next;
358 return next;
359 }).pipe(Atom.withLabel(`environment-thread-refs:${environmentId}`));
360 });
361
362 const environmentThreadRefsByProjectAtom = Atom.family((environmentId: EnvironmentId) => {
363 let previous: ReadonlyMap<
364 ProjectId,
365 ReadonlyArray<ScopedThreadRef>
366 > = EMPTY_THREAD_REFS_BY_PROJECT;
367 return Atom.make((get) => {
368 const grouped = new Map<ProjectId, ScopedThreadRef[]>();
369 for (const thread of get(environmentThreadsAtom(environmentId))) {
370 const refs = grouped.get(thread.projectId);
371 const ref = { environmentId, threadId: thread.id };
372 if (refs === undefined) {
373 grouped.set(thread.projectId, [ref]);
374 } else {
375 refs.push(ref);
376 }
377 }
378 if (grouped.size === 0) {
379 previous = EMPTY_THREAD_REFS_BY_PROJECT;
380 return previous;
381 }
382 const next = new Map<ProjectId, ReadonlyArray<ScopedThreadRef>>();
383 for (const [projectId, refs] of grouped) {
384 const previousRefs = previous.get(projectId);
385 next.set(
386 projectId,
387 previousRefs !== undefined && threadRefsEqual(previousRefs, refs) ? previousRefs : refs,
388 );
389 }
390 const previousProjectIds = [...previous.keys()];
391 if (
392 next.size === previous.size &&
393 [...next].every(
394 ([projectId, refs], index) =>
395 previousProjectIds[index] === projectId && previous.get(projectId) === refs,
396 )
397 ) {
398 return previous;
399 }
400 previous = next;
401 return previous;
402 }).pipe(Atom.withLabel(`environment-thread-refs-by-project:${environmentId}`));
403 });
404
405 const threadShellAtomFamily = Atom.family((key: string) => {
406 const ref = parseThreadKey(key);
407 return Atom.make((get) => {
408 const source = get(environmentThreadIndexAtom(ref.environmentId)).get(ref.threadId) ?? null;
409 return source === null
410 ? null
411 : scopedThread(
412 ref.environmentId,
413 source,
414 get(pendingSubagentRequestsAtom(ref.environmentId)),
415 );
416 }).pipe(Atom.withLabel(`environment-thread-shell:${key}`));
417 });
418
419 // Keeps the previous shell while only streaming ticks changed. During a run,
420 // its timestamps, item counts and latest message are as of the last kept
421 // change; anything that needs them current reads threadShellAtom.
422 const quietThreadShellAtomFamily = Atom.family((key: string) => {
423 let previous: EnvironmentThreadShell | null = null;
424 return Atom.make((get) => {
425 const next = get(threadShellAtomFamily(key));
426 if (previous !== null && next !== null && sameApartFromStreaming(previous, next)) {
427 return previous;
428 }
429 previous = next;
430 return next;
431 }).pipe(Atom.withLabel(`environment-quiet-thread-shell:${key}`));
432 });
433
434 // Child approvals and questions are hidden with their sidebar rows. Read their
435 // shell summaries, without subscribing to every child transcript. A task
436 // delegated to a linked environment is a child too, with its own subagents.
437 const childThreadInputsAtomFamily = Atom.family((key: string) => {
438 let previous: ReadonlyArray<EnvironmentThreadShell> = [];
439 return Atom.make((get) => {
440 const children = new Map<
441 string,
442 Array<{ environmentId: EnvironmentId; thread: OrchestrationV2ThreadShell }>
443 >();
444 const addChild = (
445 parent: ScopedThreadRef,
446 environmentId: EnvironmentId,
447 thread: OrchestrationV2ThreadShell,
448 ) => {
449 const siblings = children.get(threadKey(parent));
450 if (siblings) siblings.push({ environmentId, thread });
451 else children.set(threadKey(parent), [{ environmentId, thread }]);
452 };
453 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
454 for (const thread of get(environmentThreadsAtom(environmentId))) {
455 const parent = thread.lineage.parentThreadId;
456 if (
457 parent !== null &&
458 (thread.lineage.relationshipToParent === "subagent" || thread.lineage.sideChat === true)
459 ) {
460 addChild({ environmentId, threadId: parent }, environmentId, thread);
461 } else if (thread.delegatedFrom !== undefined) {
462 addChild(thread.delegatedFrom, environmentId, thread);
463 }
464 }
465 }
466 const seen = new Set<string>([key]);
467 const pending = [key];
468 const next: EnvironmentThreadShell[] = [];
469 for (const parent of pending) {
470 for (const { environmentId, thread } of children.get(parent) ?? []) {
471 const childKey = threadKey({ environmentId, threadId: thread.id });
472 if (seen.has(childKey)) continue;
473 seen.add(childKey);
474 pending.push(childKey);
475 if (subagentRequest(thread) !== null || sideChatRequest(thread) !== null) {
476 next.push(
477 scopedThread(environmentId, thread, get(pendingSubagentRequestsAtom(environmentId))),
478 );
479 }
480 }
481 }
482 // Indicators read only ids, titles and the request kind, so other shell
483 // updates keep the old array.
484 if (
485 previous.length === next.length &&
486 previous.every(
487 (thread, index) =>
488 thread.environmentId === next[index]?.environmentId &&
489 thread.id === next[index]?.id &&
490 thread.title === next[index]?.title &&
491 thread.hasPendingApprovals === next[index]?.hasPendingApprovals,
492 )
493 ) {
494 return previous;
495 }
496 previous = next;
497 return next;
498 }).pipe(Atom.withLabel(`environment-child-thread-inputs:${key}`));
499 });
500
501 // A parent's side chats. Rows read only ids, titles and whether one needs the user, so
502 // streaming updates keep the old array.
503 const sideChatsAtomFamily = Atom.family((key: string) => {
504 const ref = parseThreadKey(key);
505 let previous = EMPTY_THREAD_SHELLS;
506 return Atom.make((get) => {
507 const sideChats = get(environmentSideChatsByParentAtom(ref.environmentId)).get(ref.threadId);
508 if (sideChats === undefined) {
509 previous = EMPTY_THREAD_SHELLS;
510 return previous;
511 }
512 const subagentRequests = get(pendingSubagentRequestsAtom(ref.environmentId));
513 const next = sideChats.map((thread) =>
514 scopedThread(ref.environmentId, thread, subagentRequests),
515 );
516 if (
517 previous.length === next.length &&
518 previous.every(
519 (thread, index) =>
520 thread.id === next[index]?.id &&
521 thread.title === next[index]?.title &&
522 thread.hasPendingApprovals === next[index]?.hasPendingApprovals &&
523 thread.hasPendingUserInput === next[index]?.hasPendingUserInput &&
524 thread.pendingSubagentRequest === next[index]?.pendingSubagentRequest,
525 )
526 ) {
527 return previous;
528 }
529 previous = next;
530 return next;
531 }).pipe(Atom.withLabel(`environment-side-chats:${key}`));
532 });
533
534 const threadShellsForProjectRefsAtomFamily = Atom.family((key: string) => {
535 const projectRefs = parseProjectRefCollectionKey(key);
536 let previous: ReadonlyArray<EnvironmentThreadShell> = [];
537 return Atom.make((get) => {
538 const next: EnvironmentThreadShell[] = [];
539 const seen = new Set<string>();
540 for (const projectRef of projectRefs) {
541 const refs =
542 get(environmentThreadRefsByProjectAtom(projectRef.environmentId)).get(
543 projectRef.projectId,
544 ) ?? EMPTY_SCOPED_THREAD_REFS;
545 if (refs.length === 0) continue;
546 const threads = get(environmentThreadIndexAtom(projectRef.environmentId));
547 const subagentRequests = get(pendingSubagentRequestsAtom(projectRef.environmentId));
548 for (const ref of refs) {
549 const key = threadKey(ref);
550 if (seen.has(key)) {
551 continue;
552 }
553 seen.add(key);
554 const thread = threads.get(ref.threadId);
555 if (thread !== undefined) {
556 next.push(scopedThread(ref.environmentId, thread, subagentRequests));
557 }
558 }
559 }
560 if (arrayElementsEqual(previous, next)) {
561 return previous;
562 }
563 previous = next;
564 return previous;
565 }).pipe(Atom.withLabel(`environment-thread-shells-for-projects:${key}`));
566 });
567
568 let previousThreadRefs: ReadonlyArray<ScopedThreadRef> = [];
569 const threadRefsAtom = Atom.make((get) => {
570 const refs: ScopedThreadRef[] = [];
571 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
572 refs.push(...get(environmentThreadRefsAtom(environmentId)));
573 }
574 if (threadRefsEqual(previousThreadRefs, refs)) {
575 return previousThreadRefs;
576 }
577 previousThreadRefs = refs;
578 return refs;
579 }).pipe(Atom.withLabel("environment-thread-refs"));
580
581 let previousThreadShells: ReadonlyArray<EnvironmentThreadShell> = [];
582 const threadShellsAtom = Atom.make((get) => {
583 const next: EnvironmentThreadShell[] = [];
584 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
585 const subagentRequests = get(pendingSubagentRequestsAtom(environmentId));
586 for (const thread of get(environmentThreadsAtom(environmentId))) {
587 next.push(scopedThread(environmentId, thread, subagentRequests));
588 }
589 }
590 if (arrayElementsEqual(previousThreadShells, next)) {
591 return previousThreadShells;
592 }
593 previousThreadShells = next;
594 return previousThreadShells;
595 }).pipe(Atom.withLabel("environment-thread-shell-list"));
596
597 // Point reads and threadShellsAtom stay complete, so code that asks whether a thread
598 // exists (outboxes, drafts, removals) never mistakes a hidden thread for a missing one.
599 let previousHiddenThreadKeys: ReadonlySet<string> = NO_HIDDEN_THREADS;
600 const hiddenThreadKeysAtom = Atom.make((get) => {
601 const environments = Array.from(
602 enabledEnvironmentIds(get(input.catalogValueAtom)),
603 (environmentId) => ({
604 environmentId,
605 hiddenTags: get(input.hiddenThreadTagsAtom(environmentId)),
606 }),
607 );
608 // Read shells only while some environment hides tags, so streaming ticks cost nothing otherwise.
609 if (!environments.some((environment) => environment.hiddenTags !== null)) {
610 previousHiddenThreadKeys = NO_HIDDEN_THREADS;
611 return NO_HIDDEN_THREADS;
612 }
613 const next = hiddenThreadKeys(
614 environments.map((environment) => ({
615 ...environment,
616 threads: get(environmentThreadsAtom(environment.environmentId)),
617 })),
618 );
619 if (
620 next.size === previousHiddenThreadKeys.size &&
621 [...next].every((key) => previousHiddenThreadKeys.has(key))
622 ) {
623 return previousHiddenThreadKeys;
624 }
625 previousHiddenThreadKeys = next;
626 return next;
627 }).pipe(Atom.withLabel("environment-hidden-thread-keys"));
628
629 const anyEnvironmentHidingAtom = Atom.make((get) =>
630 Array.from(enabledEnvironmentIds(get(input.catalogValueAtom))).some(
631 (environmentId) => get(input.hiddenThreadTagsAtom(environmentId)) !== null,
632 ),
633 ).pipe(Atom.withLabel("environment-any-hiding-thread-tags"));
634
635 /** threadShellsAtom without the threads the tag filter hides, for pickers and palettes. */
636 let previousVisibleThreadShells: ReadonlyArray<EnvironmentThreadShell> = [];
637 const visibleThreadShellsAtom = Atom.make((get) => {
638 const threads = get(threadShellsAtom);
639 const hidden = get(hiddenThreadKeysAtom);
640 if (hidden.size === 0) return threads;
641 const next = threads.filter(
642 (thread) =>
643 !hidden.has(threadKey({ environmentId: thread.environmentId, threadId: thread.id })),
644 );
645 if (arrayElementsEqual(previousVisibleThreadShells, next)) return previousVisibleThreadShells;
646 previousVisibleThreadShells = next;
647 return next;
648 }).pipe(Atom.withLabel("environment-visible-thread-shell-list"));
649
650 // Lists that arrange threads read this, so a running thread's streaming
651 // ticks reach only what reads its live shell.
652 let previousQuietThreadShells: ReadonlyArray<EnvironmentThreadShell> = [];
653 const quietThreadShellsAtom = Atom.make((get) => {
654 const next: EnvironmentThreadShell[] = [];
655 const hidden = get(hiddenThreadKeysAtom);
656 for (const ref of get(threadRefsAtom)) {
657 if (hidden.has(threadKey(ref))) continue;
658 const shell = get(quietThreadShellAtomFamily(threadKey(ref)));
659 if (shell !== null) next.push(shell);
660 }
661 if (arrayElementsEqual(previousQuietThreadShells, next)) return previousQuietThreadShells;
662 previousQuietThreadShells = next;
663 return next;
664 }).pipe(Atom.withLabel("environment-quiet-thread-shell-list"));
665
666 let previousDelegatedUnderParent: ReadonlySet<string> = new Set();
667 const delegatedUnderParentAtom = Atom.make((get) => {
668 const environmentIds = new Set(enabledEnvironmentIds(get(input.catalogValueAtom)));
669 const next = new Set<string>();
670 for (const environmentId of environmentIds) {
671 for (const thread of get(environmentThreadsAtom(environmentId))) {
672 const parent = thread.delegatedFrom;
673 if (
674 parent !== undefined &&
675 environmentIds.has(parent.environmentId) &&
676 // An archived parent leaves navigation; its task there stays listed on its own.
677 get(environmentThreadIndexAtom(parent.environmentId)).get(parent.threadId)?.archivedAt ===
678 null
679 ) {
680 next.add(threadKey({ environmentId, threadId: thread.id }));
681 }
682 }
683 }
684 if (
685 next.size === previousDelegatedUnderParent.size &&
686 [...next].every((key) => previousDelegatedUnderParent.has(key))
687 ) {
688 return previousDelegatedUnderParent;
689 }
690 previousDelegatedUnderParent = next;
691 return next;
692 }).pipe(Atom.withLabel("environment-delegated-under-parent"));
693
694 let previousNavigationShells: ReadonlyArray<EnvironmentThreadShell> = [];
695 const navigationThreadShellsAtom = Atom.make((get) => {
696 const next: EnvironmentThreadShell[] = [];
697 const underParent = get(delegatedUnderParentAtom);
698 const hidden = get(hiddenThreadKeysAtom);
699 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
700 const subagentRequests = get(pendingSubagentRequestsAtom(environmentId));
701 for (const thread of get(environmentThreadsAtom(environmentId))) {
702 if (
703 thread.archivedAt !== null ||
704 isShownUnderParent({ environmentId, threadId: thread.id }, thread.lineage, underParent) ||
705 hidden.has(threadKey({ environmentId, threadId: thread.id }))
706 ) {
707 continue;
708 }
709 next.push(scopedThread(environmentId, thread, subagentRequests));
710 }
711 }
712 if (arrayElementsEqual(previousNavigationShells, next)) return previousNavigationShells;
713 previousNavigationShells = next;
714 return next;
715 }).pipe(Atom.withLabel("environment-navigation-thread-shells"));
716
717 return {
718 environmentThreadsAtom,
719 environmentThreadIndexAtom,
720 environmentThreadRefsAtom,
721 environmentThreadRefsByProjectAtom,
722 threadRefsAtom,
723 threadShellsAtom,
724 visibleThreadShellsAtom,
725 hiddenThreadKeysAtom,
726 anyEnvironmentHidingAtom,
727 quietThreadShellsAtom,
728 navigationThreadShellsAtom,
729 delegatedUnderParentAtom,
730 pendingSubagentRequestsAtom,
731 threadShellsForProjectRefsAtom: (refs: ReadonlyArray<ScopedProjectRef>) =>
732 threadShellsForProjectRefsAtomFamily(projectRefCollectionKey(refs)),
733 threadShellAtom: (ref: ScopedThreadRef) => threadShellAtomFamily(threadKey(ref)),
734 quietThreadShellAtom: (ref: ScopedThreadRef) => quietThreadShellAtomFamily(threadKey(ref)),
735 childThreadInputsAtom: (ref: ScopedThreadRef) => childThreadInputsAtomFamily(threadKey(ref)),
736 sideChatsAtom: (ref: ScopedThreadRef) => sideChatsAtomFamily(threadKey(ref)),
737 };
738}