1import type { OrchestrationV2ThreadProjection, ScopedThreadRef } from "@t3tools/contracts";
2import * as Option from "effect/Option";
3import { AsyncResult, Atom } from "effect/reactivity";
4
5import {
6 deriveThreadQueueWorkflowState,
7 getUserQueuedThreadRuns,
8 type ThreadQueueWorkflowState,
9} from "./threadWorkflows.ts";
10import { deriveThreadTurnSubagents, type ThreadTurnSubagents } from "./threadSubagents.ts";
11import type { EnvironmentThread } from "./models.ts";
12import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threadState.ts";
13import {
14 createQuestionHistoryProjector,
15 derivePendingThreadRequests,
16 type PendingThreadRequests,
17} from "./threadRequests.ts";
18import { arrayElementsEqual, parseThreadKey, threadKey } from "./entities.ts";
19
20const EMPTY_VISIBLE_TURN_ITEMS: OrchestrationV2ThreadProjection["visibleTurnItems"] = Object.freeze(
21 [],
22);
23
24export function createEnvironmentThreadDetailAtoms<E>(
25 threadStateAtom: (
26 environmentId: ScopedThreadRef["environmentId"],
27 threadId: ScopedThreadRef["threadId"],
28 ) => Atom.Atom<AsyncResult.AsyncResult<EnvironmentThreadState, E>>,
29) {
30 const threadStateValueAtomFamily = Atom.family((key: string) => {
31 const ref = parseThreadKey(key);
32 return Atom.make((get) =>
33 Option.getOrElse(
34 AsyncResult.value(get(threadStateAtom(ref.environmentId, ref.threadId))),
35 () => EMPTY_ENVIRONMENT_THREAD_STATE,
36 ),
37 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-state-value:${key}`));
38 });
39
40 const threadAtomFamily = Atom.family((key: string) => {
41 const ref = parseThreadKey(key);
42 let previousProjection: OrchestrationV2ThreadProjection | null = null;
43 let previousValue: EnvironmentThread | null = null;
44 return Atom.make((get) => {
45 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
46 if (projection === previousProjection) return previousValue;
47 previousProjection = projection;
48 previousValue = projection === null ? null : { environmentId: ref.environmentId, projection };
49 return previousValue;
50 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread:${key}`));
51 });
52
53 const visibleTurnItemsAtomFamily = Atom.family((key: string) => {
54 const projectQuestionHistory = createQuestionHistoryProjector();
55 return Atom.make((get): OrchestrationV2ThreadProjection["visibleTurnItems"] => {
56 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
57 return projection === null ? EMPTY_VISIBLE_TURN_ITEMS : projectQuestionHistory(projection);
58 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-visible-turn-items:${key}`));
59 });
60
61 const queueWorkflowAtomFamily = Atom.family((key: string) => {
62 let previous: Pick<
63 OrchestrationV2ThreadProjection,
64 "thread" | "runs" | "messages" | "providerThreads" | "providerSessions" | "providerTurns"
65 > | null = null;
66 let value: ThreadQueueWorkflowState | null = null;
67 return Atom.make((get) => {
68 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
69 if (projection === null) {
70 previous = null;
71 value = null;
72 } else if (
73 previous === null ||
74 projection.thread !== previous.thread ||
75 projection.runs !== previous.runs ||
76 projection.messages !== previous.messages ||
77 projection.providerThreads !== previous.providerThreads ||
78 projection.providerSessions !== previous.providerSessions ||
79 projection.providerTurns !== previous.providerTurns
80 ) {
81 value = deriveThreadQueueWorkflowState(projection);
82 const { thread, runs, messages, providerThreads, providerSessions, providerTurns } =
83 projection;
84 previous = { thread, runs, messages, providerThreads, providerSessions, providerTurns };
85 }
86 return value;
87 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-queue:${key}`));
88 });
89 const queuedCountAtomFamily = Atom.family((key: string) => {
90 let previous: Pick<OrchestrationV2ThreadProjection, "runs" | "messages"> | null = null;
91 let count = 0;
92 return Atom.make((get) => {
93 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
94 if (projection === null) {
95 previous = null;
96 count = 0;
97 } else if (projection.runs !== previous?.runs || projection.messages !== previous?.messages) {
98 const { runs, messages } = projection;
99 previous = { runs, messages };
100 count = getUserQueuedThreadRuns(previous).length;
101 }
102 return count;
103 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-queue-count:${key}`));
104 });
105
106 const turnSubagentsAtomFamily = Atom.family((key: string) => {
107 let previous: Pick<OrchestrationV2ThreadProjection, "runs" | "subagents"> | null = null;
108 let value: ThreadTurnSubagents | null = null;
109 return Atom.make((get) => {
110 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
111 if (projection === null) {
112 previous = null;
113 value = null;
114 } else if (
115 projection.runs !== previous?.runs ||
116 projection.subagents !== previous?.subagents
117 ) {
118 const { runs, subagents } = projection;
119 previous = { runs, subagents };
120 value = deriveThreadTurnSubagents(previous);
121 }
122 return value;
123 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-turn-subagents:${key}`));
124 });
125
126 const worktreePathAtomFamily = Atom.family((key: string) =>
127 Atom.make(
128 (get) =>
129 Option.getOrNull(get(threadStateValueAtomFamily(key)).data)?.thread.worktreePath ?? null,
130 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-worktree:${key}`)),
131 );
132 const pendingRequestsAtomFamily = Atom.family((key: string) => {
133 let previous: Pick<OrchestrationV2ThreadProjection, "runtimeRequests" | "turnItems"> | null =
134 null;
135 let value: PendingThreadRequests | null = null;
136 return Atom.make((get) => {
137 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
138 if (projection === null) {
139 previous = null;
140 value = null;
141 return value;
142 }
143 const runtimeRequests = projection.runtimeRequests.filter(
144 (request) => request.status === "pending",
145 );
146 const ids = new Set(runtimeRequests.map((request) => request.id));
147 const turnItems =
148 ids.size === 0
149 ? []
150 : projection.turnItems.filter(
151 (item) =>
152 (item.type === "approval_request" || item.type === "user_input_request") &&
153 ids.has(item.requestId),
154 );
155 if (
156 previous === null ||
157 !arrayElementsEqual(previous.runtimeRequests, runtimeRequests) ||
158 !arrayElementsEqual(previous.turnItems, turnItems)
159 ) {
160 previous = { runtimeRequests, turnItems };
161 value = derivePendingThreadRequests(previous);
162 }
163 return value;
164 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-pending-requests:${key}`));
165 });
166
167 const statusAtomFamily = Atom.family((key: string) =>
168 Atom.make((get) => get(threadStateValueAtomFamily(key)).status).pipe(
169 Atom.setIdleTTL(0),
170 Atom.withLabel(`environment-thread-status:${key}`),
171 ),
172 );
173 const errorAtomFamily = Atom.family((key: string) =>
174 Atom.make((get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).error)).pipe(
175 Atom.setIdleTTL(0),
176 Atom.withLabel(`environment-thread-error:${key}`),
177 ),
178 );
179 const historyAtomFamily = Atom.family((key: string) =>
180 Atom.make((get) => get(threadStateValueAtomFamily(key)).history).pipe(
181 Atom.setIdleTTL(0),
182 Atom.withLabel(`environment-thread-history:${key}`),
183 ),
184 );
185
186 return {
187 worktreePathAtom: (ref: ScopedThreadRef) => worktreePathAtomFamily(threadKey(ref)),
188 pendingRequestsAtom: (ref: ScopedThreadRef) => pendingRequestsAtomFamily(threadKey(ref)),
189 queueWorkflowAtom: (ref: ScopedThreadRef) => queueWorkflowAtomFamily(threadKey(ref)),
190 queuedCountAtom: (ref: ScopedThreadRef) => queuedCountAtomFamily(threadKey(ref)),
191 turnSubagentsAtom: (ref: ScopedThreadRef) => turnSubagentsAtomFamily(threadKey(ref)),
192 stateAtom: (ref: ScopedThreadRef) => threadStateValueAtomFamily(threadKey(ref)),
193 threadAtom: (ref: ScopedThreadRef) => threadAtomFamily(threadKey(ref)),
194 visibleTurnItemsAtom: (ref: ScopedThreadRef) => visibleTurnItemsAtomFamily(threadKey(ref)),
195 statusAtom: (ref: ScopedThreadRef) => statusAtomFamily(threadKey(ref)),
196 errorAtom: (ref: ScopedThreadRef) => errorAtomFamily(threadKey(ref)),
197 historyAtom: (ref: ScopedThreadRef) => historyAtomFamily(threadKey(ref)),
198 };
199}