1import type { ThreadId } from "@t3tools/contracts";
2import * as Crypto from "effect/Crypto";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import { Atom } from "effect/reactivity";
6import {
7 WS_METHODS,
8 type EnvironmentId,
9 type OrchestrationV2ShellSnapshot,
10} from "@t3tools/contracts";
11
12import { createOptimisticThreadLifecycle } from "./threadLifecycle.ts";
13import * as DateTime from "effect/DateTime";
14
15import {
16 createAtomCommandScheduler,
17 createEnvironmentCommand,
18 createEnvironmentRpcCommand,
19} from "./runtime.ts";
20import {
21 type ThreadCommandInput,
22 type ArchiveThreadInput,
23 type CancelQueuedRunInput,
24 type RetryWorkspacePreparationInput,
25 type CreateThreadInput,
26 type DeleteThreadInput,
27 type EditQueuedRunInput,
28 type InterruptThreadTurnInput,
29 type MarkThreadUnreadInput,
30 type ForkThreadFromRunInput,
31 type MergeThreadBackInput,
32 type PromoteQueuedRunInput,
33 type ReorderQueuedRunInput,
34 type LinkThreadPullRequestInput,
35 type RespondToThreadApprovalInput,
36 type RespondToThreadUserInputInput,
37 type DismissThreadUserInputInput,
38 type RevertThreadCheckpointInput,
39 type SetThreadInteractionModeInput,
40 type SetThreadRuntimeModeInput,
41 type PinThreadInput,
42 type ReorderPinnedThreadInput,
43 type ReorderActiveThreadInput,
44 type SetThreadAutoSettleInput,
45 type SettleThreadInput,
46 type SnoozeThreadInput,
47 type StartThreadTurnInput,
48 type StopThreadSessionInput,
49 type UnarchiveThreadInput,
50 type UnlinkThreadPullRequestInput,
51 type UnpinThreadInput,
52 type WatchThreadPullRequestInput,
53 type UnsettleThreadInput,
54 type UnsnoozeThreadInput,
55 type UpdateThreadMetadataInput,
56 type VisitThreadInput,
57 archiveThread,
58 cancelQueuedRun,
59 createThread,
60 deleteThread,
61 editQueuedRun,
62 interruptThreadTurn,
63 forkThreadFromRun,
64 markThreadUnread,
65 mergeThreadBack,
66 promoteQueuedRun,
67 reorderQueuedRun,
68 resumeThreadQueue,
69 retryWorkspacePreparation,
70 linkThreadPullRequest,
71 respondToThreadApproval,
72 respondToThreadUserInput,
73 dismissThreadUserInput,
74 revertThreadCheckpoint,
75 setThreadInteractionMode,
76 setThreadRuntimeMode,
77 pinThread,
78 reorderPinnedThread,
79 reorderActiveThread,
80 setThreadAutoSettle,
81 settleThread,
82 snoozeThread,
83 startThreadTurn,
84 stopThreadSession,
85 unarchiveThread,
86 unlinkThreadPullRequest,
87 unpinThread,
88 unsettleThread,
89 unsnoozeThread,
90 updateThreadMetadata,
91 visitThread,
92 watchThreadPullRequest,
93} from "../operations/commands.ts";
94import type { EnvironmentRegistry } from "../connection/registry.ts";
95import * as EnvironmentSupervisor from "../connection/supervisor.ts";
96import * as ThreadHistoryController from "./threadHistoryController.ts";
97
98export type LoadEarlierThreadHistoryInput = {
99 readonly threadId: ThreadId;
100 readonly throughEntryId?: string;
101};
102
103export type {
104 ArchiveThreadInput,
105 CancelQueuedRunInput,
106 CreateThreadInput,
107 DeleteThreadInput,
108 EditQueuedRunInput,
109 InterruptThreadTurnInput,
110 MarkThreadUnreadInput,
111 ForkThreadFromRunInput,
112 MergeThreadBackInput,
113 PromoteQueuedRunInput,
114 ReorderQueuedRunInput,
115 LinkThreadPullRequestInput,
116 RespondToThreadApprovalInput,
117 RespondToThreadUserInputInput,
118 DismissThreadUserInputInput,
119 RevertThreadCheckpointInput,
120 SetThreadInteractionModeInput,
121 SetThreadRuntimeModeInput,
122 PinThreadInput,
123 ReorderPinnedThreadInput,
124 ReorderActiveThreadInput,
125 SetThreadAutoSettleInput,
126 SettleThreadInput,
127 SnoozeThreadInput,
128 StartThreadTurnInput,
129 StopThreadSessionInput,
130 ThreadCommandInput,
131 UnarchiveThreadInput,
132 UnlinkThreadPullRequestInput,
133 UnpinThreadInput,
134 UnsettleThreadInput,
135 UnsnoozeThreadInput,
136 UpdateThreadMetadataInput,
137 VisitThreadInput,
138 WatchThreadPullRequestInput,
139} from "../operations/commands.ts";
140
141export function createThreadEnvironmentAtoms<R, E>(
142 runtime: Atom.AtomRuntime<EnvironmentRegistry | Crypto.Crypto | R, E>,
143 snapshotAtom: (environmentId: EnvironmentId) => Atom.Atom<OrchestrationV2ShellSnapshot | null>,
144) {
145 const scheduler = createAtomCommandScheduler();
146 const concurrency = {
147 mode: "serial" as const,
148 key: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) =>
149 JSON.stringify([environmentId, input.threadId]),
150 };
151 const commands = {
152 create: createEnvironmentCommand(runtime, {
153 label: "environment-data:commands:thread:create",
154 execute: (input: CreateThreadInput) => createThread(input),
155 scheduler,
156 concurrency,
157 }),
158 delete: createEnvironmentCommand(runtime, {
159 label: "environment-data:commands:thread:delete",
160 execute: (input: DeleteThreadInput) => deleteThread(input),
161 scheduler,
162 concurrency,
163 }),
164 archive: createEnvironmentCommand(runtime, {
165 label: "environment-data:commands:thread:archive",
166 execute: (input: ArchiveThreadInput) => archiveThread(input),
167 scheduler,
168 concurrency,
169 }),
170 unarchive: createEnvironmentCommand(runtime, {
171 label: "environment-data:commands:thread:unarchive",
172 execute: (input: UnarchiveThreadInput) => unarchiveThread(input),
173 scheduler,
174 concurrency,
175 }),
176 settle: createEnvironmentCommand(runtime, {
177 label: "environment-data:commands:thread:settle",
178 execute: (input: SettleThreadInput) => settleThread(input),
179 scheduler,
180 concurrency,
181 }),
182 unsettle: createEnvironmentCommand(runtime, {
183 label: "environment-data:commands:thread:unsettle",
184 execute: (input: UnsettleThreadInput) => unsettleThread(input),
185 scheduler,
186 concurrency,
187 }),
188 snooze: createEnvironmentCommand(runtime, {
189 label: "environment-data:commands:thread:snooze",
190 execute: (input: SnoozeThreadInput) => snoozeThread(input),
191 scheduler,
192 concurrency,
193 }),
194 unsnooze: createEnvironmentCommand(runtime, {
195 label: "environment-data:commands:thread:unsnooze",
196 execute: (input: UnsnoozeThreadInput) => unsnoozeThread(input),
197 scheduler,
198 concurrency,
199 }),
200 pin: createEnvironmentCommand(runtime, {
201 label: "environment-data:commands:thread:pin",
202 execute: (input: PinThreadInput) => pinThread(input),
203 scheduler,
204 concurrency,
205 }),
206 unpin: createEnvironmentCommand(runtime, {
207 label: "environment-data:commands:thread:unpin",
208 execute: (input: UnpinThreadInput) => unpinThread(input),
209 scheduler,
210 concurrency,
211 }),
212 reorderPin: createEnvironmentCommand(runtime, {
213 label: "environment-data:commands:thread:reorder-pin",
214 execute: (input: ReorderPinnedThreadInput) => reorderPinnedThread(input),
215 scheduler,
216 concurrency,
217 }),
218 setAutoSettle: createEnvironmentCommand(runtime, {
219 label: "environment-data:commands:thread:set-auto-settle",
220 execute: (input: SetThreadAutoSettleInput) => setThreadAutoSettle(input),
221 scheduler,
222 concurrency,
223 }),
224 reorderActive: createEnvironmentCommand(runtime, {
225 label: "environment-data:commands:thread:reorder-active",
226 execute: (input: ReorderActiveThreadInput) => reorderActiveThread(input),
227 scheduler,
228 concurrency,
229 }),
230 visit: createEnvironmentCommand(runtime, {
231 label: "environment-data:commands:thread:visit",
232 execute: (input: VisitThreadInput) => visitThread(input),
233 scheduler,
234 concurrency,
235 }),
236 markUnread: createEnvironmentCommand(runtime, {
237 label: "environment-data:commands:thread:mark-unread",
238 execute: (input: MarkThreadUnreadInput) => markThreadUnread(input),
239 scheduler,
240 concurrency,
241 }),
242 updateMetadata: createEnvironmentCommand(runtime, {
243 label: "environment-data:commands:thread:update-metadata",
244 execute: (input: UpdateThreadMetadataInput) => updateThreadMetadata(input),
245 scheduler,
246 concurrency,
247 }),
248 linkPullRequest: createEnvironmentCommand(runtime, {
249 label: "environment-data:commands:thread:link-pull-request",
250 execute: (input: LinkThreadPullRequestInput) => linkThreadPullRequest(input),
251 scheduler,
252 concurrency,
253 }),
254 unlinkPullRequest: createEnvironmentCommand(runtime, {
255 label: "environment-data:commands:thread:unlink-pull-request",
256 execute: (input: UnlinkThreadPullRequestInput) => unlinkThreadPullRequest(input),
257 scheduler,
258 concurrency,
259 }),
260 watchPullRequest: createEnvironmentCommand(runtime, {
261 label: "environment-data:commands:thread:watch-pull-request",
262 execute: (input: WatchThreadPullRequestInput) => watchThreadPullRequest(input),
263 scheduler,
264 concurrency,
265 }),
266 setRuntimeMode: createEnvironmentCommand(runtime, {
267 label: "environment-data:commands:thread:set-runtime-mode",
268 execute: (input: SetThreadRuntimeModeInput) => setThreadRuntimeMode(input),
269 scheduler,
270 concurrency,
271 }),
272 setInteractionMode: createEnvironmentCommand(runtime, {
273 label: "environment-data:commands:thread:set-interaction-mode",
274 execute: (input: SetThreadInteractionModeInput) => setThreadInteractionMode(input),
275 scheduler,
276 concurrency,
277 }),
278 startTurn: createEnvironmentCommand(runtime, {
279 label: "environment-data:commands:thread:start-turn",
280 execute: (input: StartThreadTurnInput) => startThreadTurn(input),
281 scheduler,
282 concurrency,
283 }),
284 interruptTurn: createEnvironmentCommand(runtime, {
285 label: "environment-data:commands:thread:interrupt-turn",
286 execute: (input: InterruptThreadTurnInput) => interruptThreadTurn(input),
287 scheduler,
288 concurrency,
289 }),
290 respondToApproval: createEnvironmentCommand(runtime, {
291 label: "environment-data:commands:thread:respond-to-approval",
292 execute: (input: RespondToThreadApprovalInput) => respondToThreadApproval(input),
293 scheduler,
294 concurrency,
295 }),
296 respondToUserInput: createEnvironmentCommand(runtime, {
297 label: "environment-data:commands:thread:respond-to-user-input",
298 execute: (input: RespondToThreadUserInputInput) => respondToThreadUserInput(input),
299 scheduler,
300 concurrency,
301 }),
302 dismissUserInput: createEnvironmentCommand(runtime, {
303 label: "environment-data:commands:thread:dismiss-user-input",
304 execute: (input: DismissThreadUserInputInput) => dismissThreadUserInput(input),
305 scheduler,
306 concurrency,
307 }),
308 revertCheckpoint: createEnvironmentCommand(runtime, {
309 label: "environment-data:commands:thread:revert-checkpoint",
310 execute: (input: RevertThreadCheckpointInput) => revertThreadCheckpoint(input),
311 scheduler,
312 concurrency,
313 }),
314 stopSession: createEnvironmentCommand(runtime, {
315 label: "environment-data:commands:thread:stop-session",
316 execute: (input: StopThreadSessionInput) => stopThreadSession(input),
317 scheduler,
318 concurrency,
319 }),
320 forkFromRun: createEnvironmentCommand(runtime, {
321 label: "environment-data:commands:thread:fork-from-run",
322 execute: (input: ForkThreadFromRunInput) => forkThreadFromRun(input),
323 scheduler,
324 concurrency: {
325 mode: "serial",
326 key: ({ environmentId, input }) => JSON.stringify([environmentId, input.sourceThreadId]),
327 },
328 }),
329 mergeBack: createEnvironmentCommand(runtime, {
330 label: "environment-data:commands:thread:merge-back",
331 execute: (input: MergeThreadBackInput) => mergeThreadBack(input),
332 scheduler,
333 concurrency: {
334 mode: "serial",
335 key: ({ environmentId, input }) =>
336 JSON.stringify([environmentId, input.sourceThreadId, input.targetThreadId]),
337 },
338 }),
339 resumeThreadQueue: createEnvironmentCommand(runtime, {
340 label: "environment-data:commands:thread:resume-queue",
341 execute: (input: ThreadCommandInput) => resumeThreadQueue(input),
342 scheduler,
343 concurrency,
344 }),
345 reorderQueuedRun: createEnvironmentCommand(runtime, {
346 label: "environment-data:commands:thread:reorder-queued-run",
347 execute: (input: ReorderQueuedRunInput) => reorderQueuedRun(input),
348 scheduler,
349 concurrency,
350 }),
351 promoteQueuedRun: createEnvironmentCommand(runtime, {
352 label: "environment-data:commands:thread:promote-queued-run",
353 execute: (input: PromoteQueuedRunInput) => promoteQueuedRun(input),
354 scheduler,
355 concurrency,
356 }),
357 cancelQueuedRun: createEnvironmentCommand(runtime, {
358 label: "environment-data:commands:thread:cancel-queued-run",
359 execute: (input: CancelQueuedRunInput) => cancelQueuedRun(input),
360 scheduler,
361 concurrency,
362 }),
363 retryWorkspacePreparation: createEnvironmentCommand(runtime, {
364 label: "environment-data:commands:thread:retry-workspace-preparation",
365 execute: (input: RetryWorkspacePreparationInput) => retryWorkspacePreparation(input),
366 scheduler,
367 concurrency,
368 }),
369 editQueuedRun: createEnvironmentCommand(runtime, {
370 label: "environment-data:commands:thread:edit-queued-run",
371 execute: (input: EditQueuedRunInput) => editQueuedRun(input),
372 scheduler,
373 concurrency,
374 }),
375 loadEarlierHistory: createEnvironmentCommand(runtime, {
376 label: "environment-data:commands:thread:load-earlier-history",
377 execute: (input: LoadEarlierThreadHistoryInput) =>
378 Effect.gen(function* () {
379 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
380 const controller = yield* Effect.serviceOption(
381 ThreadHistoryController.ThreadHistoryController,
382 );
383 if (Option.isNone(controller)) {
384 return {
385 _tag: "noop",
386 } satisfies ThreadHistoryController.ThreadHistoryLoadEarlierResult;
387 }
388 return yield* controller.value.loadEarlier(
389 supervisor.target.environmentId,
390 input.threadId,
391 input.throughEntryId,
392 );
393 }),
394 scheduler,
395 concurrency: {
396 mode: "serial",
397 key: ({ environmentId, input }) => JSON.stringify([environmentId, input.threadId]),
398 },
399 }),
400 uploadFeedback: createEnvironmentRpcCommand(runtime, {
401 label: "environment-data:commands:thread:upload-feedback",
402 tag: WS_METHODS.providerUploadFeedback,
403 scheduler,
404 concurrency,
405 }),
406 };
407 const optimistic = createOptimisticThreadLifecycle(snapshotAtom);
408 return {
409 ...commands,
410 snapshotAtom: optimistic.snapshotAtom,
411 settle: optimistic.wrap(commands.settle, (thread, _input, now, accepted) =>
412 !accepted &&
413 (thread.pendingRuntimeRequest !== null ||
414 ["preparing", "queued", "starting", "running", "waiting"].includes(thread.status))
415 ? thread
416 : {
417 ...thread,
418 pendingRuntimeRequest: null,
419 settledOverride: "settled",
420 settledAt: thread.settledOverride === "settled" ? (thread.settledAt ?? now) : now,
421 unsettledAt: null,
422 activeOrderKey: null,
423 pinnedAt: null,
424 pinOrderKey: null,
425 snoozedAt: null,
426 snoozedUntil: null,
427 },
428 ),
429 unsettle: optimistic.wrap(commands.unsettle, (thread, input, now) => ({
430 ...thread,
431 settledOverride: input.reason === "user" ? "active" : null,
432 settledAt: null,
433 unsettledAt: thread.settledOverride === "active" ? (thread.unsettledAt ?? null) : now,
434 })),
435 snooze: optimistic.wrap(commands.snooze, (thread, input, now, accepted) =>
436 (!accepted &&
437 (thread.pendingRuntimeRequest !== null ||
438 ["preparing", "queued", "starting"].includes(thread.status))) ||
439 !(Date.parse(input.snoozedUntil) > DateTime.toEpochMillis(now))
440 ? thread
441 : {
442 ...thread,
443 pendingRuntimeRequest: null,
444 snoozedUntil: DateTime.makeUnsafe(input.snoozedUntil),
445 snoozedAt:
446 thread.snoozedUntil != null &&
447 DateTime.formatIso(thread.snoozedUntil) === input.snoozedUntil
448 ? (thread.snoozedAt ?? now)
449 : now,
450 },
451 ),
452 unsnooze: optimistic.wrap(commands.unsnooze, (thread) => ({
453 ...thread,
454 snoozedUntil: null,
455 snoozedAt: null,
456 })),
457 setAutoSettle: optimistic.wrap(commands.setAutoSettle, (thread, input, now) => ({
458 ...thread,
459 autoSettleDisabledAt: input.enabled ? null : (thread.autoSettleDisabledAt ?? now),
460 })),
461 pin: optimistic.wrap(commands.pin, (thread, input, now) => ({
462 ...thread,
463 pinnedAt: thread.pinnedAt ?? now,
464 pinOrderKey: thread.pinnedAt == null ? (input.orderKey ?? null) : thread.pinOrderKey,
465 ...(thread.settledOverride === "settled"
466 ? {
467 settledOverride: "active" as const,
468 settledAt: null,
469 unsettledAt: now,
470 }
471 : {}),
472 snoozedUntil: null,
473 snoozedAt: null,
474 })),
475 unpin: optimistic.wrap(commands.unpin, (thread) => ({
476 ...thread,
477 pinnedAt: null,
478 pinOrderKey: null,
479 })),
480 reorderPin: optimistic.wrap(commands.reorderPin, (thread, input) => ({
481 ...thread,
482 pinOrderKey: input.orderKey,
483 })),
484 reorderActive: optimistic.wrap(commands.reorderActive, (thread, input) => ({
485 ...thread,
486 activeOrderKey: input.orderKey,
487 })),
488 };
489}