1import {
2 isOrchestrationV2WorkActive,
3 ORCHESTRATION_V2_WS_METHODS,
4 type EnvironmentId as EnvironmentIdType,
5 type OrchestrationV2ThreadDetailSnapshot,
6 type OrchestrationV2ThreadProjection,
7 type OrchestrationV2ThreadStreamItem,
8 type OrchestrationV2TurnItem,
9 type ThreadId as ThreadIdType,
10} from "@t3tools/contracts";
11import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
12import { pendingBackgroundTurnItems } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
13import * as Cause from "effect/Cause";
14import * as Effect from "effect/Effect";
15import * as Fiber from "effect/Fiber";
16import * as Scope from "effect/Scope";
17import * as Option from "effect/Option";
18import * as Queue from "effect/Queue";
19import * as Ref from "effect/Ref";
20import * as Result from "effect/Result";
21import * as Semaphore from "effect/Semaphore";
22import * as Stream from "effect/Stream";
23import * as SubscriptionRef from "effect/SubscriptionRef";
24import { HttpClient } from "effect/http";
25import { Atom } from "effect/reactivity";
26
27import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
28import * as EnvironmentRegistry from "../connection/registry.ts";
29import { connectionProjectionPhase } from "../connection/model.ts";
30import * as EnvironmentSupervisor from "../connection/supervisor.ts";
31import * as ConnectionWakeups from "../connection/wakeups.ts";
32import * as Persistence from "../platform/persistence.ts";
33import { runCachePersistence } from "./cachePersistence.ts";
34import * as ManagedRelay from "../relay/managedRelay.ts";
35import {
36 isAuthorizationFailure,
37 request,
38 subscribeDynamic,
39 subscriptionRetryDelay,
40} from "../rpc/client.ts";
41import { parseThreadKey, threadKey } from "./entities.ts";
42import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts";
43import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts";
44import { followStreamInEnvironment } from "./runtime.ts";
45import * as ThreadHistoryController from "./threadHistoryController.ts";
46import { fetchEnvironmentThreadHistoryPage } from "./threadHistoryHttp.ts";
47import {
48 applyHistoryPageMeta,
49 clearActiveHistoryLoading,
50 EMPTY_THREAD_HISTORY_META,
51 isActiveHistoryRequestCursor,
52 mergeOlderHistoryIntoProjection,
53 type ThreadHistoryMeta,
54} from "./threadHistoryMerge.ts";
55import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts";
56import { makeThreadRequestsResolvedAnnouncer } from "./threadRequests.ts";
57import {
58 EMPTY_ENVIRONMENT_THREAD_STATE,
59 type EnvironmentThreadState,
60 type EnvironmentThreadStatus,
61} from "./threadState.ts";
62
63const THREAD_SUBSCRIPTION_RETRY_DELAY = "250 millis";
64
65function statusWithoutLiveData(
66 data: Option.Option<OrchestrationV2ThreadProjection>,
67): EnvironmentThreadStatus {
68 return Option.isSome(data) ? "cached" : "empty";
69}
70
71function formatThreadError(cause: Cause.Cause<unknown>): string {
72 const error = Cause.squash(cause);
73 return error instanceof Error && error.message.trim().length > 0
74 ? error.message
75 : "Could not synchronize the thread.";
76}
77
78function formatHistoryError(error: unknown): string {
79 if (error instanceof Error && error.message.trim().length > 0) {
80 return error.message;
81 }
82 return "Could not load earlier activity.";
83}
84
85// Steers join a running turn; every other user message opens one. The timeline shows every
86// notification as a response boundary, whether it opened its run or arrived mid-run.
87function startsTurn(item: OrchestrationV2TurnItem): boolean {
88 return (
89 item.type === "notification" ||
90 (item.type === "user_message" &&
91 item.inputIntent !== "steer" &&
92 item.inputIntent !== "promoted_queued_to_steer")
93 );
94}
95
96function historyMetaFromCachedSnapshot(
97 snapshot: OrchestrationV2ThreadDetailSnapshot,
98): ThreadHistoryMeta {
99 const historyCursor = snapshot.historyCursor ?? null;
100 const hasMoreHistory = snapshot.hasMoreHistory ?? false;
101 return {
102 historyCursor,
103 hasMoreHistory,
104 loading: false,
105 error: null,
106 expanded: false,
107 latestLocalTurnOrdinal: snapshot.latestLocalTurnOrdinal ?? null,
108 };
109}
110
111function snapshotToPersist(
112 snapshotSequence: number,
113 projection: OrchestrationV2ThreadProjection,
114 history: ThreadHistoryMeta,
115 acceptsBoundedSnapshots: boolean,
116): OrchestrationV2ThreadDetailSnapshot {
117 // A complete bounded snapshot still proves paging support. Retain that evidence
118 // so a thread that grows while closed can resume with a bounded fallback.
119 if (acceptsBoundedSnapshots || history.hasMoreHistory || history.historyCursor !== null) {
120 return {
121 snapshotSequence,
122 projection,
123 historyCursor: history.historyCursor,
124 hasMoreHistory: history.hasMoreHistory,
125 latestLocalTurnOrdinal: history.latestLocalTurnOrdinal,
126 };
127 }
128 return { snapshotSequence, projection };
129}
130
131function shouldPersistThread(
132 thread: OrchestrationV2ThreadProjection,
133 history: ThreadHistoryMeta,
134): boolean {
135 // After the user loads older pages the in-memory timeline can grow large.
136 // Keep those expanded projections out of the monolithic cache.
137 if (history.expanded) {
138 return false;
139 }
140 return !thread.runs.some(
141 (run) => run.status === "preparing" || run.status === "starting" || run.status === "running",
142 );
143}
144
145// Only still-active local copies take the server's item: events applied meanwhile win,
146// and inherited rows belong to another thread.
147function settleTurnItems(
148 threadId: ThreadIdType,
149 projection: OrchestrationV2ThreadProjection,
150 settled: ReadonlyMap<string, OrchestrationV2TurnItem>,
151): OrchestrationV2ThreadProjection {
152 let changed = false;
153 const settle = (item: OrchestrationV2TurnItem, id: string) => {
154 const server = settled.get(id);
155 if (server === undefined || !isOrchestrationV2WorkActive(item.status)) return item;
156 changed = true;
157 return server;
158 };
159 const turnItems = projection.turnItems.map((item) => settle(item, item.id));
160 const visibleTurnItems = projection.visibleTurnItems.map((row) => {
161 if (row.sourceThreadId !== threadId) return row;
162 const item = settle(row.item, row.sourceItemId);
163 return item === row.item ? row : { ...row, item };
164 });
165 return changed ? { ...projection, turnItems, visibleTurnItems } : projection;
166}
167
168interface ThreadResumeSnapshot {
169 readonly state: EnvironmentThreadState;
170 readonly sequence: number;
171 readonly persisted: boolean;
172 readonly acceptsBoundedSnapshots?: boolean;
173}
174
175interface ThreadResumeCache {
176 snapshot: ThreadResumeSnapshot | undefined;
177 owner: object | undefined;
178}
179
180function matchesThreadSnapshot(
181 current: ThreadResumeSnapshot,
182 projection: OrchestrationV2ThreadProjection | null,
183 sequence: number,
184 history: Pick<ThreadHistoryMeta, "historyCursor" | "hasMoreHistory" | "latestLocalTurnOrdinal">,
185): boolean {
186 if (current.sequence !== sequence || Option.getOrNull(current.state.data) !== projection)
187 return false;
188 const currentHistory = current.state.history;
189 return (
190 currentHistory.historyCursor === history.historyCursor &&
191 currentHistory.hasMoreHistory === history.hasMoreHistory &&
192 (!(history.hasMoreHistory || history.historyCursor !== null) ||
193 currentHistory.latestLocalTurnOrdinal === history.latestLocalTurnOrdinal)
194 );
195}
196
197function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadState {
198 return {
199 ...value,
200 status:
201 value.status === "deleted" || (value.status === "live" && Option.isSome(value.data))
202 ? value.status
203 : statusWithoutLiveData(value.data),
204 error: Option.none(),
205 history:
206 value.history.loading || value.history.error !== null
207 ? { ...value.history, loading: false, error: null }
208 : value.history,
209 };
210}
211
212function hasMissingPendingQuestionCard(projection: OrchestrationV2ThreadProjection): boolean {
213 const pending = new Set(
214 projection.runtimeRequests
215 .filter((request) => request.kind === "user_input" && request.status === "pending")
216 .map((request) => request.id),
217 );
218 if (pending.size === 0) return false;
219 for (const item of projection.turnItems) {
220 if (item.type === "user_input_request") pending.delete(item.requestId);
221 if (pending.size === 0) return false;
222 }
223 return true;
224}
225
226export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* (
227 threadId: ThreadIdType,
228 resumeCache?: ThreadResumeCache,
229) {
230 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
231 const cache = yield* Persistence.EnvironmentCacheStore;
232 const snapshotLoader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader;
233 const historyController = yield* Effect.serviceOption(
234 ThreadHistoryController.ThreadHistoryController,
235 );
236 const httpClient = yield* Effect.serviceOption(HttpClient.HttpClient);
237 const dpopSigner = yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner);
238 const remoteAuthorization = yield* Effect.serviceOption(
239 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization,
240 );
241 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);
242 const environmentId = supervisor.target.environmentId;
243 const candidate = resumeCache?.snapshot;
244 const retained =
245 candidate &&
246 Option.isSome(candidate.state.data) &&
247 hasMissingPendingQuestionCard(candidate.state.data.value)
248 ? undefined
249 : candidate;
250 const owner = {};
251 if (resumeCache) resumeCache.owner = owner;
252 const loaded =
253 retained === undefined
254 ? yield* cache.loadThread(environmentId, threadId).pipe(
255 Effect.catch((error) =>
256 Effect.logWarning("Could not load cached thread.").pipe(
257 Effect.annotateLogs({
258 environmentId,
259 threadId,
260 error: error.message,
261 }),
262 Effect.as(Option.none<OrchestrationV2ThreadDetailSnapshot>()),
263 ),
264 ),
265 )
266 : Option.none<OrchestrationV2ThreadDetailSnapshot>();
267 // Old bounded caches can omit a pending question's card; replay alone cannot restore it.
268 const cached = Option.filter(
269 loaded,
270 (snapshot) => !hasMissingPendingQuestionCard(snapshot.projection),
271 );
272 const cachedThread = Option.map(cached, (snapshot) => snapshot.projection);
273 // Seed the resume cursor from the cached snapshot so a warm cache can catch up
274 // via `afterSequence` instead of re-downloading the full thread body.
275 const initialSequence =
276 retained?.sequence ??
277 Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence });
278 const initialState: EnvironmentThreadState = retained
279 ? cachedThreadState(retained.state)
280 : {
281 data: cachedThread,
282 status: statusWithoutLiveData(cachedThread),
283 error: Option.none(),
284 history: Option.match(cached, {
285 onNone: () => EMPTY_THREAD_HISTORY_META,
286 onSome: historyMetaFromCachedSnapshot,
287 }),
288 sequence: initialSequence,
289 };
290 const state = yield* SubscriptionRef.make(initialState);
291 const canLoadHistory = Option.isSome(httpClient) && Option.isSome(historyController);
292 const acceptsBoundedSocketSnapshots = yield* Ref.make(canLoadHistory);
293 const lastSequence = yield* SubscriptionRef.make(initialSequence);
294 let committed: ThreadResumeSnapshot = {
295 state: initialState,
296 sequence: initialSequence,
297 persisted: retained?.persisted ?? Option.isSome(cached),
298 acceptsBoundedSnapshots: canLoadHistory,
299 };
300 if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
301 const awaitingCompletion = yield* Ref.make(false);
302 const applyLock = yield* Semaphore.make(1);
303 // Save only completed data/cursor updates. A canceled scope must not cache
304 // a cursor whose event has not reached the data yet.
305 const announceRequests = makeThreadRequestsResolvedAnnouncer(environmentId, threadId);
306 const remember = Effect.gen(function* () {
307 const current = yield* SubscriptionRef.get(state);
308 const sequence = yield* SubscriptionRef.get(lastSequence);
309 // Only a synchronized projection from the scope that owns the thread is
310 // authoritative: a cached one can predate a request the user already has
311 // drafts for, and a superseded scope can lag its successor.
312 const owns = resumeCache === undefined || resumeCache.owner === owner;
313 if (owns && current.status === "deleted") announceRequests([]);
314 else if (owns && current.status === "live" && Option.isSome(current.data))
315 announceRequests(current.data.value.runtimeRequests);
316 committed = {
317 state: current,
318 sequence,
319 acceptsBoundedSnapshots: yield* Ref.get(acceptsBoundedSocketSnapshots),
320 persisted:
321 committed.persisted &&
322 matchesThreadSnapshot(committed, Option.getOrNull(current.data), sequence, current.history),
323 };
324 if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
325 });
326 const persistence = yield* Queue.sliding<OrchestrationV2ThreadDetailSnapshot>(1);
327 let queuedSnapshot = Option.getOrUndefined(cached);
328 let savedSnapshot = queuedSnapshot;
329 const queuePersist = (snapshot: OrchestrationV2ThreadDetailSnapshot) =>
330 Effect.suspend(() => {
331 queuedSnapshot = snapshot;
332 return Queue.offer(persistence, snapshot);
333 });
334
335 const persist = Effect.fn("EnvironmentThreadState.persist")(function* (
336 snapshot: OrchestrationV2ThreadDetailSnapshot,
337 ) {
338 if (resumeCache !== undefined && resumeCache.owner !== owner) return;
339 if (committed.state.status === "deleted") return;
340 if (
341 committed.persisted &&
342 matchesThreadSnapshot(
343 committed,
344 snapshot.projection,
345 snapshot.snapshotSequence,
346 historyMetaFromCachedSnapshot(snapshot),
347 )
348 )
349 return;
350 yield* cache.saveThread(environmentId, snapshot).pipe(
351 Effect.tap(() =>
352 Effect.sync(() => {
353 savedSnapshot = snapshot;
354 if (
355 !matchesThreadSnapshot(
356 committed,
357 snapshot.projection,
358 snapshot.snapshotSequence,
359 historyMetaFromCachedSnapshot(snapshot),
360 )
361 )
362 return;
363 committed = { ...committed, persisted: true };
364 if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
365 }),
366 ),
367 Effect.catch((error) =>
368 Effect.logWarning("Could not persist the thread cache.").pipe(
369 Effect.annotateLogs({
370 environmentId,
371 threadId,
372 error: error.message,
373 }),
374 ),
375 ),
376 );
377 });
378
379 yield* Effect.addFinalizer(() =>
380 Effect.suspend(() => {
381 const { state: current, sequence: snapshotSequence } = committed;
382 return Option.match(current.data, {
383 onNone: () => Effect.void,
384 onSome: (projection) =>
385 shouldPersistThread(projection, current.history)
386 ? persist(
387 snapshotToPersist(
388 snapshotSequence,
389 projection,
390 current.history,
391 committed.acceptsBoundedSnapshots === true,
392 ),
393 )
394 : // The consumer may still be waiting out its delay; an expanded thread queues
395 // only cache repairs, which must survive leaving the thread.
396 queuedSnapshot !== undefined && queuedSnapshot !== savedSnapshot
397 ? persist(queuedSnapshot)
398 : Effect.void,
399 });
400 }),
401 );
402
403 yield* runCachePersistence(persistence, persist).pipe(Effect.forkScoped);
404
405 const setConnecting = SubscriptionRef.update(state, (current) =>
406 current.status === "deleted" || Option.isSome(current.error)
407 ? current
408 : {
409 ...current,
410 status: "synchronizing" as const,
411 error: Option.none(),
412 },
413 );
414 const setReady = SubscriptionRef.update(state, (current) =>
415 current.status === "live" || current.status === "deleted" || Option.isSome(current.error)
416 ? current
417 : {
418 ...current,
419 status: "synchronizing" as const,
420 error: Option.none(),
421 },
422 );
423 const setDisconnected = Effect.gen(function* () {
424 yield* Ref.set(awaitingCompletion, false);
425 yield* SubscriptionRef.update(state, (current) => ({
426 ...current,
427 status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
428 }));
429 });
430 const setStreamError = (message: string) =>
431 Ref.set(awaitingCompletion, false).pipe(
432 Effect.andThen(
433 SubscriptionRef.update(state, (current) => ({
434 ...current,
435 status:
436 current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
437 error: Option.some(message),
438 })),
439 ),
440 );
441
442 const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* (
443 thread: OrchestrationV2ThreadProjection,
444 options?: {
445 readonly resetHistory?: boolean;
446 readonly history?: ThreadHistoryMeta;
447 },
448 ) {
449 const waiting = yield* Ref.get(awaitingCompletion);
450 const sequence = yield* SubscriptionRef.get(lastSequence);
451 // Atomic with concurrent history meta updates: never get-then-set the whole
452 // state when only the projection changes. Bounded installs pass history so
453 // projection + cursor persist together in one enqueue.
454 const next = yield* SubscriptionRef.updateAndGet(state, (previous) => {
455 const history =
456 options?.history !== undefined
457 ? options.history
458 : options?.resetHistory === true
459 ? EMPTY_THREAD_HISTORY_META
460 : previous.history;
461 return {
462 ...previous,
463 data: Option.some(thread),
464 status: Option.isSome(previous.error)
465 ? ("cached" as const)
466 : waiting
467 ? ("synchronizing" as const)
468 : ("live" as const),
469 error: previous.error,
470 history,
471 sequence,
472 };
473 });
474 // Active projections can update many times per second and retain large tool
475 // payloads. Persist once the run settles so cache encoding stays off the
476 // streaming path. Progressive meta rides along when the window is incomplete.
477 if (shouldPersistThread(thread, next.history)) {
478 const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
479 yield* queuePersist(
480 snapshotToPersist(
481 snapshotSequence,
482 thread,
483 next.history,
484 yield* Ref.get(acceptsBoundedSocketSnapshots),
485 ),
486 );
487 }
488 });
489
490 const patchHistoryMeta = (patch: (history: ThreadHistoryMeta) => ThreadHistoryMeta) =>
491 SubscriptionRef.update(state, (current) => ({
492 ...current,
493 history: patch(current.history),
494 }));
495
496 const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () {
497 yield* Ref.set(awaitingCompletion, false);
498 yield* SubscriptionRef.set(state, {
499 data: Option.none(),
500 status: "deleted",
501 error: Option.none(),
502 history: EMPTY_THREAD_HISTORY_META,
503 sequence: yield* SubscriptionRef.get(lastSequence),
504 });
505 yield* remember;
506 if (resumeCache !== undefined && resumeCache.owner !== owner) return;
507 yield* cache.removeThread(environmentId, threadId).pipe(
508 Effect.catch((error) =>
509 Effect.logWarning("Could not remove the cached thread.").pipe(
510 Effect.annotateLogs({
511 environmentId,
512 threadId,
513 error: error.message,
514 }),
515 ),
516 ),
517 );
518 });
519
520 type EventItem = Extract<OrchestrationV2ThreadStreamItem, { kind: "event" }>;
521 type SequencedItem = Extract<
522 OrchestrationV2ThreadStreamItem,
523 { kind: "event" | "unknown-event" }
524 >;
525 // Per-batch streaming spans add trace volume without a useful timing boundary.
526 const applyEventsLocked = Effect.fnUntraced(function* (items: ReadonlyArray<SequencedItem>) {
527 const appliedSequence = yield* SubscriptionRef.get(lastSequence);
528 let sequence = appliedSequence;
529 const fresh: EventItem[] = [];
530 for (const item of items) {
531 if (item.sequence <= sequence) continue;
532 sequence = item.sequence;
533 if (item.kind === "event") {
534 fresh.push(item);
535 continue;
536 }
537 yield* Effect.logDebug("Skipped a thread event type this client does not know.").pipe(
538 Effect.annotateLogs({
539 environmentId,
540 threadId,
541 eventType: item.eventType.slice(0, 64),
542 sequence: item.sequence,
543 }),
544 );
545 }
546 if (sequence === appliedSequence) return;
547 yield* SubscriptionRef.set(lastSequence, sequence);
548 if (fresh.length === 0) return;
549
550 const waiting = yield* Ref.get(awaitingCompletion);
551 // Apply against the latest projection/history in one update so a concurrent
552 // loadEarlier merge (or history-meta patch) cannot be clobbered by a stale
553 // get-then-set rebuild.
554 type EventApplyResult =
555 | { readonly _tag: "noop" }
556 | { readonly _tag: "delete" }
557 | {
558 readonly _tag: "applied";
559 readonly projection: OrchestrationV2ThreadProjection;
560 readonly history: ThreadHistoryMeta;
561 };
562
563 const applyEvent = (
564 current: EnvironmentThreadState,
565 item: EventItem,
566 ): readonly [EventApplyResult, EnvironmentThreadState] => {
567 if (current.status === "deleted") {
568 return [{ _tag: "noop" }, current];
569 }
570 if (Option.isNone(current.data)) {
571 return [
572 item.event.type === "thread.deleted" ? { _tag: "delete" } : { _tag: "noop" },
573 current,
574 ];
575 }
576 if (item.event.type === "thread.deleted") {
577 return [{ _tag: "delete" }, current];
578 }
579
580 // Incomplete progressive windows only (hasMore or open cursor). Do not
581 // use expanded: it remains true after the last page as a cache marker.
582 const partial =
583 current.history.hasMoreHistory || current.history.historyCursor !== null
584 ? {
585 partialTimeline: true as const,
586 latestLocalTurnOrdinal: current.history.latestLocalTurnOrdinal,
587 }
588 : undefined;
589 const next = applyOrchestrationV2ProjectionEvent(current.data.value, item.event, partial);
590 if (next === null || next === current.data.value) {
591 return [{ _tag: "noop" }, current];
592 }
593
594 let history = current.history;
595 if (partial !== undefined && item.event.type === "turn-item.updated") {
596 const ordinal = item.event.payload.ordinal;
597 const watermark = history.latestLocalTurnOrdinal;
598 if (watermark === null || ordinal > watermark) {
599 history = { ...history, latestLocalTurnOrdinal: ordinal };
600 }
601 }
602
603 const updated: EnvironmentThreadState = {
604 ...current,
605 data: Option.some(next),
606 status: waiting ? "synchronizing" : "live",
607 error: Option.none(),
608 history,
609 };
610 return [{ _tag: "applied", projection: next, history: updated.history }, updated];
611 };
612
613 const result = yield* SubscriptionRef.modify(
614 state,
615 (current): readonly [EventApplyResult, EnvironmentThreadState] => {
616 let result: EventApplyResult = { _tag: "noop" };
617 for (const item of fresh) {
618 const [applied, next] = applyEvent(current, item);
619 current = next;
620 if (applied._tag !== "noop") result = applied;
621 if (applied._tag === "delete") break;
622 }
623 // A batch that changed nothing keeps its state object, so readers do not re-render.
624 return [result, result._tag === "noop" ? current : { ...current, sequence }];
625 },
626 );
627
628 if (result._tag === "delete") {
629 yield* setDeleted();
630 return;
631 }
632 if (result._tag === "applied" && shouldPersistThread(result.projection, result.history)) {
633 const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
634 yield* queuePersist(
635 snapshotToPersist(
636 snapshotSequence,
637 result.projection,
638 result.history,
639 yield* Ref.get(acceptsBoundedSocketSnapshots),
640 ),
641 );
642 }
643 });
644
645 const applyItemLocked = Effect.fn("EnvironmentThreadState.applyItemLocked")(function* (
646 item: OrchestrationV2ThreadStreamItem,
647 ) {
648 if (item.kind === "synchronized") {
649 yield* Ref.set(awaitingCompletion, false);
650 yield* SubscriptionRef.update(state, (current) =>
651 Option.isSome(current.data) && current.status !== "deleted" && Option.isNone(current.error)
652 ? { ...current, status: "live" as const, error: Option.none() }
653 : current,
654 );
655 return;
656 }
657
658 if (item.kind === "snapshot") {
659 yield* SubscriptionRef.set(lastSequence, item.snapshotSequence);
660 const hasProgressiveHistory =
661 item.historyCursor !== undefined ||
662 item.hasMoreHistory !== undefined ||
663 item.latestLocalTurnOrdinal !== undefined;
664 yield* setThread(
665 boundedSnapshotProjection(item),
666 hasProgressiveHistory
667 ? {
668 history: {
669 historyCursor: item.historyCursor ?? null,
670 hasMoreHistory: item.hasMoreHistory ?? false,
671 loading: false,
672 error: null,
673 expanded: false,
674 latestLocalTurnOrdinal: item.latestLocalTurnOrdinal ?? null,
675 },
676 }
677 : { resetHistory: true },
678 );
679 return;
680 }
681
682 yield* applyEventsLocked([item]);
683 });
684
685 // Keep this per-batch streaming path untraced for the same reason as applyEventsLocked.
686 const applyItems = Effect.fnUntraced(function* (
687 items: ReadonlyArray<OrchestrationV2ThreadStreamItem>,
688 ) {
689 yield* applyLock.withPermits(1)(
690 Effect.gen(function* () {
691 let events: SequencedItem[] = [];
692 for (const item of items) {
693 if (
694 item.kind === "unknown-event" ||
695 (item.kind === "event" && item.event.type !== "thread.deleted")
696 ) {
697 events.push(item);
698 continue;
699 }
700 yield* applyEventsLocked(events);
701 events = [];
702 yield* applyItemLocked(item);
703 }
704 yield* applyEventsLocked(events);
705 yield* remember;
706 }),
707 );
708 });
709
710 const loadEarlier = Effect.fn("EnvironmentThreadState.loadEarlier")(function* (
711 throughEntryId?: string,
712 workAnchorId?: string,
713 ): Effect.fn.Return<ThreadHistoryController.ThreadHistoryLoadEarlierResult> {
714 const workCursor = (latest: EnvironmentThreadState) =>
715 Option.getOrNull(latest.data)?.visibleTurnItems.find(
716 (row) => row.sourceItemId === workAnchorId,
717 )?.item.deferredWork?.cursor ?? null;
718 const current = yield* SubscriptionRef.get(state);
719 const initialCursor =
720 workAnchorId === undefined ? current.history.historyCursor : workCursor(current);
721 if (
722 current.status === "deleted" ||
723 Option.isNone(current.data) ||
724 initialCursor === null ||
725 (workAnchorId === undefined && !current.history.hasMoreHistory)
726 )
727 return { _tag: "noop" };
728 if (workAnchorId === undefined && current.history.loading) return { _tag: "busy" };
729 let requestCursor = initialCursor;
730 const isCurrent = (latest: EnvironmentThreadState) =>
731 workAnchorId === undefined
732 ? isActiveHistoryRequestCursor(requestCursor, latest.history)
733 : workCursor(latest) === requestCursor;
734 const staleResult = (
735 latest: EnvironmentThreadState,
736 ): ThreadHistoryController.ThreadHistoryLoadEarlierResult =>
737 workAnchorId !== undefined && workCursor(latest) !== null
738 ? { _tag: "error", message: "Work history changed. Retry loading it." }
739 : { _tag: "noop" };
740 if (workAnchorId === undefined)
741 yield* patchHistoryMeta((history) =>
742 isActiveHistoryRequestCursor(requestCursor, history)
743 ? { ...history, loading: true, error: null }
744 : history,
745 );
746 const failLoad = (message: string) =>
747 SubscriptionRef.modify(
748 state,
749 (
750 latest,
751 ): readonly [
752 ThreadHistoryController.ThreadHistoryLoadEarlierResult,
753 EnvironmentThreadState,
754 ] =>
755 isCurrent(latest)
756 ? [
757 { _tag: "error", message },
758 {
759 ...latest,
760 history:
761 workAnchorId === undefined
762 ? { ...latest.history, loading: false, error: message }
763 : latest.history,
764 },
765 ]
766 : [staleResult(latest), latest],
767 );
768
769 const runLoad = Effect.gen(function* () {
770 while (true) {
771 const preparedOption = yield* SubscriptionRef.get(supervisor.prepared);
772 if (Option.isNone(preparedOption) || Option.isNone(httpClient))
773 return yield* failLoad("Environment is not connected.");
774 const views =
775 workAnchorId !== undefined
776 ? ["work" as const]
777 : throughEntryId === undefined
778 ? [undefined]
779 : ["conversation" as const, undefined];
780 let nextCursor: string | null = null;
781 for (const view of views) {
782 const pageResult = yield* fetchEnvironmentThreadHistoryPage({
783 prepared: preparedOption.value,
784 threadId,
785 cursor: requestCursor,
786 throughEntryId: workAnchorId ?? throughEntryId,
787 view,
788 signer: dpopSigner,
789 remoteAuthorization,
790 }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient.value), Effect.result);
791 if (Result.isFailure(pageResult))
792 return yield* failLoad(formatHistoryError(pageResult.failure));
793 const page = pageResult.success;
794 const complete = view !== "conversation";
795 const keepPaging =
796 complete &&
797 (workAnchorId !== undefined
798 ? page.nextCursor !== null
799 : throughEntryId === undefined &&
800 page.hasMoreHistory &&
801 !page.items.some(({ item }) => startsTurn(item)));
802 if (keepPaging && (page.nextCursor === null || page.nextCursor === requestCursor))
803 return yield* failLoad(
804 workAnchorId === undefined
805 ? "History did not advance."
806 : "Work history did not advance.",
807 );
808 const waiting = yield* Ref.get(awaitingCompletion);
809 const result = yield* applyLock.withPermits(1)(
810 SubscriptionRef.modify(
811 state,
812 (
813 latest,
814 ): readonly [
815 ThreadHistoryController.ThreadHistoryLoadEarlierResult,
816 EnvironmentThreadState,
817 ] => {
818 if (!isCurrent(latest)) return [staleResult(latest), latest];
819 if (Option.isNone(latest.data) || latest.status === "deleted")
820 return [{ _tag: "noop" }, { ...latest, history: EMPTY_THREAD_HISTORY_META }];
821 return [
822 { _tag: "loaded" },
823 {
824 ...latest,
825 data: Option.some(
826 mergeOlderHistoryIntoProjection(
827 latest.data.value,
828 page.items,
829 workAnchorId === undefined
830 ? undefined
831 : { anchorId: workAnchorId, nextCursor: page.nextCursor },
832 ),
833 ),
834 status:
835 workAnchorId === undefined && complete && waiting
836 ? "synchronizing"
837 : latest.status,
838 history:
839 workAnchorId === undefined && complete
840 ? { ...applyHistoryPageMeta(latest.history, page), loading: keepPaging }
841 : { ...latest.history, expanded: true },
842 },
843 ];
844 },
845 ),
846 );
847 if (result._tag !== "loaded") return result;
848 if (complete) yield* applyLock.withPermits(1)(remember);
849 nextCursor = keepPaging ? page.nextCursor : null;
850 }
851 if (nextCursor === null) return { _tag: "loaded" } as const;
852 requestCursor = nextCursor;
853 }
854 });
855 return yield* runLoad.pipe(
856 Effect.onInterrupt(() =>
857 SubscriptionRef.update(state, (latest) => ({
858 ...latest,
859 history: clearActiveHistoryLoading(requestCursor, latest.history),
860 })),
861 ),
862 );
863 });
864
865 // A cached item can stay running after the server finished it, and no event will ever
866 // name it again: it can sit outside every bounded window, and a repeated Stop records
867 // nothing new. Stop asks the server about each pending item instead of
868 // inferring status from what a snapshot omits, and settles only those items.
869 const reconcilePendingWork = Effect.fn("EnvironmentThreadState.reconcilePendingWork")(
870 function* () {
871 const current = yield* SubscriptionRef.get(state);
872 if (current.status === "deleted" || Option.isNone(current.data)) return;
873 const pending = pendingBackgroundTurnItems(current.data.value);
874 if (pending.length === 0) return;
875 const queried = new Set<string>(pending.map((item) => item.id));
876 const fetched = yield* Effect.forEach(
877 pending,
878 (item) =>
879 request(ORCHESTRATION_V2_WS_METHODS.getTurnItem, { threadId, itemId: item.id }).pipe(
880 Effect.timeout("10 seconds"),
881 Effect.map((result) => result.item),
882 Effect.orElseSucceed(() => null),
883 ),
884 { concurrency: 4 },
885 );
886 const settled = new Map(
887 fetched.flatMap((item) =>
888 item === null || isOrchestrationV2WorkActive(item.status) ? [] : [[item.id, item]],
889 ),
890 );
891 yield* applyLock.withPermits(1)(
892 Effect.gen(function* () {
893 const latest = yield* SubscriptionRef.get(state);
894 if (latest.status === "deleted" || Option.isNone(latest.data)) return;
895 const thread = settleTurnItems(threadId, latest.data.value, settled);
896 if (thread !== latest.data.value) {
897 yield* setThread(thread);
898 yield* remember;
899 }
900 // An expanded timeline is never saved, so reload would bring the stale row back.
901 // Repair the snapshot last handed to the cache instead, keeping its window and cursor.
902 // A resumed state skipped the cache load, so read the saved snapshot then.
903 if (shouldPersistThread(thread, latest.history)) return;
904 // A terminal event that landed during the lookup is newer than any reply, and the
905 // expanded live state never saves it, whatever the lookup returned.
906 const latestItems = new Map(settled);
907 for (const item of thread.turnItems) {
908 if (queried.has(item.id) && !isOrchestrationV2WorkActive(item.status)) {
909 latestItems.set(item.id, item);
910 }
911 }
912 if (latestItems.size === 0) return;
913 const saved =
914 queuedSnapshot ??
915 Option.getOrUndefined(
916 yield* cache
917 .loadThread(environmentId, threadId)
918 .pipe(Effect.orElseSucceed(() => Option.none())),
919 );
920 if (saved === undefined) return;
921 const projection = settleTurnItems(threadId, saved.projection, latestItems);
922 if (projection !== saved.projection) yield* queuePersist({ ...saved, projection });
923 }),
924 );
925 },
926 );
927
928 if (Option.isSome(historyController)) {
929 const scope = yield* Scope.Scope;
930 const registration = yield* historyController.value.register(environmentId, threadId, {
931 loadEarlier: (throughEntryId, workAnchorId) =>
932 loadEarlier(throughEntryId, workAnchorId).pipe(
933 Effect.forkIn(scope),
934 Effect.flatMap(Fiber.join),
935 ),
936 reconcilePendingWork: () =>
937 reconcilePendingWork().pipe(
938 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
939 Effect.forkIn(scope),
940 Effect.flatMap(Fiber.join),
941 ),
942 });
943 yield* Effect.addFinalizer(() => historyController.value.unregister(registration));
944 }
945
946 yield* SubscriptionRef.changes(supervisor.state).pipe(
947 Stream.runForEach((connectionState) => {
948 switch (connectionProjectionPhase(connectionState)) {
949 case "synchronizing":
950 return setConnecting;
951 case "disconnected":
952 return setDisconnected;
953 case "ready":
954 return setReady;
955 }
956 }),
957 Effect.forkScoped,
958 );
959
960 const foregroundResubscriptions = Option.match(wakeups, {
961 onNone: () => Stream.never,
962 onSome: (service) =>
963 service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),
964 });
965
966 // Only the first subscription after a warm live resume keeps the retained
967 // status. A replacement session or foreground resubscribe on the same scope
968 // may have missed events, so those show sync progress until confirmed.
969 const resumingLive = yield* Ref.make(initialState.status === "live");
970 // A cold open tries the socket snapshot once. Any later cold attempt, after a
971 // failure or a dropped session, asks HTTP, which also tells a missing thread
972 // apart from a failed load.
973 const socketOpenTried = yield* Ref.make(false);
974 const socketOnlyOpen = yield* Ref.make(false);
975 const failedSocketOnlyOpen = Effect.gen(function* () {
976 return (
977 (yield* Ref.get(socketOnlyOpen)) && Option.isNone((yield* SubscriptionRef.get(state)).data)
978 );
979 });
980 // A stream defect parks the subscription until something resubscribes it.
981 const resubscribes = yield* Queue.unbounded<void>();
982 const markSynchronizing = Effect.gen(function* () {
983 if (yield* Ref.get(resumingLive)) return;
984 yield* SubscriptionRef.update(state, (current) =>
985 current.status === "deleted"
986 ? current
987 : { ...current, status: "synchronizing" as const, error: Option.none() },
988 );
989 });
990
991 yield* markSynchronizing;
992 // A 404 can arrive before the thread exists, so a miss asks HTTP again while mounted.
993 let httpMisses = 0;
994 let defectRetries = 0;
995 const backOffAfterDefect = Effect.suspend(() => {
996 const retryDelay = subscriptionRetryDelay(THREAD_SUBSCRIPTION_RETRY_DELAY, defectRetries);
997 defectRetries += 1;
998 return setStreamError("Could not synchronize the thread.").pipe(
999 Effect.andThen(Effect.sleep(retryDelay)),
1000 );
1001 });
1002 const followThread = (): Effect.Effect<
1003 void,
1004 never,
1005 EnvironmentSupervisor.EnvironmentSupervisor
1006 > =>
1007 subscribeDynamic(
1008 ORCHESTRATION_V2_WS_METHODS.subscribeThread,
1009 Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) {
1010 let current = yield* SubscriptionRef.get(state);
1011 // A delete event already cleared this thread. Park without opening the socket.
1012 if (current.status === "deleted" && httpMisses === 0) {
1013 return yield* Effect.never;
1014 }
1015
1016 const config = yield* Effect.option(session.initialConfig);
1017 yield* Ref.set(awaitingCompletion, true);
1018 yield* markSynchronizing;
1019 yield* Ref.set(resumingLive, false);
1020
1021 // A server that pages snapshots sends the same bounded window over the
1022 // socket, so a cold open can skip the HTTP round trip.
1023 const socketOnly =
1024 Option.isNone(current.data) &&
1025 canLoadHistory &&
1026 Option.exists(config, (value) => value.threadSnapshotPagination === true) &&
1027 !(yield* Ref.getAndSet(socketOpenTried, true));
1028 yield* Ref.set(socketOnlyOpen, socketOnly);
1029
1030 if (Option.isNone(current.data) && !socketOnly) {
1031 const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(
1032 Effect.flatMap(
1033 Option.match({
1034 onSome: Effect.succeed,
1035 onNone: () =>
1036 SubscriptionRef.changes(supervisor.prepared).pipe(
1037 Stream.filter(Option.isSome),
1038 Stream.map((value) => value.value),
1039 Stream.runHead,
1040 Effect.map(Option.getOrThrow),
1041 ),
1042 }),
1043 ),
1044 );
1045 const httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult =
1046 yield* snapshotLoader.load(prepared, threadId);
1047 switch (httpResult._tag) {
1048 case "present": {
1049 httpMisses = 0;
1050 if (canLoadHistory && httpResult.history !== undefined) {
1051 yield* Ref.set(acceptsBoundedSocketSnapshots, true);
1052 }
1053 // Atomic projection + progressive meta so a settled bounded window
1054 // never persists as a complete full-timeline cache entry. Socket
1055 // snapshots still go through applyItem (resetHistory).
1056 yield* applyLock.withPermits(1)(
1057 Effect.gen(function* () {
1058 yield* SubscriptionRef.set(lastSequence, httpResult.snapshot.snapshotSequence);
1059 const history: ThreadHistoryMeta =
1060 httpResult.history !== undefined
1061 ? {
1062 historyCursor: httpResult.history.historyCursor,
1063 hasMoreHistory: httpResult.history.hasMoreHistory,
1064 loading: false,
1065 error: null,
1066 expanded: false,
1067 latestLocalTurnOrdinal: httpResult.history.latestLocalTurnOrdinal ?? null,
1068 }
1069 : EMPTY_THREAD_HISTORY_META;
1070 yield* setThread(httpResult.snapshot.projection, { history });
1071 yield* remember;
1072 }),
1073 );
1074 current = yield* SubscriptionRef.get(state);
1075 break;
1076 }
1077 case "missing": {
1078 if (current.status !== "deleted") yield* setDeleted();
1079 // A 404 also means "not created yet". Only a delete event may outlive this mount.
1080 if (resumeCache?.owner === owner) resumeCache.snapshot = undefined;
1081 yield* Effect.sleep(
1082 subscriptionRetryDelay(THREAD_SUBSCRIPTION_RETRY_DELAY, httpMisses++),
1083 );
1084 yield* Queue.offer(resubscribes, undefined);
1085 return yield* Effect.never;
1086 }
1087 case "unavailable": {
1088 break;
1089 }
1090 }
1091 }
1092
1093 const sequence = yield* SubscriptionRef.get(lastSequence);
1094 const canResume = Option.isSome(current.data);
1095 const acceptBoundedSnapshot = yield* Ref.get(acceptsBoundedSocketSnapshots);
1096
1097 return {
1098 threadId,
1099 ...(canResume ? { afterSequence: sequence } : {}),
1100 requestCompletionMarker: true as const,
1101 ...(acceptBoundedSnapshot
1102 ? {
1103 acceptBoundedSnapshot: true as const,
1104 acceptDeferredWork: true as const,
1105 acceptCompactTurnItems: true as const,
1106 acceptCompactIds: true as const,
1107 }
1108 : {}),
1109 acceptDedupedBodies: true as const,
1110 };
1111 }),
1112 {
1113 // A socket-only open that fails retries over HTTP, which tells a
1114 // missing thread apart from a failure, before showing an error.
1115 onDefect: Effect.fn("EnvironmentThreadState.onDefect")(function* (cause) {
1116 if (yield* failedSocketOnlyOpen) {
1117 yield* Ref.set(socketOnlyOpen, false);
1118 return yield* Queue.offer(resubscribes, undefined);
1119 }
1120 if (!Cause.hasDies(cause)) {
1121 return yield* setStreamError("Could not synchronize the thread.");
1122 }
1123 yield* backOffAfterDefect;
1124 yield* Queue.offer(resubscribes, undefined);
1125 }),
1126 onExpectedFailure: Effect.fn("EnvironmentThreadState.onExpectedFailure")(function* (cause) {
1127 const retryOverHttp = !isAuthorizationFailure(cause) && (yield* failedSocketOnlyOpen);
1128 if (!retryOverHttp) yield* setStreamError(formatThreadError(cause));
1129 }),
1130 retryExpectedFailureAfter: THREAD_SUBSCRIPTION_RETRY_DELAY,
1131 resubscribe: Stream.merge(foregroundResubscriptions, Stream.fromQueue(resubscribes)),
1132 },
1133 ).pipe(
1134 Stream.runForEachArray((items) =>
1135 applyItems(items).pipe(
1136 Effect.andThen(
1137 Effect.sync(() => {
1138 defectRetries = 0;
1139 }),
1140 ),
1141 ),
1142 ),
1143 // A defect while applying items ends the stream; resubscribe for a fresh snapshot.
1144 Effect.catchCause((cause) =>
1145 Cause.hasInterrupts(cause)
1146 ? Effect.interrupt
1147 : Effect.logWarning("Thread subscription failed; resubscribing.", {
1148 cause: Cause.pretty(cause),
1149 environmentId,
1150 threadId,
1151 }).pipe(
1152 Effect.andThen(backOffAfterDefect),
1153 Effect.andThen(Effect.suspend(followThread)),
1154 ),
1155 ),
1156 );
1157 yield* Effect.forkScoped(followThread());
1158
1159 return state;
1160});
1161
1162function threadStateChanges(
1163 environmentId: EnvironmentIdType,
1164 threadId: ThreadIdType,
1165 resumeCache?: ThreadResumeCache,
1166) {
1167 return followStreamInEnvironment(
1168 environmentId,
1169 Stream.unwrap(
1170 makeEnvironmentThreadState(threadId, resumeCache).pipe(Effect.map(SubscriptionRef.changes)),
1171 ),
1172 );
1173}
1174
1175export function createEnvironmentThreadStateAtoms<R, E>(
1176 runtime: Atom.AtomRuntime<
1177 | EnvironmentRegistry.EnvironmentRegistry
1178 | Persistence.EnvironmentCacheStore
1179 | ThreadSnapshotLoader.ThreadSnapshotLoader
1180 | R,
1181 E
1182 >,
1183) {
1184 // Cache definitions must outlive collectible live-atom definitions. The
1185 // registry retains these nodes without retaining environment or RPC scopes.
1186 const resumeFamily = Atom.family((key: string) =>
1187 Atom.make((): ThreadResumeCache => ({
1188 snapshot: undefined,
1189 owner: undefined,
1190 })).pipe(
1191 Atom.setIdleTTL(THREAD_SNAPSHOT_IDLE_TTL_MS),
1192 Atom.withLabel(`environment-thread-resume:${key}`),
1193 ),
1194 );
1195 const family = Atom.family((key: string) => {
1196 const { environmentId, threadId } = parseThreadKey(key);
1197 const resumeAtom = resumeFamily(key);
1198 return runtime
1199 .atom(
1200 (get) => {
1201 get.mount(resumeAtom);
1202 const resume = get.once(resumeAtom);
1203 const live = threadStateChanges(environmentId, threadId, resume);
1204 return resume.snapshot === undefined
1205 ? live
1206 : Stream.concat(Stream.succeed(cachedThreadState(resume.snapshot.state)), live);
1207 },
1208 {
1209 initialValue: EMPTY_ENVIRONMENT_THREAD_STATE,
1210 },
1211 )
1212 .pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-state:${key}`));
1213 });
1214
1215 return {
1216 stateAtom: (environmentId: EnvironmentIdType, threadId: ThreadIdType) =>
1217 family(threadKey({ environmentId, threadId })),
1218 };
1219}
1220
1221export * from "./archivedThreads.ts";
1222export * from "./checkpointDiff.ts";
1223export * as BoundedThreadSnapshotLoader from "./boundedThreadSnapshotHttp.ts";
1224export * as ThreadHistoryController from "./threadHistoryController.ts";
1225export type { ThreadHistoryLoadEarlierResult } from "./threadHistoryController.ts";
1226export * from "./threadHistoryMerge.ts";
1227export * from "./threadSnapshotHttp.ts";
1228export * from "./composerPathSearch.ts";
1229export * from "./threadCommands.ts";
1230export * from "./threadFeedback.ts";
1231export * from "./threadDetail.ts";
1232export * from "./threadFamilies.ts";
1233export * from "./olderSettledThreads.ts";
1234export * from "./threadShell.ts";
1235export * from "./threadState.ts";
1236export * from "./threadTagFilter.ts";