1import {
2 OrchestrationV2AppThread,
3 OrchestrationV2TurnItem,
4 type OrchestrationV2ThreadProjection,
5 type ScopedThreadRef,
6} from "@t3tools/contracts";
7import * as Option from "effect/Option";
8import * as Schema from "effect/Schema";
9import { AsyncResult, Atom } from "effect/reactivity";
10
11import {
12 deriveThreadQueueWorkflowState,
13 getUserQueuedThreadRuns,
14 type ThreadQueueWorkflowState,
15} from "./threadWorkflows.ts";
16import { deriveThreadTurnSubagents, type ThreadTurnSubagents } from "./threadSubagents.ts";
17import {
18 deriveUsageLimitNoticeRows,
19 presentRunStop,
20 withoutUsageLimitNoticeRows,
21 type RunStopPresentation,
22 type UsageLimitNoticeRows,
23} from "./threadExecution.ts";
24import {
25 deriveThreadCheckpointSummaries,
26 type ThreadCheckpointSummary,
27} from "./threadCheckpoints.ts";
28import type {
29 EnvironmentThread,
30 EnvironmentThreadStructure,
31 ThreadStructure,
32 ThreadStructureThread,
33 ThreadStructureTurnItem,
34 ThreadStructureVisibleTurnItem,
35} from "./models.ts";
36import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threadState.ts";
37import {
38 createQuestionHistoryProjector,
39 derivePendingThreadRequests,
40 type PendingThreadRequests,
41} from "./threadRequests.ts";
42import { arrayElementsEqual, parseThreadKey, threadKey } from "./entities.ts";
43import { getProjectionArrayReplacementIndex } from "./orchestrationV2Projection.ts";
44
45const EMPTY_VISIBLE_TURN_ITEMS: OrchestrationV2ThreadProjection["visibleTurnItems"] = Object.freeze(
46 [],
47);
48const EMPTY_THREAD_STRUCTURE_TURN_ITEMS: ReadonlyArray<ThreadStructureVisibleTurnItem> =
49 Object.freeze([]);
50const EMPTY_CHECKPOINT_SUMMARIES: ReadonlyArray<ThreadCheckpointSummary> = Object.freeze([]);
51
52/**
53 * Changes when the agent may have changed the workspace, so file views and host images refresh.
54 * File tools are explicit; commands, MCP or dynamic tools and subagents are included because
55 * they can mutate the workspace without reporting the paths they touched.
56 */
57export function deriveWorkspaceMutationId(
58 visibleTurnItems: ReadonlyArray<ThreadStructureVisibleTurnItem>,
59 latestCheckpointCompletedAt: string | null,
60) {
61 let itemId: string | null = null;
62 for (let index = visibleTurnItems.length - 1; index >= 0; index -= 1) {
63 const item = visibleTurnItems[index]?.item;
64 if (!item) continue;
65 if (
66 item.type !== "command_execution" &&
67 item.type !== "file_change" &&
68 item.type !== "dynamic_tool" &&
69 item.type !== "subagent"
70 ) {
71 continue;
72 }
73 if (item.status === "pending" || item.status === "running" || item.status === "waiting") {
74 continue;
75 }
76 itemId = item.id;
77 break;
78 }
79 return itemId === null && latestCheckpointCompletedAt === null
80 ? null
81 : JSON.stringify([itemId, latestCheckpointCompletedAt]);
82}
83
84type TurnItem = OrchestrationV2ThreadProjection["turnItems"][number];
85type VisibleTurnItem = OrchestrationV2ThreadProjection["visibleTurnItems"][number];
86type AppThread = OrchestrationV2ThreadProjection["thread"];
87
88// Items are stripped once each: unchanged items keep their identity across
89// structure changes, which the incremental timeline relies on.
90const structureTurnItems = new WeakMap<TurnItem, ThreadStructureTurnItem>();
91const structureVisibleTurnItems = new WeakMap<VisibleTurnItem, ThreadStructureVisibleTurnItem>();
92
93// Queued text, attachments and attribution all come from the message, so its identity covers them.
94function queueWorkflowStatesEqual(
95 left: ThreadQueueWorkflowState,
96 right: ThreadQueueWorkflowState,
97): boolean {
98 return (
99 left.activeRun === right.activeRun &&
100 left.isHeld === right.isHeld &&
101 left.canSendNow === right.canSendNow &&
102 left.canPromoteToSteer === right.canPromoteToSteer &&
103 left.promoteSkipsCompaction === right.promoteSkipsCompaction &&
104 left.queuedRuns.length === right.queuedRuns.length &&
105 left.queuedRuns.every(
106 (queued, index) =>
107 queued.run === right.queuedRuns[index]!.run &&
108 queued.message === right.queuedRuns[index]!.message,
109 )
110 );
111}
112
113function toThreadStructureTurnItem(item: TurnItem): ThreadStructureTurnItem {
114 if (item.type !== "assistant_message" && item.type !== "reasoning") return item;
115 let structure = structureTurnItems.get(item);
116 if (structure === undefined) {
117 const { text: _text, updatedAt: _updatedAt, ...rest } = item;
118 structure = rest;
119 structureTurnItems.set(item, structure);
120 }
121 return structure;
122}
123
124function toThreadStructureVisibleTurnItem(
125 visibleItem: VisibleTurnItem,
126): ThreadStructureVisibleTurnItem {
127 const item = toThreadStructureTurnItem(visibleItem.item);
128 if (item === visibleItem.item) return visibleItem;
129 let structure = structureVisibleTurnItems.get(visibleItem);
130 if (structure === undefined) {
131 structure = { ...visibleItem, item };
132 structureVisibleTurnItems.set(visibleItem, structure);
133 }
134 return structure;
135}
136
137/** Whether the structure kept every field of this row: true for all but streamed items. */
138export function isCompleteTurnItemRow(row: ThreadStructureVisibleTurnItem): row is VisibleTurnItem {
139 return row.item.type !== "assistant_message" && row.item.type !== "reasoning";
140}
141
142// The open thread is marked visited several times while a reply streams. Visit
143// state is read from the shell, or from lastVisitedAtAtom while the thread is
144// not in the shell index yet, so a mark is not structure.
145function toThreadStructureThread(thread: AppThread): ThreadStructureThread {
146 const { updatedAt: _updatedAt, lastVisitedAt: _lastVisitedAt, ...structure } = thread;
147 return structure;
148}
149
150function toThreadStructure(projection: OrchestrationV2ThreadProjection): ThreadStructure {
151 const { updatedAt: _updatedAt, ...structure } = projection;
152 return {
153 ...structure,
154 thread: toThreadStructureThread(projection.thread),
155 turnItems: projection.turnItems.map(toThreadStructureTurnItem),
156 visibleTurnItems: projection.visibleTurnItems.map(toThreadStructureVisibleTurnItem),
157 };
158}
159
160const sameTurnItem = Schema.toEquivalence(OrchestrationV2TurnItem);
161const sameAppThread = Schema.toEquivalence(OrchestrationV2AppThread);
162
163// Overlays previous's structure on next, so the fields the strip functions
164// remove are exactly the ones ignored here.
165function sameApartFromStripped<T extends object>(
166 same: (left: T, right: T) => boolean,
167 previous: T,
168 previousStructure: object,
169 next: T,
170): boolean {
171 return (
172 Object.keys(previous).length === Object.keys(next).length &&
173 same(Object.assign({}, next, previousStructure), next)
174 );
175}
176
177function sameTurnItemStructure(previous: TurnItem, next: TurnItem): boolean {
178 if (previous === next) return true;
179 const structure = toThreadStructureTurnItem(previous);
180 if (previous.id !== next.id || previous.type !== next.type || structure === previous) {
181 return false;
182 }
183 return sameApartFromStripped(sameTurnItem, previous, structure, next);
184}
185
186function sameProjectionArray<T>(
187 previous: ReadonlyArray<T>,
188 next: ReadonlyArray<T>,
189 same: (previous: T, next: T) => boolean,
190): boolean {
191 if (previous === next) return true;
192 if (previous.length !== next.length) return false;
193 const replacedIndex = getProjectionArrayReplacementIndex(previous, next);
194 if (replacedIndex !== undefined) return same(previous[replacedIndex]!, next[replacedIndex]!);
195 return previous.every((item, index) => same(item, next[index]!));
196}
197
198function sameVisibleTurnItemStructure(before: VisibleTurnItem, after: VisibleTurnItem): boolean {
199 return (
200 before === after ||
201 (before.position === after.position &&
202 before.visibility === after.visibility &&
203 before.sourceThreadId === after.sourceThreadId &&
204 before.sourceItemId === after.sourceItemId &&
205 sameTurnItemStructure(before.item, after.item))
206 );
207}
208
209function isStreamingAssistantMessage(visibleItem: VisibleTurnItem): boolean {
210 return visibleItem.item.type === "assistant_message" && visibleItem.item.streaming;
211}
212
213/** Whether `next` only streams more text into assistant messages that were already streaming. */
214function sameApartFromStreamedAssistantText(
215 previous: OrchestrationV2ThreadProjection["visibleTurnItems"],
216 next: OrchestrationV2ThreadProjection["visibleTurnItems"],
217): boolean {
218 return sameProjectionArray(
219 previous,
220 next,
221 (before, after) =>
222 before === after ||
223 (isStreamingAssistantMessage(before) &&
224 isStreamingAssistantMessage(after) &&
225 sameVisibleTurnItemStructure(before, after)),
226 );
227}
228
229/**
230 * Whether `previous` and `next` have the same ThreadStructure: they differ
231 * only by text streamed into existing assistant and reasoning items or by a
232 * visit mark. The same items exist, with the same status, and every other
233 * field is unchanged.
234 */
235export function sameThreadStructure(
236 previous: OrchestrationV2ThreadProjection,
237 next: OrchestrationV2ThreadProjection,
238): boolean {
239 if (previous === next) return true;
240 for (const key of Object.keys(next) as Array<keyof OrchestrationV2ThreadProjection>) {
241 if (
242 key === "thread" ||
243 key === "turnItems" ||
244 key === "visibleTurnItems" ||
245 key === "updatedAt"
246 ) {
247 continue;
248 }
249 if (previous[key] !== next[key]) return false;
250 }
251 if (
252 previous.thread !== next.thread &&
253 !sameApartFromStripped(
254 sameAppThread,
255 previous.thread,
256 toThreadStructureThread(previous.thread),
257 next.thread,
258 )
259 ) {
260 return false;
261 }
262 return (
263 sameProjectionArray(previous.turnItems, next.turnItems, sameTurnItemStructure) &&
264 sameProjectionArray(
265 previous.visibleTurnItems,
266 next.visibleTurnItems,
267 sameVisibleTurnItemStructure,
268 )
269 );
270}
271
272export function createEnvironmentThreadDetailAtoms<E>(
273 threadStateAtom: (
274 environmentId: ScopedThreadRef["environmentId"],
275 threadId: ScopedThreadRef["threadId"],
276 ) => Atom.Atom<AsyncResult.AsyncResult<EnvironmentThreadState, E>>,
277) {
278 const threadStateValueAtomFamily = Atom.family((key: string) => {
279 const ref = parseThreadKey(key);
280 return Atom.make((get) =>
281 AsyncResult.getOrElse(
282 get(threadStateAtom(ref.environmentId, ref.threadId)),
283 () => EMPTY_ENVIRONMENT_THREAD_STATE,
284 ),
285 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-state-value:${key}`));
286 });
287
288 const threadAtomFamily = Atom.family((key: string) => {
289 const ref = parseThreadKey(key);
290 let previousProjection: OrchestrationV2ThreadProjection | null = null;
291 let previousValue: EnvironmentThread | null = null;
292 return Atom.make((get) => {
293 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
294 if (projection === previousProjection) return previousValue;
295 previousProjection = projection;
296 previousValue = projection === null ? null : { environmentId: ref.environmentId, projection };
297 return previousValue;
298 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread:${key}`));
299 });
300
301 // Streamed assistant and reasoning text replaces an item several times a
302 // second. Readers of the thread's structure (runs, requests, which items
303 // exist and their status) keep the previous value through those updates.
304 const quietThreadAtomFamily = Atom.family((key: string) => {
305 let previousProjection: OrchestrationV2ThreadProjection | null = null;
306 let previous: EnvironmentThreadStructure | null = null;
307 return Atom.make((get) => {
308 const next = get(threadAtomFamily(key));
309 if (next === null) {
310 previousProjection = null;
311 previous = null;
312 } else if (
313 previousProjection === null ||
314 !sameThreadStructure(previousProjection, next.projection)
315 ) {
316 previous = {
317 environmentId: next.environmentId,
318 projection: toThreadStructure(next.projection),
319 };
320 }
321 // Comparing the latest projection lets consecutive replacements bypass history scans.
322 previousProjection = next?.projection ?? null;
323 return previous;
324 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-quiet-thread:${key}`));
325 });
326
327 // Read from the structure so streamed text does not rescan runs and turn items.
328 const usageLimitNoticeRowsAtomFamily = Atom.family((key: string) => {
329 let previous: EnvironmentThreadStructure | null = null;
330 let value: UsageLimitNoticeRows | null = null;
331 return Atom.make((get) => {
332 const thread = get(quietThreadAtomFamily(key));
333 if (thread !== previous) {
334 previous = thread;
335 value = thread === null ? null : deriveUsageLimitNoticeRows(thread.projection);
336 }
337 return value;
338 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-usage-limit-notice:${key}`));
339 });
340
341 // Presentation only: the projection keeps every row for other readers.
342 const visibleTurnItemsAtomFamily = Atom.family((key: string) => {
343 const projectQuestionHistory = createQuestionHistoryProjector();
344 let previousRows: OrchestrationV2ThreadProjection["visibleTurnItems"] | null = null;
345 let previousNotice: UsageLimitNoticeRows | null = null;
346 let previous = EMPTY_VISIBLE_TURN_ITEMS;
347 return Atom.make((get): OrchestrationV2ThreadProjection["visibleTurnItems"] => {
348 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
349 if (projection === null) return EMPTY_VISIBLE_TURN_ITEMS;
350 const rows = projectQuestionHistory(projection);
351 const notice = get(usageLimitNoticeRowsAtomFamily(key));
352 if (rows !== previousRows || notice !== previousNotice) {
353 previousRows = rows;
354 previousNotice = notice;
355 previous = withoutUsageLimitNoticeRows(rows, notice);
356 }
357 return previous;
358 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-visible-turn-items:${key}`));
359 });
360
361 const quietVisibleTurnItemsAtomFamily = Atom.family((key: string) => {
362 const projectQuestionHistory = createQuestionHistoryProjector<ThreadStructureVisibleTurnItem>();
363 return Atom.make((get) => {
364 const thread = get(quietThreadAtomFamily(key));
365 return thread === null
366 ? EMPTY_THREAD_STRUCTURE_TURN_ITEMS
367 : projectQuestionHistory(thread.projection);
368 }).pipe(
369 Atom.setIdleTTL(0),
370 Atom.withLabel(`environment-quiet-thread-visible-turn-items:${key}`),
371 );
372 });
373
374 // Holds while only streamed assistant text moves, so the timeline re-renders on structure;
375 // each streaming message reads its text through assistantMessageTextAtom.
376 const timelineVisibleTurnItemsAtomFamily = Atom.family((key: string) => {
377 let previous = EMPTY_VISIBLE_TURN_ITEMS;
378 return Atom.make((get) => {
379 const next = get(visibleTurnItemsAtomFamily(key));
380 if (!sameApartFromStreamedAssistantText(previous, next)) previous = next;
381 return previous;
382 }).pipe(
383 Atom.setIdleTTL(0),
384 Atom.withLabel(`environment-thread-timeline-visible-turn-items:${key}`),
385 );
386 });
387
388 const assistantMessageTextAtomFamily = Atom.family((key: string) => {
389 const separator = key.lastIndexOf("\u0000");
390 const thread = key.slice(0, separator);
391 const itemId = key.slice(separator + 1);
392 return Atom.make((get) => {
393 const item = get(visibleTurnItemsAtomFamily(thread)).find(
394 (visibleItem) => visibleItem.item.id === itemId,
395 )?.item;
396 return item?.type === "assistant_message" ? item.text : null;
397 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-assistant-message-text:${key}`));
398 });
399
400 const queueWorkflowAtomFamily = Atom.family((key: string) => {
401 let previous: Pick<
402 OrchestrationV2ThreadProjection,
403 "thread" | "runs" | "messages" | "providerThreads" | "providerSessions" | "providerTurns"
404 > | null = null;
405 let value: ThreadQueueWorkflowState | null = null;
406 return Atom.make((get) => {
407 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
408 if (projection === null) {
409 previous = null;
410 value = null;
411 } else if (
412 previous === null ||
413 projection.thread !== previous.thread ||
414 projection.runs !== previous.runs ||
415 projection.messages !== previous.messages ||
416 projection.providerThreads !== previous.providerThreads ||
417 projection.providerSessions !== previous.providerSessions ||
418 projection.providerTurns !== previous.providerTurns
419 ) {
420 const next = deriveThreadQueueWorkflowState(projection);
421 if (value === null || !queueWorkflowStatesEqual(value, next)) value = next;
422 const { thread, runs, messages, providerThreads, providerSessions, providerTurns } =
423 projection;
424 previous = { thread, runs, messages, providerThreads, providerSessions, providerTurns };
425 }
426 return value;
427 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-queue:${key}`));
428 });
429 const runStopAtomFamily = Atom.family((key: string) => {
430 let previous: Pick<OrchestrationV2ThreadProjection, "runs" | "turnItems"> | null = null;
431 let stop: RunStopPresentation | null = null;
432 return Atom.make((get) => {
433 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
434 if (projection === null) {
435 previous = null;
436 stop = null;
437 } else if (
438 projection.runs !== previous?.runs ||
439 projection.turnItems !== previous?.turnItems
440 ) {
441 const { runs, turnItems } = projection;
442 previous = { runs, turnItems };
443 const next = presentRunStop(previous);
444 if (
445 next?.state !== stop?.state ||
446 next?.runId !== stop?.runId ||
447 next?.requestedAt !== stop?.requestedAt ||
448 next?.background !== stop?.background
449 )
450 stop = next;
451 }
452 return stop;
453 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-run-stop:${key}`));
454 });
455 const queuedCountAtomFamily = Atom.family((key: string) => {
456 let previous: Pick<OrchestrationV2ThreadProjection, "runs" | "messages"> | null = null;
457 let count = 0;
458 return Atom.make((get) => {
459 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
460 if (projection === null) {
461 previous = null;
462 count = 0;
463 } else if (projection.runs !== previous?.runs || projection.messages !== previous?.messages) {
464 const { runs, messages } = projection;
465 previous = { runs, messages };
466 count = getUserQueuedThreadRuns(previous).length;
467 }
468 return count;
469 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-queue-count:${key}`));
470 });
471
472 const turnSubagentsAtomFamily = Atom.family((key: string) => {
473 let previous: Pick<OrchestrationV2ThreadProjection, "runs" | "subagents"> | null = null;
474 let value: ThreadTurnSubagents | null = null;
475 return Atom.make((get) => {
476 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
477 if (projection === null) {
478 previous = null;
479 value = null;
480 } else if (
481 projection.runs !== previous?.runs ||
482 projection.subagents !== previous?.subagents
483 ) {
484 const { runs, subagents } = projection;
485 previous = { runs, subagents };
486 value = deriveThreadTurnSubagents(previous);
487 }
488 return value;
489 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-turn-subagents:${key}`));
490 });
491
492 const checkpointSummariesAtomFamily = Atom.family((key: string) => {
493 let previous: Pick<OrchestrationV2ThreadProjection, "checkpoints" | "messages"> | null = null;
494 let value = EMPTY_CHECKPOINT_SUMMARIES;
495 return Atom.make((get) => {
496 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
497 if (projection === null) {
498 previous = null;
499 value = EMPTY_CHECKPOINT_SUMMARIES;
500 } else if (
501 projection.checkpoints !== previous?.checkpoints ||
502 projection.messages !== previous?.messages
503 ) {
504 const { checkpoints, messages } = projection;
505 const next = deriveThreadCheckpointSummaries({ checkpoints, messages });
506 // The same checkpoints give the same summaries; messages only pick each assistant message.
507 if (
508 checkpoints !== previous?.checkpoints ||
509 next.some(
510 (summary, index) => summary.assistantMessageId !== value[index]?.assistantMessageId,
511 )
512 )
513 value = next;
514 previous = { checkpoints, messages };
515 }
516 return value;
517 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-checkpoint-summaries:${key}`));
518 });
519
520 const lastVisitedAtAtomFamily = Atom.family((key: string) =>
521 Atom.make(
522 (get) =>
523 Option.getOrNull(get(threadStateValueAtomFamily(key)).data)?.thread.lastVisitedAt ?? null,
524 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-last-visited:${key}`)),
525 );
526 const updatedAtAtomFamily = Atom.family((key: string) =>
527 Atom.make(
528 (get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).data)?.updatedAt ?? null,
529 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-updated-at:${key}`)),
530 );
531 const worktreePathAtomFamily = Atom.family((key: string) =>
532 Atom.make(
533 (get) =>
534 Option.getOrNull(get(threadStateValueAtomFamily(key)).data)?.thread.worktreePath ?? null,
535 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-worktree:${key}`)),
536 );
537 const pendingRequestsAtomFamily = Atom.family((key: string) => {
538 let previous: Pick<OrchestrationV2ThreadProjection, "runtimeRequests" | "turnItems"> | null =
539 null;
540 let value: PendingThreadRequests | null = null;
541 return Atom.make((get) => {
542 const projection = Option.getOrNull(get(threadStateValueAtomFamily(key)).data);
543 if (projection === null) {
544 previous = null;
545 value = null;
546 return value;
547 }
548 const runtimeRequests = projection.runtimeRequests.filter(
549 (request) => request.status === "pending",
550 );
551 const ids = new Set(runtimeRequests.map((request) => request.id));
552 const turnItems =
553 ids.size === 0
554 ? []
555 : projection.turnItems.filter(
556 (item) =>
557 (item.type === "approval_request" || item.type === "user_input_request") &&
558 ids.has(item.requestId),
559 );
560 if (
561 previous === null ||
562 !arrayElementsEqual(previous.runtimeRequests, runtimeRequests) ||
563 !arrayElementsEqual(previous.turnItems, turnItems)
564 ) {
565 previous = { runtimeRequests, turnItems };
566 value = derivePendingThreadRequests(previous);
567 }
568 return value;
569 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-pending-requests:${key}`));
570 });
571
572 const statusAtomFamily = Atom.family((key: string) =>
573 Atom.make((get) => get(threadStateValueAtomFamily(key)).status).pipe(
574 Atom.setIdleTTL(0),
575 Atom.withLabel(`environment-thread-status:${key}`),
576 ),
577 );
578 const sequenceAtomFamily = Atom.family((key: string) =>
579 Atom.make((get) => get(threadStateValueAtomFamily(key)).sequence).pipe(
580 Atom.setIdleTTL(0),
581 Atom.withLabel(`environment-thread-sequence:${key}`),
582 ),
583 );
584 const errorAtomFamily = Atom.family((key: string) =>
585 Atom.make((get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).error)).pipe(
586 Atom.setIdleTTL(0),
587 Atom.withLabel(`environment-thread-error:${key}`),
588 ),
589 );
590 const historyAtomFamily = Atom.family((key: string) =>
591 Atom.make((get) => get(threadStateValueAtomFamily(key)).history).pipe(
592 Atom.setIdleTTL(0),
593 Atom.withLabel(`environment-thread-history:${key}`),
594 ),
595 );
596
597 const workspaceMutationIdAtomFamily = Atom.family((key: string) =>
598 Atom.make((get) =>
599 deriveWorkspaceMutationId(
600 get(quietVisibleTurnItemsAtomFamily(key)),
601 get(checkpointSummariesAtomFamily(key)).at(-1)?.completedAt ?? null,
602 ),
603 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-workspace-mutation-id:${key}`)),
604 );
605
606 return {
607 worktreePathAtom: (ref: ScopedThreadRef) => worktreePathAtomFamily(threadKey(ref)),
608 workspaceMutationIdAtom: (ref: ScopedThreadRef) =>
609 workspaceMutationIdAtomFamily(threadKey(ref)),
610 updatedAtAtom: (ref: ScopedThreadRef) => updatedAtAtomFamily(threadKey(ref)),
611 lastVisitedAtAtom: (ref: ScopedThreadRef) => lastVisitedAtAtomFamily(threadKey(ref)),
612 pendingRequestsAtom: (ref: ScopedThreadRef) => pendingRequestsAtomFamily(threadKey(ref)),
613 queueWorkflowAtom: (ref: ScopedThreadRef) => queueWorkflowAtomFamily(threadKey(ref)),
614 queuedCountAtom: (ref: ScopedThreadRef) => queuedCountAtomFamily(threadKey(ref)),
615 runStopAtom: (ref: ScopedThreadRef) => runStopAtomFamily(threadKey(ref)),
616 turnSubagentsAtom: (ref: ScopedThreadRef) => turnSubagentsAtomFamily(threadKey(ref)),
617 checkpointSummariesAtom: (ref: ScopedThreadRef) =>
618 checkpointSummariesAtomFamily(threadKey(ref)),
619 stateAtom: (ref: ScopedThreadRef) => threadStateValueAtomFamily(threadKey(ref)),
620 threadAtom: (ref: ScopedThreadRef) => threadAtomFamily(threadKey(ref)),
621 quietThreadAtom: (ref: ScopedThreadRef) => quietThreadAtomFamily(threadKey(ref)),
622 visibleTurnItemsAtom: (ref: ScopedThreadRef) => visibleTurnItemsAtomFamily(threadKey(ref)),
623 quietVisibleTurnItemsAtom: (ref: ScopedThreadRef) =>
624 quietVisibleTurnItemsAtomFamily(threadKey(ref)),
625 timelineVisibleTurnItemsAtom: (ref: ScopedThreadRef) =>
626 timelineVisibleTurnItemsAtomFamily(threadKey(ref)),
627 assistantMessageTextAtom: (ref: ScopedThreadRef, itemId: string) =>
628 assistantMessageTextAtomFamily(`${threadKey(ref)}\u0000${itemId}`),
629 statusAtom: (ref: ScopedThreadRef) => statusAtomFamily(threadKey(ref)),
630 sequenceAtom: (ref: ScopedThreadRef) => sequenceAtomFamily(threadKey(ref)),
631 errorAtom: (ref: ScopedThreadRef) => errorAtomFamily(threadKey(ref)),
632 historyAtom: (ref: ScopedThreadRef) => historyAtomFamily(threadKey(ref)),
633 };
634}