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