1import { createChildThreadApprovalsAtom } from "@t3tools/client-runtime/state/thread-requests";
2import { useAtomValue } from "@effect/atom-react";
3import { enabledEnvironmentIds } from "@t3tools/client-runtime/state/connections";
4import { arrayElementsEqual } from "@t3tools/client-runtime/state/entities";
5import {
6 createEnvironmentThreadDetailAtoms,
7 createEnvironmentThreadShellAtoms,
8 createEnvironmentThreadStateAtoms,
9 EMPTY_ENVIRONMENT_THREAD_STATE,
10 type EnvironmentThreadState,
11 createOlderSettledThreadAtoms,
12 createThreadEnvironmentAtoms,
13 createThreadFamilyAtoms,
14} from "@t3tools/client-runtime/state/threads";
15import type { EnvironmentId, OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts";
16import * as Option from "effect/Option";
17import { AsyncResult, Atom } from "effect/reactivity";
18
19import { environmentCatalog } from "../connection/catalog";
20import { connectionAtomRuntime } from "../connection/runtime";
21import { serverEnvironment } from "./server";
22import { environmentShell } from "./shell";
23
24export const threadFamilies = createThreadFamilyAtoms(
25 connectionAtomRuntime,
26 environmentShell.stateAtom,
27);
28export const olderSettledThreads = createOlderSettledThreadAtoms(
29 connectionAtomRuntime,
30 environmentShell.stateAtom,
31);
32export const threadEnvironment = createThreadEnvironmentAtoms(
33 connectionAtomRuntime,
34 threadFamilies.snapshotAtom,
35);
36const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime);
37export const environmentThreadDetails = createEnvironmentThreadDetailAtoms(
38 environmentThreads.stateAtom,
39);
40export const environmentThreadShells = createEnvironmentThreadShellAtoms({
41 catalogValueAtom: environmentCatalog.catalogValueAtom,
42 snapshotAtom: threadEnvironment.snapshotAtom,
43 hiddenThreadTagsAtom: serverEnvironment.hiddenThreadTagsAtom,
44});
45export const childThreadApprovalsAtom = createChildThreadApprovalsAtom({
46 childThreadInputsAtom: environmentThreadShells.childThreadInputsAtom,
47 pendingRequestsAtom: environmentThreadDetails.pendingRequestsAtom,
48});
49
50const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe(
51 Atom.withLabel("web-environment-thread:empty"),
52);
53
54export function useEnvironmentThread(
55 environmentId: EnvironmentId | null,
56 threadId: ThreadId | null,
57): EnvironmentThreadState {
58 const result = useAtomValue(
59 environmentId !== null && threadId !== null
60 ? environmentThreads.stateAtom(environmentId, threadId)
61 : EMPTY_THREAD_STATE_ATOM,
62 );
63 const state = AsyncResult.getOrElse(result, () => EMPTY_ENVIRONMENT_THREAD_STATE);
64 return state;
65}
66
67const isRunning = (status: string) =>
68 status === "preparing" || status === "queued" || status === "starting" || status === "running";
69
70type KeptThreads = ReadonlyMap<EnvironmentId, ReadonlySet<ThreadId>>;
71
72function isDetailDone<E>(result: AsyncResult.AsyncResult<EnvironmentThreadState, E>): boolean {
73 if (!AsyncResult.isSuccess(result)) return true;
74 const { status, data, error } = result.value;
75 if (status === "deleted" || Option.isSome(error)) return true;
76 return (
77 status === "live" &&
78 !Option.exists(data, (thread) => thread.runs.some((run) => isRunning(run.status)))
79 );
80}
81
82/** Keeps stopped threads mounted until their independent detail stream is live and also reports stopped. */
83export function createRunningThreadKeepAliveAtom<E>(input: {
84 readonly environmentIdsAtom: Atom.Atom<ReadonlyArray<EnvironmentId>>;
85 readonly threadsAtom: (
86 environmentId: EnvironmentId,
87 ) => Atom.Atom<ReadonlyArray<Pick<OrchestrationV2ThreadShell, "id" | "status">>>;
88 readonly stateAtom: (
89 environmentId: EnvironmentId,
90 threadId: ThreadId,
91 ) => Atom.Atom<AsyncResult.AsyncResult<EnvironmentThreadState, E>>;
92}) {
93 const runningThreadIdsAtom = Atom.family((environmentId: EnvironmentId) => {
94 let previous: ReadonlyArray<ThreadId> = [];
95 return Atom.make((get) => {
96 const running = get(input.threadsAtom(environmentId)).flatMap((thread) =>
97 isRunning(thread.status) ? [thread.id] : [],
98 );
99 if (arrayElementsEqual(previous, running)) return previous;
100 previous = running;
101 return running;
102 }).pipe(Atom.withLabel(`web-running-thread-ids:${environmentId}`));
103 });
104
105 return Atom.make((get): KeptThreads => {
106 const previous = Option.getOrUndefined(get.self<KeptThreads>());
107 const kept = new Map<EnvironmentId, ReadonlySet<ThreadId>>();
108 for (const environmentId of get(input.environmentIdsAtom)) {
109 const threadIds = new Set(get(runningThreadIdsAtom(environmentId)));
110 for (const threadId of previous?.get(environmentId) ?? []) {
111 if (threadIds.has(threadId)) continue;
112 const stateAtom = input.stateAtom(environmentId, threadId);
113 // A get dependency would hold stopped streams open until another change rebuilds this atom.
114 if (isDetailDone(get.once(stateAtom))) continue;
115 threadIds.add(threadId);
116 get.subscribe(stateAtom, (state) => {
117 if (isDetailDone(state)) get.refreshSelf();
118 });
119 }
120 for (const threadId of threadIds) get.mount(input.stateAtom(environmentId, threadId));
121 kept.set(environmentId, threadIds);
122 }
123 return kept;
124 }).pipe(Atom.withLabel("web-running-thread-keep-alive"));
125}
126
127export const runningThreadKeepAliveAtom = createRunningThreadKeepAliveAtom({
128 environmentIdsAtom: Atom.map(environmentCatalog.catalogValueAtom, (catalog) => [
129 ...enabledEnvironmentIds(catalog),
130 ]),
131 threadsAtom: environmentThreadShells.environmentThreadsAtom,
132 stateAtom: environmentThreads.stateAtom,
133});