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