1import type * as DateTime from "effect/DateTime";
2import { useAtomValue } from "@effect/atom-react";
3import type {
4 EnvironmentProject,
5 EnvironmentThread,
6 EnvironmentThreadShell,
7 EnvironmentThreadStructure,
8 ThreadStructureVisibleTurnItem,
9} from "@t3tools/client-runtime/state/shell";
10import {
11 EMPTY_THREAD_HISTORY_META,
12 type EnvironmentThreadStatus,
13 type ThreadHistoryMeta,
14} from "@t3tools/client-runtime/state/threads";
15import {
16 arrayElementsEqual,
17 parseThreadKey,
18 threadKey,
19} from "@t3tools/client-runtime/state/entities";
20import type { ScopedProjectRef, ScopedThreadRef } from "@t3tools/contracts";
21import type {
22 EnvironmentId,
23 OrchestrationV2ProjectedTurnItem,
24 OrchestrationV2RemoteTaskChild,
25 OrchestrationV2ThreadProjection,
26 OrchestrationV2ThreadShell,
27 NodeId,
28 ThreadId,
29 TurnItemId,
30} from "@t3tools/contracts";
31import type { ThreadCheckpointSummary } from "@t3tools/client-runtime/state/thread-checkpoints";
32import { Atom } from "effect/reactivity";
33import { appAtomRegistry } from "../rpc/atomRegistry";
34import { environmentProjects } from "./projects";
35import { environmentPresentations } from "./presentation";
36import { environmentServerConfigsAtom } from "./server";
37import {
38 allEnvironmentProjectSnapshotsReadyAtom,
39 allEnvironmentShellsBootstrappedAtom,
40} from "./shell";
41import { environmentThreadDetails, environmentThreadShells } from "./threads";
42import { waitForAtomValue } from "./waitForAtomValue";
43
44const EMPTY_THREAD_REFS: ReadonlyArray<ScopedThreadRef> = Object.freeze([]);
45const EMPTY_VISIBLE_TURN_ITEMS: ReadonlyArray<OrchestrationV2ProjectedTurnItem> = Object.freeze([]);
46
47const EMPTY_PROJECT_ATOM = Atom.make<EnvironmentProject | null>(null).pipe(
48 Atom.withLabel("web-project:empty"),
49);
50const EMPTY_THREAD_REFS_ATOM = Atom.make(EMPTY_THREAD_REFS).pipe(
51 Atom.withLabel("web-thread-refs:empty"),
52);
53const EMPTY_THREAD_SHELLS_ATOM = Atom.make<ReadonlyArray<EnvironmentThreadShell>>(
54 Object.freeze([]),
55).pipe(Atom.withLabel("web-thread-shells:empty"));
56const EMPTY_THREAD_SHELL_ATOM = Atom.make<EnvironmentThreadShell | null>(null).pipe(
57 Atom.withLabel("web-thread-shell:empty"),
58);
59const EMPTY_THREAD_PROJECTION_ATOM = Atom.make<EnvironmentThread | null>(null).pipe(
60 Atom.withLabel("web-thread-projection:empty"),
61);
62const EMPTY_THREAD_STATUS_ATOM = Atom.make<EnvironmentThreadStatus>("empty").pipe(
63 Atom.withLabel("web-thread-status:empty"),
64);
65const EMPTY_VISIBLE_TURN_ITEMS_ATOM = Atom.make(EMPTY_VISIBLE_TURN_ITEMS).pipe(
66 Atom.withLabel("web-thread-visible-turn-items:empty"),
67);
68const EMPTY_CHECKPOINT_SUMMARIES_ATOM = Atom.make<ReadonlyArray<ThreadCheckpointSummary>>(
69 Object.freeze([]),
70).pipe(Atom.withLabel("web-thread-checkpoint-summaries:empty"));
71const EMPTY_ASSISTANT_MESSAGE_TEXT_ATOM = Atom.make<string | null>(null).pipe(
72 Atom.withLabel("web-assistant-message-text:empty"),
73);
74const EMPTY_LAST_VISITED_AT_ATOM = Atom.make<DateTime.Utc | null>(null).pipe(
75 Atom.withLabel("web-thread-last-visited:empty"),
76);
77const EMPTY_UPDATED_AT_ATOM = Atom.make<DateTime.Utc | null>(null).pipe(
78 Atom.withLabel("web-thread-updated-at:empty"),
79);
80const EMPTY_THREAD_HISTORY_ATOM = Atom.make<ThreadHistoryMeta>(EMPTY_THREAD_HISTORY_META).pipe(
81 Atom.withLabel("web-thread-history:empty"),
82);
83
84export function useThreadRefs(enabled = true): ReadonlyArray<ScopedThreadRef> {
85 return useAtomValue(enabled ? environmentThreadShells.threadRefsAtom : EMPTY_THREAD_REFS_ATOM);
86}
87
88export function useEnvironmentThreadRefs(
89 environmentId: EnvironmentId | null,
90): ReadonlyArray<ScopedThreadRef> {
91 return useAtomValue(
92 environmentId === null
93 ? EMPTY_THREAD_REFS_ATOM
94 : environmentThreadShells.environmentThreadRefsAtom(environmentId),
95 );
96}
97
98export function useProjects(): ReadonlyArray<EnvironmentProject> {
99 return useAtomValue(environmentProjects.projectsAtom);
100}
101
102/** Providers are left out; read them with `useEnvironmentProviders()`. */
103export function useServerConfigs() {
104 return useAtomValue(environmentPresentations.serverConfigsAtom);
105}
106
107export function useThreadShells(enabled = true): ReadonlyArray<EnvironmentThreadShell> {
108 return useAtomValue(
109 enabled ? environmentThreadShells.threadShellsAtom : EMPTY_THREAD_SHELLS_ATOM,
110 );
111}
112
113/** Whether any connected environment is hiding tagged threads right now. */
114export function useAnyEnvironmentHiding(): boolean {
115 return useAtomValue(environmentThreadShells.anyEnvironmentHidingAtom);
116}
117
118/** Keys of threads the tag filter hides from display, by `threadKey`. */
119export function useHiddenThreadKeys(): ReadonlySet<string> {
120 return useAtomValue(environmentThreadShells.hiddenThreadKeysAtom);
121}
122
123/** Thread shells minus those the tag filter hides, for lists that show threads to the user. */
124export function useVisibleThreadShells(enabled = true): ReadonlyArray<EnvironmentThreadShell> {
125 return useAtomValue(
126 enabled ? environmentThreadShells.visibleThreadShellsAtom : EMPTY_THREAD_SHELLS_ATOM,
127 );
128}
129
130export function useAllEnvironmentShellsBootstrapped(): boolean {
131 return useAtomValue(allEnvironmentShellsBootstrappedAtom);
132}
133
134export function useAllEnvironmentProjectSnapshotsReady(): boolean {
135 return useAtomValue(allEnvironmentProjectSnapshotsReadyAtom);
136}
137
138/** Tasks linked environments run for a listed thread, by thread key; lists show them under it. */
139export function useDelegatedUnderParent(): ReadonlySet<string> {
140 return useAtomValue(environmentThreadShells.delegatedUnderParentAtom);
141}
142
143export function useThreadShellsForProjectRefs(
144 refs: ReadonlyArray<ScopedProjectRef>,
145): ReadonlyArray<EnvironmentThreadShell> {
146 return useAtomValue(environmentThreadShells.threadShellsForProjectRefsAtom(refs));
147}
148
149export function useProject(ref: ScopedProjectRef | null): EnvironmentProject | null {
150 return useAtomValue(ref === null ? EMPTY_PROJECT_ATOM : environmentProjects.projectAtom(ref));
151}
152
153export function useThreadShell(ref: ScopedThreadRef | null): EnvironmentThreadShell | null {
154 return useAtomValue(
155 ref === null ? EMPTY_THREAD_SHELL_ATOM : environmentThreadShells.threadShellAtom(ref),
156 );
157}
158
159/** The shell without its streaming timestamps; see quietThreadShellAtom. */
160export function useQuietThreadShell(ref: ScopedThreadRef | null): EnvironmentThreadShell | null {
161 return useAtomValue(
162 ref === null ? EMPTY_THREAD_SHELL_ATOM : environmentThreadShells.quietThreadShellAtom(ref),
163 );
164}
165
166/** Every thread shell, held through streaming ticks; see quietThreadShellAtom. */
167export function useQuietThreadShells(): ReadonlyArray<EnvironmentThreadShell> {
168 return useAtomValue(environmentThreadShells.quietThreadShellsAtom);
169}
170
171export function useChildThreadInputs(ref: ScopedThreadRef | null) {
172 return useAtomValue(
173 ref === null ? EMPTY_THREAD_SHELLS_ATOM : environmentThreadShells.childThreadInputsAtom(ref),
174 );
175}
176
177export function useThreadProjection(ref: ScopedThreadRef | null): EnvironmentThread | null {
178 return useAtomValue(
179 ref === null ? EMPTY_THREAD_PROJECTION_ATOM : environmentThreadDetails.threadAtom(ref),
180 );
181}
182
183/** The thread's structure, held through streamed assistant and reasoning text; see ThreadStructure. */
184export function useQuietThreadProjection(
185 ref: ScopedThreadRef | null,
186): EnvironmentThreadStructure | null {
187 return useAtomValue(
188 ref === null ? EMPTY_THREAD_PROJECTION_ATOM : environmentThreadDetails.quietThreadAtom(ref),
189 );
190}
191
192/** The detail's live visit mark, which the thread structure leaves out. */
193export function useThreadLastVisitedAt(ref: ScopedThreadRef | null): DateTime.Utc | null {
194 return useAtomValue(
195 ref === null ? EMPTY_LAST_VISITED_AT_ATOM : environmentThreadDetails.lastVisitedAtAtom(ref),
196 );
197}
198
199/** The detail's live `updatedAt`, which the thread structure leaves out. */
200export function useThreadUpdatedAt(ref: ScopedThreadRef | null): DateTime.Utc | null {
201 return useAtomValue(
202 ref === null ? EMPTY_UPDATED_AT_ATOM : environmentThreadDetails.updatedAtAtom(ref),
203 );
204}
205
206export function useThreadStatus(ref: ScopedThreadRef | null): EnvironmentThreadStatus {
207 return useAtomValue(
208 ref === null ? EMPTY_THREAD_STATUS_ATOM : environmentThreadDetails.statusAtom(ref),
209 );
210}
211
212export function useThreadHistory(ref: ScopedThreadRef | null): ThreadHistoryMeta {
213 return useAtomValue(
214 ref === null ? EMPTY_THREAD_HISTORY_ATOM : environmentThreadDetails.historyAtom(ref),
215 );
216}
217
218export function resolveThreadDetailRef(
219 ref: ScopedThreadRef | null,
220 options: {
221 shellExists: boolean;
222 waitForShell: boolean;
223 },
224): ScopedThreadRef | null {
225 return ref !== null && (!options.waitForShell || options.shellExists) ? ref : null;
226}
227
228export function useThreadVisibleTurnItems(
229 ref: ScopedThreadRef | null,
230): ReadonlyArray<OrchestrationV2ProjectedTurnItem> {
231 return useAtomValue(
232 ref === null
233 ? EMPTY_VISIBLE_TURN_ITEMS_ATOM
234 : environmentThreadDetails.visibleTurnItemsAtom(ref),
235 );
236}
237
238/** Visible turn items held through streamed assistant text; see useAssistantMessageText. */
239export function useTimelineVisibleTurnItems(
240 ref: ScopedThreadRef | null,
241): ReadonlyArray<OrchestrationV2ProjectedTurnItem> {
242 return useAtomValue(
243 ref === null
244 ? EMPTY_VISIBLE_TURN_ITEMS_ATOM
245 : environmentThreadDetails.timelineVisibleTurnItemsAtom(ref),
246 );
247}
248
249/** One assistant message's live text, or null when the thread does not hold the item. */
250export function useAssistantMessageText(
251 ref: ScopedThreadRef | null,
252 itemId: string | undefined,
253): string | null {
254 return useAtomValue(
255 ref === null || itemId === undefined
256 ? EMPTY_ASSISTANT_MESSAGE_TEXT_ATOM
257 : environmentThreadDetails.assistantMessageTextAtom(ref, itemId),
258 );
259}
260
261/** Visible turn items without their streamed text; see ThreadStructure. */
262export function useQuietThreadVisibleTurnItems(
263 ref: ScopedThreadRef | null,
264): ReadonlyArray<ThreadStructureVisibleTurnItem> {
265 return useAtomValue(
266 ref === null
267 ? EMPTY_VISIBLE_TURN_ITEMS_ATOM
268 : environmentThreadDetails.quietVisibleTurnItemsAtom(ref),
269 );
270}
271
272export function useThreadCheckpointSummaries(
273 ref: ScopedThreadRef | null,
274): ReadonlyArray<ThreadCheckpointSummary> {
275 return useAtomValue(
276 ref === null
277 ? EMPTY_CHECKPOINT_SUMMARIES_ATOM
278 : environmentThreadDetails.checkpointSummariesAtom(ref),
279 );
280}
281
282export function readProject(ref: ScopedProjectRef): EnvironmentProject | null {
283 return appAtomRegistry.get(environmentProjects.projectAtom(ref));
284}
285
286export function subscribeProject(
287 ref: ScopedProjectRef,
288 listener: (project: EnvironmentProject | null) => void,
289): () => void {
290 return appAtomRegistry.subscribe(environmentProjects.projectAtom(ref), listener);
291}
292
293export function readProjects(): ReadonlyArray<EnvironmentProject> {
294 return appAtomRegistry.get(environmentProjects.projectsAtom);
295}
296
297export function waitForProject(
298 ref: ScopedProjectRef,
299 timeoutMs = 10_000,
300): Promise<EnvironmentProject> {
301 const current = readProject(ref);
302 if (current !== null) return Promise.resolve(current);
303
304 return new Promise((resolve, reject) => {
305 let unsubscribe: (() => void) | null = null;
306 const timeout = setTimeout(() => {
307 unsubscribe?.();
308 reject(new Error("The project did not appear in the desktop app."));
309 }, timeoutMs);
310 const finish = (project: EnvironmentProject | null) => {
311 if (project === null) return;
312 clearTimeout(timeout);
313 unsubscribe?.();
314 resolve(project);
315 };
316 unsubscribe = appAtomRegistry.subscribe(environmentProjects.projectAtom(ref), finish);
317 finish(readProject(ref));
318 });
319}
320
321/** Waits for removal from the live store; delete replies arrive before the shell stream. */
322export function waitForProjectRemoval(ref: ScopedProjectRef, timeoutMs = 5_000): Promise<boolean> {
323 return waitForAtomValue({
324 registry: appAtomRegistry,
325 atom: environmentProjects.projectAtom(ref),
326 predicate: (project) => project === null,
327 timeoutMs,
328 });
329}
330
331export function readThreadShell(ref: ScopedThreadRef): EnvironmentThreadShell | null {
332 return appAtomRegistry.get(environmentThreadShells.threadShellAtom(ref));
333}
334
335/** The shell list leaves out idle subagents, so an open one's lineage comes from its detail. */
336export function readThreadLineage(ref: ScopedThreadRef) {
337 return (
338 readThreadShell(ref)?.lineage ??
339 appAtomRegistry.get(environmentThreadDetails.quietThreadAtom(ref))?.projection.thread.lineage ??
340 null
341 );
342}
343
344export function waitForThreadShell(ref: ScopedThreadRef, timeoutMs = 5_000): Promise<boolean> {
345 return waitForAtomValue({
346 registry: appAtomRegistry,
347 atom: environmentThreadShells.threadShellAtom(ref),
348 predicate: (thread) => thread !== null,
349 timeoutMs,
350 });
351}
352
353export function useEnvironmentSupportsServerBrowser(environmentId: EnvironmentId | null): boolean {
354 const configs = useServerConfigs();
355 return (
356 environmentId !== null &&
357 configs.get(environmentId)?.environment.capabilities.serverBrowser === true
358 );
359}
360
361export function readEnvironmentSupportsServerBrowser(environmentId: EnvironmentId): boolean {
362 return (
363 appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities
364 .serverBrowser === true
365 );
366}
367
368export function readEnvironmentServerConfigLoaded(environmentId: EnvironmentId): boolean {
369 return appAtomRegistry.get(environmentServerConfigsAtom).has(environmentId);
370}
371
372export function readEnvironmentSupportsAutoSettleOptOut(environmentId: EnvironmentId): boolean {
373 return (
374 appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities
375 .threadAutoSettleOptOut === true
376 );
377}
378
379export function readEnvironmentThreadRefs(
380 environmentId: EnvironmentId,
381): ReadonlyArray<ScopedThreadRef> {
382 return appAtomRegistry.get(environmentThreadShells.environmentThreadRefsAtom(environmentId));
383}
384
385export function readThreadShells(): ReadonlyArray<EnvironmentThreadShell> {
386 return appAtomRegistry.get(environmentThreadShells.threadShellsAtom);
387}
388
389function shellParentThreadId(
390 thread: Pick<OrchestrationV2ThreadShell, "forkedFrom" | "lineage">,
391): ThreadId | null {
392 return thread.forkedFrom?.type === "run"
393 ? thread.forkedFrom.threadId
394 : thread.lineage.parentThreadId;
395}
396
397/** Retains unchanged reachable shells so unrelated streams do not re-render the panel. */
398export function selectThreadRelationshipShells(
399 threadId: ThreadId,
400 projection: Pick<
401 OrchestrationV2ThreadProjection,
402 "thread" | "subagents" | "contextTransfers"
403 > | null,
404 threads: ReadonlyArray<OrchestrationV2ThreadShell>,
405): ReadonlyArray<OrchestrationV2ThreadShell> {
406 const named = new Set<string>([threadId]);
407 if (projection !== null) {
408 const parentThreadId = shellParentThreadId(projection.thread);
409 if (parentThreadId !== null) named.add(parentThreadId);
410 for (const subagent of projection.subagents) {
411 if (subagent.childThreadId !== null) named.add(subagent.childThreadId);
412 for (const agent of subagent.workflow?.agents ?? []) {
413 if (agent.childThreadId !== undefined) named.add(agent.childThreadId);
414 }
415 }
416 for (const transfer of projection.contextTransfers) {
417 named.add(transfer.sourceThreadId);
418 named.add(transfer.targetThreadId);
419 }
420 }
421 const current = threads.find((thread) => thread.id === threadId);
422 const currentParentId = current === undefined ? null : shellParentThreadId(current);
423 if (currentParentId !== null) named.add(currentParentId);
424 return threads.filter((thread) => {
425 const parentId = shellParentThreadId(thread);
426 return (
427 named.has(thread.id) ||
428 parentId === threadId ||
429 (currentParentId !== null && parentId === currentParentId)
430 );
431 });
432}
433
434const relationshipShellsAtomFamily = Atom.family((key: string) => {
435 const ref = parseThreadKey(key);
436 let previous: ReadonlyArray<OrchestrationV2ThreadShell> = [];
437 return Atom.make((get) => {
438 const next = selectThreadRelationshipShells(
439 ref.threadId,
440 get(environmentThreadDetails.threadAtom(ref))?.projection ?? null,
441 get(environmentThreadShells.environmentThreadsAtom(ref.environmentId)),
442 );
443 if (arrayElementsEqual(previous, next)) return previous;
444 previous = next;
445 return next;
446 }).pipe(Atom.withLabel(`web-thread-relationship-shells:${key}`));
447});
448
449export function useThreadRelationshipShells(
450 ref: ScopedThreadRef,
451): ReadonlyArray<OrchestrationV2ThreadShell> {
452 return useAtomValue(relationshipShellsAtomFamily(threadKey(ref)));
453}
454
455interface SubagentChild {
456 readonly threadId: ThreadId;
457 /** Set when the task ran on a linked environment, which holds its thread. */
458 readonly environmentId?: EnvironmentId;
459 readonly title: string | null | undefined;
460}
461
462/**
463 * The subagents a response reported on, read once on demand so a timeline row does not subscribe
464 * to the thread for it. A result reaches the thread as a message from the subagent's thread or as
465 * a completion notice, at the start of a run or steered into one, so the inputs just before the
466 * response decide it. Turn items are read because older history loads nothing else.
467 */
468export function readResponseSubagents(
469 ref: ScopedThreadRef,
470 itemId: TurnItemId,
471): ReadonlyArray<Omit<SubagentChild, "title"> & { readonly title: string }> {
472 const projection = appAtomRegistry.get(environmentThreadDetails.quietThreadAtom(ref))?.projection;
473 if (!projection) return [];
474 const items = projection.visibleTurnItems;
475 const isInput = (index: number) => {
476 const type = items[index]?.item.type;
477 return type === "user_message" || type === "notification";
478 };
479 let index = items.findIndex((projected) => projected.sourceItemId === itemId) - 1;
480 while (index >= 0 && !isInput(index)) index -= 1;
481
482 const byTask = new Map<NodeId, SubagentChild>();
483 const local = new Map<ThreadId, SubagentChild>();
484 const remember = (
485 taskId: NodeId,
486 childThreadId: ThreadId | null,
487 remote: OrchestrationV2RemoteTaskChild | undefined,
488 title: string | null,
489 ) => {
490 if (childThreadId !== null) {
491 const child = { threadId: childThreadId, title };
492 byTask.set(taskId, child);
493 local.set(childThreadId, child);
494 } else if (remote) {
495 // A task sent to a linked environment has no thread here; its thread lives there.
496 const { threadId, environmentId } = remote;
497 byTask.set(taskId, { threadId, environmentId, title: title ?? remote.title });
498 }
499 };
500 for (const { item } of items) {
501 if (item.type === "subagent") {
502 remember(item.subagentId, item.childThreadId, item.remoteChild, null);
503 }
504 }
505 for (const subagent of projection.subagents) {
506 remember(subagent.id, subagent.childThreadId, subagent.remoteChild, subagent.title);
507 }
508
509 const found = new Map<string, SubagentChild>();
510 const add = (child: SubagentChild | undefined) => {
511 if (child) found.set(`${child.environmentId ?? ""}:${child.threadId}`, child);
512 };
513 for (; index >= 0 && isInput(index); index -= 1) {
514 const item = items[index]!.item;
515 if (item.type === "notification") {
516 // A notice names its child itself, so it counts even when the task is no longer loaded.
517 if ("childThreadId" in item.source && item.source.childThreadId !== undefined) {
518 const threadId = item.source.childThreadId;
519 add(local.get(threadId) ?? { threadId, title: null });
520 }
521 for (const taskId of "taskIds" in item.source ? item.source.taskIds : []) {
522 add(byTask.get(taskId));
523 }
524 } else if (item.type === "user_message") {
525 const message = projection.messages.find(({ id }) => id === item.messageId);
526 for (const taskId of message?.delegatedCompletion?.taskIds ?? []) add(byTask.get(taskId));
527 // A sender counts only as this thread's subagent: a child's first message comes from its parent.
528 if (item.senderThreadId !== undefined) add(local.get(item.senderThreadId));
529 }
530 }
531 return [...found.values()].map(({ title, ...child }) => ({
532 ...child,
533 title:
534 title?.trim() ||
535 appAtomRegistry
536 .get(
537 environmentThreadShells.quietThreadShellAtom({
538 environmentId: child.environmentId ?? ref.environmentId,
539 threadId: child.threadId,
540 }),
541 )
542 ?.title.trim() ||
543 "Agent",
544 }));
545}