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