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