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
335export function waitForThreadShell(ref: ScopedThreadRef, timeoutMs = 5_000): Promise<boolean> {
336 return waitForAtomValue({
337 registry: appAtomRegistry,
338 atom: environmentThreadShells.threadShellAtom(ref),
339 predicate: (thread) => thread !== null,
340 timeoutMs,
341 });
342}
343
344export function useEnvironmentSupportsServerBrowser(environmentId: EnvironmentId | null): boolean {
345 const configs = useServerConfigs();
346 return (
347 environmentId !== null &&
348 configs.get(environmentId)?.environment.capabilities.serverBrowser === true
349 );
350}
351
352export function readEnvironmentSupportsServerBrowser(environmentId: EnvironmentId): boolean {
353 return (
354 appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities
355 .serverBrowser === true
356 );
357}
358
359export function readEnvironmentServerConfigLoaded(environmentId: EnvironmentId): boolean {
360 return appAtomRegistry.get(environmentServerConfigsAtom).has(environmentId);
361}
362
363export function readEnvironmentSupportsAutoSettleOptOut(environmentId: EnvironmentId): boolean {
364 return (
365 appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities
366 .threadAutoSettleOptOut === true
367 );
368}
369
370export function readEnvironmentThreadRefs(
371 environmentId: EnvironmentId,
372): ReadonlyArray<ScopedThreadRef> {
373 return appAtomRegistry.get(environmentThreadShells.environmentThreadRefsAtom(environmentId));
374}
375
376export function readThreadShells(): ReadonlyArray<EnvironmentThreadShell> {
377 return appAtomRegistry.get(environmentThreadShells.threadShellsAtom);
378}
379
380function shellParentThreadId(
381 thread: Pick<OrchestrationV2ThreadShell, "forkedFrom" | "lineage">,
382): ThreadId | null {
383 return thread.forkedFrom?.type === "run"
384 ? thread.forkedFrom.threadId
385 : thread.lineage.parentThreadId;
386}
387
388/** Retains unchanged reachable shells so unrelated streams do not re-render the panel. */
389export function selectThreadRelationshipShells(
390 threadId: ThreadId,
391 projection: Pick<
392 OrchestrationV2ThreadProjection,
393 "thread" | "subagents" | "contextTransfers"
394 > | null,
395 threads: ReadonlyArray<OrchestrationV2ThreadShell>,
396): ReadonlyArray<OrchestrationV2ThreadShell> {
397 const named = new Set<string>([threadId]);
398 if (projection !== null) {
399 const parentThreadId = shellParentThreadId(projection.thread);
400 if (parentThreadId !== null) named.add(parentThreadId);
401 for (const subagent of projection.subagents) {
402 if (subagent.childThreadId !== null) named.add(subagent.childThreadId);
403 for (const agent of subagent.workflow?.agents ?? []) {
404 if (agent.childThreadId !== undefined) named.add(agent.childThreadId);
405 }
406 }
407 for (const transfer of projection.contextTransfers) {
408 named.add(transfer.sourceThreadId);
409 named.add(transfer.targetThreadId);
410 }
411 }
412 const current = threads.find((thread) => thread.id === threadId);
413 const currentParentId = current === undefined ? null : shellParentThreadId(current);
414 if (currentParentId !== null) named.add(currentParentId);
415 return threads.filter(
416 (thread) => named.has(thread.id) || shellParentThreadId(thread) === threadId,
417 );
418}
419
420const relationshipShellsAtomFamily = Atom.family((key: string) => {
421 const ref = parseThreadKey(key);
422 let previous: ReadonlyArray<OrchestrationV2ThreadShell> = [];
423 return Atom.make((get) => {
424 const next = selectThreadRelationshipShells(
425 ref.threadId,
426 get(environmentThreadDetails.threadAtom(ref))?.projection ?? null,
427 get(environmentThreadShells.environmentThreadsAtom(ref.environmentId)),
428 );
429 if (arrayElementsEqual(previous, next)) return previous;
430 previous = next;
431 return next;
432 }).pipe(Atom.withLabel(`web-thread-relationship-shells:${key}`));
433});
434
435export function useThreadRelationshipShells(
436 ref: ScopedThreadRef,
437): ReadonlyArray<OrchestrationV2ThreadShell> {
438 return useAtomValue(relationshipShellsAtomFamily(threadKey(ref)));
439}
440
441interface SubagentChild {
442 readonly threadId: ThreadId;
443 /** Set when the task ran on a linked environment, which holds its thread. */
444 readonly environmentId?: EnvironmentId;
445 readonly title: string | null | undefined;
446}
447
448/**
449 * The subagents a response reported on, read once on demand so a timeline row does not subscribe
450 * to the thread for it. A result reaches the thread as a message from the subagent's thread or as
451 * a completion notice, at the start of a run or steered into one, so the inputs just before the
452 * response decide it. Turn items are read because older history loads nothing else.
453 */
454export function readResponseSubagents(
455 ref: ScopedThreadRef,
456 itemId: TurnItemId,
457): ReadonlyArray<Omit<SubagentChild, "title"> & { readonly title: string }> {
458 const projection = appAtomRegistry.get(environmentThreadDetails.quietThreadAtom(ref))?.projection;
459 if (!projection) return [];
460 const items = projection.visibleTurnItems;
461 const isInput = (index: number) => {
462 const type = items[index]?.item.type;
463 return type === "user_message" || type === "notification";
464 };
465 let index = items.findIndex((projected) => projected.sourceItemId === itemId) - 1;
466 while (index >= 0 && !isInput(index)) index -= 1;
467
468 const byTask = new Map<NodeId, SubagentChild>();
469 const local = new Map<ThreadId, SubagentChild>();
470 const remember = (
471 taskId: NodeId,
472 childThreadId: ThreadId | null,
473 remote: OrchestrationV2RemoteTaskChild | undefined,
474 title: string | null,
475 ) => {
476 if (childThreadId !== null) {
477 const child = { threadId: childThreadId, title };
478 byTask.set(taskId, child);
479 local.set(childThreadId, child);
480 } else if (remote) {
481 // A task sent to a linked environment has no thread here; its thread lives there.
482 const { threadId, environmentId } = remote;
483 byTask.set(taskId, { threadId, environmentId, title: title ?? remote.title });
484 }
485 };
486 for (const { item } of items) {
487 if (item.type === "subagent") {
488 remember(item.subagentId, item.childThreadId, item.remoteChild, null);
489 }
490 }
491 for (const subagent of projection.subagents) {
492 remember(subagent.id, subagent.childThreadId, subagent.remoteChild, subagent.title);
493 }
494
495 const found = new Map<string, SubagentChild>();
496 const add = (child: SubagentChild | undefined) => {
497 if (child) found.set(`${child.environmentId ?? ""}:${child.threadId}`, child);
498 };
499 for (; index >= 0 && isInput(index); index -= 1) {
500 const item = items[index]!.item;
501 if (item.type === "notification") {
502 // A notice names its child itself, so it counts even when the task is no longer loaded.
503 if ("childThreadId" in item.source && item.source.childThreadId !== undefined) {
504 const threadId = item.source.childThreadId;
505 add(local.get(threadId) ?? { threadId, title: null });
506 }
507 for (const taskId of "taskIds" in item.source ? item.source.taskIds : []) {
508 add(byTask.get(taskId));
509 }
510 } else if (item.type === "user_message") {
511 const message = projection.messages.find(({ id }) => id === item.messageId);
512 for (const taskId of message?.delegatedCompletion?.taskIds ?? []) add(byTask.get(taskId));
513 // A sender counts only as this thread's subagent: a child's first message comes from its parent.
514 if (item.senderThreadId !== undefined) add(local.get(item.senderThreadId));
515 }
516 }
517 return [...found.values()].map(({ title, ...child }) => ({
518 ...child,
519 title:
520 title?.trim() ||
521 appAtomRegistry
522 .get(
523 environmentThreadShells.quietThreadShellAtom({
524 environmentId: child.environmentId ?? ref.environmentId,
525 threadId: child.threadId,
526 }),
527 )
528 ?.title.trim() ||
529 "Agent",
530 }));
531}