1import { useAtomValue } from "@effect/atom-react";
2import { enabledEnvironmentIds } from "@t3tools/client-runtime/state/connections";
3import { arrayElementsEqual } from "@t3tools/client-runtime/state/entities";
4import {
5 createEnvironmentThreadDetailAtoms,
6 createEnvironmentThreadShellAtoms,
7 createEnvironmentThreadStateAtoms,
8 EMPTY_ENVIRONMENT_THREAD_STATE,
9 type EnvironmentThreadState,
10 createThreadEnvironmentAtoms,
11} from "@t3tools/client-runtime/state/threads";
12import type { EnvironmentId, OrchestrationV2ThreadShell, ThreadId } from "@t3tools/contracts";
13import * as Option from "effect/Option";
14import { AsyncResult, Atom } from "effect/reactivity";
15
16import { environmentCatalog } from "../connection/catalog";
17import { connectionAtomRuntime } from "../connection/runtime";
18import { environmentSnapshotAtom } from "./shell";
19
20export const threadEnvironment = createThreadEnvironmentAtoms(
21 connectionAtomRuntime,
22 environmentSnapshotAtom,
23);
24const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime);
25export const environmentThreadDetails = createEnvironmentThreadDetailAtoms(
26 environmentThreads.stateAtom,
27);
28export const environmentThreadShells = createEnvironmentThreadShellAtoms({
29 catalogValueAtom: environmentCatalog.catalogValueAtom,
30 snapshotAtom: threadEnvironment.snapshotAtom,
31});
32
33const EMPTY_THREAD_STATE_ATOM = Atom.make(AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE)).pipe(
34 Atom.withLabel("web-environment-thread:empty"),
35);
36
37export function useEnvironmentThread(
38 environmentId: EnvironmentId | null,
39 threadId: ThreadId | null,
40): EnvironmentThreadState {
41 const result = useAtomValue(
42 environmentId !== null && threadId !== null
43 ? environmentThreads.stateAtom(environmentId, threadId)
44 : EMPTY_THREAD_STATE_ATOM,
45 );
46 const state = Option.getOrElse(
47 AsyncResult.value(result),
48 () => EMPTY_ENVIRONMENT_THREAD_STATE,
49 ) as EnvironmentThreadState;
50 return state;
51}
52
53const isRunning = (status: string) =>
54 status === "preparing" || status === "queued" || status === "starting" || status === "running";
55
56type KeptThreads = ReadonlyMap<EnvironmentId, ReadonlySet<ThreadId>>;
57
58// True once a thread's own stream no longer needs to stay open: it is in sync
59// and shows a settled session, or it cannot progress (deleted or failed). A
60// stream that is still loading or reconnecting keeps waiting for the stop.
61function isDetailDone<E>(result: AsyncResult.AsyncResult<EnvironmentThreadState, E>): boolean {
62 if (!AsyncResult.isSuccess(result)) return true;
63 const { status, data, error } = result.value;
64 if (status === "deleted" || Option.isSome(error)) return true;
65 return (
66 status === "live" &&
67 !Option.exists(data, (thread) => thread.runs.some((run) => isRunning(run.status)))
68 );
69}
70
71/**
72 * Keeps the thread state atom mounted for each running thread in the listed
73 * environments. Mount the result; its value is only bookkeeping.
74 *
75 * The shell and detail streams are independent, so the shell can report a
76 * stop before the detail loads or catches up. A stopped thread stays mounted
77 * until its own detail is live and shows the stop too. Then the stream closes
78 * and saves the settled state to disk.
79 */
80export function createRunningThreadKeepAliveAtom<E>(input: {
81 readonly environmentIdsAtom: Atom.Atom<ReadonlyArray<EnvironmentId>>;
82 readonly threadsAtom: (
83 environmentId: EnvironmentId,
84 ) => Atom.Atom<ReadonlyArray<Pick<OrchestrationV2ThreadShell, "id" | "status">>>;
85 readonly stateAtom: (
86 environmentId: EnvironmentId,
87 threadId: ThreadId,
88 ) => Atom.Atom<AsyncResult.AsyncResult<EnvironmentThreadState, E>>;
89}) {
90 // Keeps its identity until a thread starts or stops, so ordinary shell
91 // updates do not rebuild the keep-alive set.
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 // An environment that leaves the list is not visited, so its mounts drop.
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 // `once`, not `get`: a dependency on a stopped thread would hold its
114 // stream open until some other change rebuilds this atom.
115 if (isDetailDone(get.once(stateAtom))) continue;
116 threadIds.add(threadId);
117 // Rebuild when this detail is done, not on each update.
118 get.subscribe(stateAtom, (state) => {
119 if (isDetailDone(state)) get.refreshSelf();
120 });
121 }
122 for (const threadId of threadIds) get.mount(input.stateAtom(environmentId, threadId));
123 kept.set(environmentId, threadIds);
124 }
125 return kept;
126 }).pipe(Atom.withLabel("web-running-thread-keep-alive"));
127}
128
129/** Mounted by `RunningThreadKeepAlive` on desktop, for every enabled environment. */
130export const runningThreadKeepAliveAtom = createRunningThreadKeepAliveAtom({
131 environmentIdsAtom: Atom.map(environmentCatalog.catalogValueAtom, (catalog) => [
132 ...enabledEnvironmentIds(catalog),
133 ]),
134 threadsAtom: environmentThreadShells.environmentThreadsAtom,
135 stateAtom: environmentThreads.stateAtom,
136});