1import {
2 ORCHESTRATION_V2_WS_METHODS,
3 type EnvironmentId as EnvironmentIdType,
4 type OrchestrationV2ThreadDetailSnapshot,
5 type OrchestrationV2ThreadProjection,
6 type OrchestrationV2ThreadStreamItem,
7 type ThreadId as ThreadIdType,
8} from "@t3tools/contracts";
9import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
10import * as Cause from "effect/Cause";
11import * as Effect from "effect/Effect";
12import * as Fiber from "effect/Fiber";
13import * as Scope from "effect/Scope";
14import * as Option from "effect/Option";
15import * as Queue from "effect/Queue";
16import * as Ref from "effect/Ref";
17import * as Result from "effect/Result";
18import * as Semaphore from "effect/Semaphore";
19import * as Stream from "effect/Stream";
20import * as SubscriptionRef from "effect/SubscriptionRef";
21import { HttpClient } from "effect/http";
22import { Atom } from "effect/reactivity";
23
24import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
25import * as EnvironmentRegistry from "../connection/registry.ts";
26import { connectionProjectionPhase } from "../connection/model.ts";
27import * as EnvironmentSupervisor from "../connection/supervisor.ts";
28import * as ConnectionWakeups from "../connection/wakeups.ts";
29import * as Persistence from "../platform/persistence.ts";
30import { runCachePersistence } from "./cachePersistence.ts";
31import * as ManagedRelay from "../relay/managedRelay.ts";
32import { subscribeDynamic } from "../rpc/client.ts";
33import { parseThreadKey, threadKey } from "./entities.ts";
34import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts";
35import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts";
36import { followStreamInEnvironment } from "./runtime.ts";
37import * as ThreadHistoryController from "./threadHistoryController.ts";
38import { fetchEnvironmentThreadHistoryPage } from "./threadHistoryHttp.ts";
39import {
40 applyHistoryPageMeta,
41 clearActiveHistoryLoading,
42 EMPTY_THREAD_HISTORY_META,
43 isActiveHistoryRequestCursor,
44 mergeOlderHistoryIntoProjection,
45 type ThreadHistoryMeta,
46} from "./threadHistoryMerge.ts";
47import { fetchArchivedThreadSnapshot } from "./archiveThreadHttp.ts";
48import * as HubConnection from "../connection/hub.ts";
49import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts";
50import {
51 EMPTY_ENVIRONMENT_THREAD_STATE,
52 type EnvironmentThreadState,
53 type EnvironmentThreadStatus,
54} from "./threadState.ts";
55
56function statusWithoutLiveData(
57 data: Option.Option<OrchestrationV2ThreadProjection>,
58): EnvironmentThreadStatus {
59 return Option.isSome(data) ? "cached" : "empty";
60}
61
62function formatThreadError(cause: Cause.Cause<unknown>): string {
63 const error = Cause.squash(cause);
64 return error instanceof Error && error.message.trim().length > 0
65 ? error.message
66 : "Could not synchronize the thread.";
67}
68
69function formatHistoryError(error: unknown): string {
70 if (error instanceof Error && error.message.trim().length > 0) {
71 return error.message;
72 }
73 return "Could not load earlier activity.";
74}
75
76function historyMetaFromCachedSnapshot(
77 snapshot: OrchestrationV2ThreadDetailSnapshot,
78): ThreadHistoryMeta {
79 const historyCursor = snapshot.historyCursor ?? null;
80 const hasMoreHistory = snapshot.hasMoreHistory ?? false;
81 return {
82 historyCursor,
83 hasMoreHistory,
84 loading: false,
85 error: null,
86 // Cache never stores expanded progressive history (load-earlier growth).
87 expanded: false,
88 latestLocalTurnOrdinal: snapshot.latestLocalTurnOrdinal ?? null,
89 };
90}
91
92function snapshotToPersist(
93 snapshotSequence: number,
94 projection: OrchestrationV2ThreadProjection,
95 history: ThreadHistoryMeta,
96 acceptsBoundedSnapshots: boolean,
97): OrchestrationV2ThreadDetailSnapshot {
98 // A complete bounded snapshot still proves paging support. Retain that evidence
99 // so a thread that grows while closed can resume with a bounded fallback.
100 if (acceptsBoundedSnapshots || history.hasMoreHistory || history.historyCursor !== null) {
101 return {
102 snapshotSequence,
103 projection,
104 historyCursor: history.historyCursor,
105 hasMoreHistory: history.hasMoreHistory,
106 latestLocalTurnOrdinal: history.latestLocalTurnOrdinal,
107 };
108 }
109 return { snapshotSequence, projection };
110}
111
112function shouldPersistThread(
113 thread: OrchestrationV2ThreadProjection,
114 history: ThreadHistoryMeta,
115): boolean {
116 // After the user loads older pages the in-memory timeline can grow large.
117 // Keep those expanded projections out of the monolithic cache.
118 if (history.expanded) {
119 return false;
120 }
121 return !thread.runs.some(
122 (run) => run.status === "preparing" || run.status === "starting" || run.status === "running",
123 );
124}
125
126interface ThreadResumeSnapshot {
127 readonly state: EnvironmentThreadState;
128 readonly sequence: number;
129 readonly persisted: boolean;
130 readonly acceptsBoundedSnapshots?: boolean;
131}
132
133interface ThreadResumeCache {
134 snapshot: ThreadResumeSnapshot | undefined;
135 owner: object | undefined;
136}
137
138function matchesThreadSnapshot(
139 current: ThreadResumeSnapshot,
140 projection: OrchestrationV2ThreadProjection | null,
141 sequence: number,
142 history: Pick<ThreadHistoryMeta, "historyCursor" | "hasMoreHistory" | "latestLocalTurnOrdinal">,
143): boolean {
144 if (current.sequence !== sequence || Option.getOrNull(current.state.data) !== projection)
145 return false;
146 const currentHistory = current.state.history;
147 return (
148 currentHistory.historyCursor === history.historyCursor &&
149 currentHistory.hasMoreHistory === history.hasMoreHistory &&
150 (!(history.hasMoreHistory || history.historyCursor !== null) ||
151 currentHistory.latestLocalTurnOrdinal === history.latestLocalTurnOrdinal)
152 );
153}
154
155// A retained "live" state stays live: the cursor resume that follows only
156// replays what the thread missed, and on servers that send the completion
157// marker the first replayed event moves the status to "synchronizing" on its
158// own. Downgrading here would flash a sync label on every return to a
159// recently viewed thread.
160function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadState {
161 return {
162 ...value,
163 status:
164 value.status === "deleted" || (value.status === "live" && Option.isSome(value.data))
165 ? value.status
166 : statusWithoutLiveData(value.data),
167 error: Option.none(),
168 history: { ...value.history, loading: false, error: null },
169 };
170}
171
172export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* (
173 threadId: ThreadIdType,
174 resumeCache?: ThreadResumeCache,
175) {
176 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
177 const cache = yield* Persistence.EnvironmentCacheStore;
178 const snapshotLoader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader;
179 const historyController = yield* Effect.serviceOption(
180 ThreadHistoryController.ThreadHistoryController,
181 );
182 const httpClient = yield* Effect.serviceOption(HttpClient.HttpClient);
183 const archiveHub = yield* Effect.serviceOption(HubConnection.HubConnection);
184 const dpopSigner = yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner);
185 const remoteAuthorization = yield* Effect.serviceOption(
186 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization,
187 );
188 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);
189 const environmentId = supervisor.target.environmentId;
190 const retained = resumeCache?.snapshot;
191 const owner = {};
192 if (resumeCache) resumeCache.owner = owner;
193 const cached =
194 retained === undefined
195 ? yield* cache.loadThread(environmentId, threadId).pipe(
196 Effect.catch((error) =>
197 Effect.logWarning("Could not load cached thread.").pipe(
198 Effect.annotateLogs({
199 environmentId,
200 threadId,
201 error: error.message,
202 }),
203 Effect.as(Option.none<OrchestrationV2ThreadDetailSnapshot>()),
204 ),
205 ),
206 )
207 : Option.none<OrchestrationV2ThreadDetailSnapshot>();
208 const cachedThread = Option.map(cached, (snapshot) => snapshot.projection);
209 const initialState: EnvironmentThreadState = retained
210 ? cachedThreadState(retained.state)
211 : {
212 data: cachedThread,
213 status: statusWithoutLiveData(cachedThread),
214 error: Option.none(),
215 history: Option.match(cached, {
216 onNone: () => EMPTY_THREAD_HISTORY_META,
217 onSome: historyMetaFromCachedSnapshot,
218 }),
219 };
220 const state = yield* SubscriptionRef.make(initialState);
221 // Paging support belongs to the client, even when the initial HTTP request
222 // fails. A bounded socket reset retains a cursor so history can be retried.
223 const canLoadHistory = Option.isSome(httpClient) && Option.isSome(historyController);
224 const acceptsBoundedSocketSnapshots = yield* Ref.make(canLoadHistory);
225 // Seed the resume cursor from the cached snapshot so a warm cache can catch up
226 // via `afterSequence` instead of re-downloading the full thread body.
227 const initialSequence =
228 retained?.sequence ??
229 Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence });
230 const lastSequence = yield* SubscriptionRef.make(initialSequence);
231 let committed: ThreadResumeSnapshot = {
232 state: initialState,
233 sequence: initialSequence,
234 persisted: retained?.persisted ?? Option.isSome(cached),
235 acceptsBoundedSnapshots: canLoadHistory,
236 };
237 if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
238 const awaitingCompletion = yield* Ref.make(false);
239 const applyLock = yield* Semaphore.make(1);
240 // Save only completed data/cursor updates. A canceled scope must not cache
241 // a cursor whose event has not reached the data yet.
242 const remember = Effect.gen(function* () {
243 const current = yield* SubscriptionRef.get(state);
244 const sequence = yield* SubscriptionRef.get(lastSequence);
245 committed = {
246 state: current,
247 sequence,
248 acceptsBoundedSnapshots: yield* Ref.get(acceptsBoundedSocketSnapshots),
249 persisted:
250 committed.persisted &&
251 matchesThreadSnapshot(committed, Option.getOrNull(current.data), sequence, current.history),
252 };
253 if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
254 });
255 const persistence = yield* Queue.sliding<OrchestrationV2ThreadDetailSnapshot>(1);
256
257 const persist = Effect.fn("EnvironmentThreadState.persist")(function* (
258 snapshot: OrchestrationV2ThreadDetailSnapshot,
259 ) {
260 if (resumeCache !== undefined && resumeCache.owner !== owner) return;
261 // A deletion can arrive while an older snapshot waits in the persistence queue.
262 if (committed.state.status === "deleted") return;
263 if (
264 committed.persisted &&
265 matchesThreadSnapshot(
266 committed,
267 snapshot.projection,
268 snapshot.snapshotSequence,
269 historyMetaFromCachedSnapshot(snapshot),
270 )
271 )
272 return;
273 yield* cache.saveThread(environmentId, snapshot).pipe(
274 Effect.tap(() =>
275 Effect.sync(() => {
276 if (
277 !matchesThreadSnapshot(
278 committed,
279 snapshot.projection,
280 snapshot.snapshotSequence,
281 historyMetaFromCachedSnapshot(snapshot),
282 )
283 )
284 return;
285 committed = { ...committed, persisted: true };
286 if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
287 }),
288 ),
289 Effect.catch((error) =>
290 Effect.logWarning("Could not persist the thread cache.").pipe(
291 Effect.annotateLogs({
292 environmentId,
293 threadId,
294 error: error.message,
295 }),
296 ),
297 ),
298 );
299 });
300
301 yield* Effect.addFinalizer(() =>
302 Effect.suspend(() => {
303 const { state: current, sequence: snapshotSequence } = committed;
304 return Option.match(current.data, {
305 onNone: () => Effect.void,
306 onSome: (projection) =>
307 shouldPersistThread(projection, current.history)
308 ? persist(
309 snapshotToPersist(
310 snapshotSequence,
311 projection,
312 current.history,
313 committed.acceptsBoundedSnapshots === true,
314 ),
315 )
316 : Effect.void,
317 });
318 }),
319 );
320
321 yield* runCachePersistence(persistence, persist).pipe(Effect.forkScoped);
322
323 const setConnecting = SubscriptionRef.update(state, (current) =>
324 current.status === "deleted" || Option.isSome(current.error)
325 ? current
326 : {
327 ...current,
328 status: "synchronizing" as const,
329 error: Option.none(),
330 },
331 );
332 const setReady = SubscriptionRef.update(state, (current) =>
333 current.status === "live" || current.status === "deleted" || Option.isSome(current.error)
334 ? current
335 : {
336 ...current,
337 status: "synchronizing" as const,
338 error: Option.none(),
339 },
340 );
341 const setDisconnected = Effect.gen(function* () {
342 yield* Ref.set(awaitingCompletion, false);
343 yield* SubscriptionRef.update(state, (current) => ({
344 ...current,
345 status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
346 }));
347 });
348 const setStreamError = (message: string) =>
349 Ref.set(awaitingCompletion, false).pipe(
350 Effect.andThen(
351 SubscriptionRef.update(state, (current) => ({
352 ...current,
353 status:
354 current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
355 error: Option.some(message),
356 })),
357 ),
358 );
359
360 const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* (
361 thread: OrchestrationV2ThreadProjection,
362 options?: {
363 /** Socket/full snapshots: drop progressive meta with the new timeline. */
364 readonly resetHistory?: boolean;
365 readonly status?: "cached";
366 /**
367 * Explicit progressive meta installed atomically with the projection
368 * (bounded HTTP). Wins over resetHistory when both are supplied.
369 */
370 readonly history?: ThreadHistoryMeta;
371 },
372 ) {
373 const waiting = yield* Ref.get(awaitingCompletion);
374 // Atomic with concurrent history meta updates: never get-then-set the whole
375 // state when only the projection changes. Bounded installs pass history so
376 // projection + cursor persist together in one enqueue.
377 const next = yield* SubscriptionRef.updateAndGet(state, (previous) => {
378 const history =
379 options?.history !== undefined
380 ? options.history
381 : options?.resetHistory === true
382 ? EMPTY_THREAD_HISTORY_META
383 : previous.history;
384 return {
385 ...previous,
386 data: Option.some(thread),
387 // Buffered values from a failed attempt can arrive after its error.
388 status:
389 options?.status ??
390 (Option.isSome(previous.error)
391 ? ("cached" as const)
392 : waiting
393 ? ("synchronizing" as const)
394 : ("live" as const)),
395 error: previous.error,
396 history,
397 };
398 });
399 // Active projections can update many times per second and retain large tool
400 // payloads. Persist once the run settles so cache encoding stays off the
401 // streaming path. Progressive meta rides along when the window is incomplete.
402 if (shouldPersistThread(thread, next.history)) {
403 const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
404 yield* Queue.offer(
405 persistence,
406 snapshotToPersist(
407 snapshotSequence,
408 thread,
409 next.history,
410 yield* Ref.get(acceptsBoundedSocketSnapshots),
411 ),
412 );
413 }
414 });
415
416 const patchHistoryMeta = (patch: (history: ThreadHistoryMeta) => ThreadHistoryMeta) =>
417 SubscriptionRef.update(state, (current) => ({
418 ...current,
419 history: patch(current.history),
420 }));
421
422 const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () {
423 yield* Ref.set(awaitingCompletion, false);
424 yield* SubscriptionRef.set(state, {
425 data: Option.none(),
426 status: "deleted",
427 error: Option.none(),
428 history: EMPTY_THREAD_HISTORY_META,
429 });
430 yield* remember;
431 if (resumeCache !== undefined && resumeCache.owner !== owner) return;
432 yield* cache.removeThread(environmentId, threadId).pipe(
433 Effect.catch((error) =>
434 Effect.logWarning("Could not remove the cached thread.").pipe(
435 Effect.annotateLogs({
436 environmentId,
437 threadId,
438 error: error.message,
439 }),
440 ),
441 ),
442 );
443 });
444
445 type EventItem = Extract<OrchestrationV2ThreadStreamItem, { kind: "event" }>;
446 type SequencedItem = Extract<
447 OrchestrationV2ThreadStreamItem,
448 { kind: "event" | "unknown-event" }
449 >;
450 const applyEventsLocked = Effect.fn("EnvironmentThreadState.applyEventsLocked")(function* (
451 items: ReadonlyArray<SequencedItem>,
452 ) {
453 const appliedSequence = yield* SubscriptionRef.get(lastSequence);
454 let sequence = appliedSequence;
455 const fresh: EventItem[] = [];
456 for (const item of items) {
457 if (item.sequence <= sequence) continue;
458 // An event type from a newer server still moves the resume cursor past it.
459 sequence = item.sequence;
460 if (item.kind === "event") {
461 fresh.push(item);
462 continue;
463 }
464 yield* Effect.logDebug("Skipped a thread event type this client does not know.").pipe(
465 Effect.annotateLogs({
466 environmentId,
467 threadId,
468 // Bounded: the type comes from a newer server and is not validated here.
469 eventType: item.eventType.slice(0, 64),
470 sequence: item.sequence,
471 }),
472 );
473 }
474 if (sequence === appliedSequence) return;
475 yield* SubscriptionRef.set(lastSequence, sequence);
476 if (fresh.length === 0) return;
477
478 const waiting = yield* Ref.get(awaitingCompletion);
479 // Apply against the latest projection/history in one update so a concurrent
480 // loadEarlier merge (or history-meta patch) cannot be clobbered by a stale
481 // get-then-set rebuild.
482 type EventApplyResult =
483 | { readonly _tag: "noop" }
484 | { readonly _tag: "delete" }
485 | {
486 readonly _tag: "applied";
487 readonly projection: OrchestrationV2ThreadProjection;
488 readonly history: ThreadHistoryMeta;
489 };
490
491 const applyEvent = (
492 current: EnvironmentThreadState,
493 item: EventItem,
494 ): readonly [EventApplyResult, EnvironmentThreadState] => {
495 if (current.status === "deleted") {
496 return [{ _tag: "noop" }, current];
497 }
498 if (Option.isNone(current.data)) {
499 return [
500 item.event.type === "thread.deleted" ? { _tag: "delete" } : { _tag: "noop" },
501 current,
502 ];
503 }
504 if (item.event.type === "thread.deleted") {
505 return [{ _tag: "delete" }, current];
506 }
507
508 // Incomplete progressive windows only (hasMore or open cursor). Do not
509 // use expanded: it remains true after the last page as a cache marker.
510 // Full/web and fully-loaded timelines keep default append-on-miss.
511 const partial =
512 current.history.hasMoreHistory || current.history.historyCursor !== null
513 ? {
514 partialTimeline: true as const,
515 latestLocalTurnOrdinal: current.history.latestLocalTurnOrdinal,
516 }
517 : undefined;
518 const next = applyOrchestrationV2ProjectionEvent(current.data.value, item.event, partial);
519 // True no-op when the reducer deliberately returns the current projection
520 // reference (e.g. dropped old partial-timeline turn-item). Do not clear
521 // stream error/status or enqueue persistence.
522 if (next === null || next === current.data.value) {
523 return [{ _tag: "noop" }, current];
524 }
525
526 let history = current.history;
527 if (partial !== undefined && item.event.type === "turn-item.updated") {
528 const ordinal = item.event.payload.ordinal;
529 const watermark = history.latestLocalTurnOrdinal;
530 if (watermark === null || ordinal > watermark) {
531 history = { ...history, latestLocalTurnOrdinal: ordinal };
532 }
533 }
534
535 const updated: EnvironmentThreadState = {
536 ...current,
537 data: Option.some(next),
538 status: waiting ? "synchronizing" : "live",
539 error: Option.none(),
540 history,
541 };
542 return [{ _tag: "applied", projection: next, history: updated.history }, updated];
543 };
544
545 const result = yield* SubscriptionRef.modify(
546 state,
547 (current): readonly [EventApplyResult, EnvironmentThreadState] => {
548 let result: EventApplyResult = { _tag: "noop" };
549 for (const item of fresh) {
550 const [applied, next] = applyEvent(current, item);
551 current = next;
552 if (applied._tag !== "noop") result = applied;
553 if (applied._tag === "delete") break;
554 }
555 return [result, current];
556 },
557 );
558
559 if (result._tag === "delete") {
560 yield* setDeleted();
561 return;
562 }
563 if (result._tag === "applied" && shouldPersistThread(result.projection, result.history)) {
564 const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
565 yield* Queue.offer(
566 persistence,
567 snapshotToPersist(
568 snapshotSequence,
569 result.projection,
570 result.history,
571 yield* Ref.get(acceptsBoundedSocketSnapshots),
572 ),
573 );
574 }
575 });
576
577 const applyItemLocked = Effect.fn("EnvironmentThreadState.applyItemLocked")(function* (
578 item: OrchestrationV2ThreadStreamItem,
579 ) {
580 if (item.kind === "synchronized") {
581 yield* Ref.set(awaitingCompletion, false);
582 yield* SubscriptionRef.update(state, (current) =>
583 Option.isSome(current.data) && current.status !== "deleted" && Option.isNone(current.error)
584 ? { ...current, status: "live" as const, error: Option.none() }
585 : current,
586 );
587 return;
588 }
589
590 if (item.kind === "snapshot") {
591 yield* SubscriptionRef.set(lastSequence, item.snapshotSequence);
592 const hasProgressiveHistory =
593 item.historyCursor !== undefined ||
594 item.hasMoreHistory !== undefined ||
595 item.latestLocalTurnOrdinal !== undefined;
596 // Bounded socket fallbacks carry their cursor. Legacy-compatible full
597 // snapshots omit these fields and still replace progressive state.
598 yield* setThread(
599 boundedSnapshotProjection(item),
600 hasProgressiveHistory
601 ? {
602 history: {
603 historyCursor: item.historyCursor ?? null,
604 hasMoreHistory: item.hasMoreHistory ?? false,
605 loading: false,
606 error: null,
607 expanded: false,
608 latestLocalTurnOrdinal: item.latestLocalTurnOrdinal ?? null,
609 },
610 }
611 : { resetHistory: true },
612 );
613 return;
614 }
615
616 yield* applyEventsLocked([item]);
617 });
618
619 const applyItems = Effect.fn("EnvironmentThreadState.applyItems")(function* (
620 items: ReadonlyArray<OrchestrationV2ThreadStreamItem>,
621 ) {
622 yield* applyLock.withPermits(1)(
623 Effect.gen(function* () {
624 let events: SequencedItem[] = [];
625 for (const item of items) {
626 if (
627 item.kind === "unknown-event" ||
628 (item.kind === "event" && item.event.type !== "thread.deleted")
629 ) {
630 events.push(item);
631 continue;
632 }
633 yield* applyEventsLocked(events);
634 events = [];
635 yield* applyItemLocked(item);
636 }
637 yield* applyEventsLocked(events);
638 yield* remember;
639 }),
640 );
641 });
642
643 const loadEarlier = Effect.fn("EnvironmentThreadState.loadEarlier")(function* (
644 throughEntryId?: string,
645 ) {
646 const current = yield* SubscriptionRef.get(state);
647 if (
648 current.status === "deleted" ||
649 Option.isNone(current.data) ||
650 !current.history.hasMoreHistory ||
651 current.history.historyCursor === null
652 ) {
653 return { _tag: "noop" } satisfies ThreadHistoryController.ThreadHistoryLoadEarlierResult;
654 }
655 if (current.history.loading) {
656 return { _tag: "busy" } satisfies ThreadHistoryController.ThreadHistoryLoadEarlierResult;
657 }
658
659 // Capture the cursor that initiated this request. Completions/failures must
660 // no-op if a socket or new bounded snapshot replaced progressive meta mid-flight.
661 const requestCursor = current.history.historyCursor;
662 yield* patchHistoryMeta((history) =>
663 isActiveHistoryRequestCursor(requestCursor, history)
664 ? { ...history, loading: true, error: null }
665 : history,
666 );
667
668 const failLoad = (message: string) =>
669 SubscriptionRef.modify(
670 state,
671 (
672 latest,
673 ): readonly [
674 ThreadHistoryController.ThreadHistoryLoadEarlierResult,
675 EnvironmentThreadState,
676 ] =>
677 isActiveHistoryRequestCursor(requestCursor, latest.history)
678 ? [
679 { _tag: "error", message },
680 {
681 ...latest,
682 history: { ...latest.history, loading: false, error: message },
683 },
684 ]
685 : [{ _tag: "noop" }, latest],
686 );
687
688 const runLoad = Effect.gen(function* () {
689 const preparedOption = yield* SubscriptionRef.get(supervisor.prepared);
690 if (Option.isNone(preparedOption) || Option.isNone(httpClient))
691 return yield* failLoad("Environment is not connected.");
692
693 const views =
694 throughEntryId === undefined ? [undefined] : ["conversation" as const, undefined];
695 for (const view of views) {
696 const pageResult = yield* fetchEnvironmentThreadHistoryPage({
697 prepared: preparedOption.value,
698 threadId,
699 cursor: requestCursor,
700 throughEntryId,
701 view,
702 signer: dpopSigner,
703 remoteAuthorization,
704 }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient.value), Effect.result);
705 if (Result.isFailure(pageResult))
706 return yield* failLoad(formatHistoryError(pageResult.failure));
707 const page = pageResult.success;
708 const complete = view === undefined;
709 const waiting = yield* Ref.get(awaitingCompletion);
710 // Both stages merge against live state under the same lock. Only the
711 // complete page advances the cursor and releases the loading indicator.
712 const result = yield* applyLock.withPermits(1)(
713 SubscriptionRef.modify(
714 state,
715 (
716 latest,
717 ): readonly [
718 ThreadHistoryController.ThreadHistoryLoadEarlierResult,
719 EnvironmentThreadState,
720 ] => {
721 if (!isActiveHistoryRequestCursor(requestCursor, latest.history))
722 return [{ _tag: "noop" }, latest];
723 if (Option.isNone(latest.data) || latest.status === "deleted")
724 return [{ _tag: "noop" }, { ...latest, history: EMPTY_THREAD_HISTORY_META }];
725 return [
726 { _tag: "loaded" },
727 {
728 ...latest,
729 data: Option.some(mergeOlderHistoryIntoProjection(latest.data.value, page.items)),
730 status: complete && waiting ? "synchronizing" : latest.status,
731 history: complete
732 ? applyHistoryPageMeta(latest.history, page)
733 : { ...latest.history, expanded: true },
734 },
735 ];
736 },
737 ),
738 );
739 if (result._tag !== "loaded") return result;
740 if (complete) yield* remember;
741 }
742 return { _tag: "loaded" } satisfies ThreadHistoryController.ThreadHistoryLoadEarlierResult;
743 });
744
745 // On Effect interruption only: clear loading when this request cursor is
746 // still active. Never mutate stream status/error from the interrupt path.
747 return yield* runLoad.pipe(
748 Effect.onInterrupt(() =>
749 SubscriptionRef.update(state, (latest) => ({
750 ...latest,
751 history: clearActiveHistoryLoading(requestCursor, latest.history),
752 })),
753 ),
754 );
755 });
756
757 if (Option.isSome(historyController)) {
758 const scope = yield* Scope.Scope;
759 const registration = yield* historyController.value.register(environmentId, threadId, {
760 loadEarlier: (throughEntryId) =>
761 loadEarlier(throughEntryId).pipe(Effect.forkIn(scope), Effect.flatMap(Fiber.join)),
762 });
763 yield* Effect.addFinalizer(() => historyController.value.unregister(registration));
764 }
765
766 yield* SubscriptionRef.changes(supervisor.state).pipe(
767 Stream.runForEach((connectionState) => {
768 switch (connectionProjectionPhase(connectionState)) {
769 case "synchronizing":
770 return setConnecting;
771 case "disconnected":
772 return setDisconnected;
773 case "ready":
774 return setReady;
775 }
776 }),
777 Effect.forkScoped,
778 );
779
780 if (Option.isSome(archiveHub) && Option.isSome(httpClient)) {
781 yield* Stream.merge(
782 SubscriptionRef.changes(supervisor.state).pipe(Stream.map(() => undefined)),
783 SubscriptionRef.changes(archiveHub.value.state).pipe(Stream.map(() => undefined)),
784 ).pipe(
785 Stream.mapEffect(() =>
786 Effect.gen(function* () {
787 const current = yield* SubscriptionRef.get(state);
788 if (current.status === "live" || current.status === "deleted") return;
789 if (Option.isSome(current.data) && !current.history.hasMoreHistory) return;
790 const snapshot = yield* fetchArchivedThreadSnapshot(environmentId, threadId).pipe(
791 Effect.provideService(HubConnection.HubConnection, archiveHub.value),
792 Effect.provideService(HttpClient.HttpClient, httpClient.value),
793 Effect.option,
794 );
795 if (Option.isNone(snapshot)) return;
796 yield* applyLock.withPermits(1)(
797 Effect.gen(function* () {
798 const current = yield* SubscriptionRef.get(state);
799 if (
800 current.status === "live" ||
801 current.status === "deleted" ||
802 snapshot.value.snapshotSequence < (yield* SubscriptionRef.get(lastSequence))
803 )
804 return;
805 yield* SubscriptionRef.set(lastSequence, snapshot.value.snapshotSequence);
806 yield* setThread(snapshot.value.projection, { resetHistory: true, status: "cached" });
807 yield* remember;
808 }),
809 );
810 }),
811 ),
812 Stream.runDrain,
813 Effect.forkScoped,
814 );
815 }
816
817 const foregroundResubscriptions = Option.match(wakeups, {
818 onNone: () => Stream.never,
819 onSome: (service) =>
820 service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),
821 });
822
823 // Only the first subscription after a warm live resume keeps the retained
824 // status. A replacement session or foreground resubscribe on the same scope
825 // may have missed events, so those show sync progress until confirmed.
826 const resumingLive = yield* Ref.make(initialState.status === "live");
827 const markSynchronizing = Effect.gen(function* () {
828 if (yield* Ref.get(resumingLive)) return;
829 // Connection notifications do not establish that a terminated load restarted.
830 // Clear its diagnostic only when this subscription actually tries again.
831 yield* SubscriptionRef.update(state, (current) =>
832 current.status === "deleted"
833 ? current
834 : { ...current, status: "synchronizing" as const, error: Option.none() },
835 );
836 });
837
838 yield* markSynchronizing;
839 yield* Effect.forkScoped(
840 subscribeDynamic(
841 ORCHESTRATION_V2_WS_METHODS.subscribeThread,
842 Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) {
843 let current = yield* SubscriptionRef.get(state);
844 // A prior definitive miss (or delete event) already cleared this thread.
845 // Park the subscription attempt without opening the socket so we do not
846 // retry forever against a known-missing id.
847 if (current.status === "deleted") {
848 return yield* Effect.never;
849 }
850
851 const supportsCompletionMarker = yield* session.initialConfig.pipe(
852 Effect.map((config) => config.threadResumeCompletionMarker === true),
853 Effect.orElseSucceed(() => false),
854 );
855 yield* Ref.set(awaitingCompletion, supportsCompletionMarker);
856 yield* markSynchronizing;
857 yield* Ref.set(resumingLive, false);
858
859 if (Option.isNone(current.data)) {
860 const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(
861 Effect.flatMap(
862 Option.match({
863 onSome: Effect.succeed,
864 onNone: () =>
865 SubscriptionRef.changes(supervisor.prepared).pipe(
866 Stream.filter(Option.isSome),
867 Stream.map((value) => value.value),
868 Stream.runHead,
869 Effect.map(Option.getOrThrow),
870 ),
871 }),
872 ),
873 );
874 const httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult =
875 yield* snapshotLoader.load(prepared, threadId);
876 switch (httpResult._tag) {
877 case "present": {
878 if (canLoadHistory && httpResult.history !== undefined) {
879 yield* Ref.set(acceptsBoundedSocketSnapshots, true);
880 }
881 // Atomic projection + progressive meta so a settled bounded window
882 // never persists as a complete full-timeline cache entry. Socket
883 // snapshots still go through applyItem (resetHistory).
884 yield* applyLock.withPermits(1)(
885 Effect.gen(function* () {
886 yield* SubscriptionRef.set(lastSequence, httpResult.snapshot.snapshotSequence);
887 const history: ThreadHistoryMeta =
888 httpResult.history !== undefined
889 ? {
890 historyCursor: httpResult.history.historyCursor,
891 hasMoreHistory: httpResult.history.hasMoreHistory,
892 loading: false,
893 error: null,
894 expanded: false,
895 latestLocalTurnOrdinal: httpResult.history.latestLocalTurnOrdinal ?? null,
896 }
897 : EMPTY_THREAD_HISTORY_META;
898 yield* setThread(httpResult.snapshot.projection, { history });
899 yield* remember;
900 }),
901 );
902 current = yield* SubscriptionRef.get(state);
903 break;
904 }
905 case "missing": {
906 // Definitive HTTP 404: clear any stale cache and do not open or
907 // retry a socket subscription for this attempt.
908 yield* setDeleted();
909 return yield* Effect.never;
910 }
911 case "unavailable": {
912 // Transient HTTP failure: fall through to the socket path.
913 break;
914 }
915 }
916 }
917
918 const sequence = yield* SubscriptionRef.get(lastSequence);
919 const canResume = Option.isSome(current.data);
920 const acceptBoundedSnapshot = yield* Ref.get(acceptsBoundedSocketSnapshots);
921 if (!supportsCompletionMarker && canResume) {
922 yield* SubscriptionRef.update(state, (value) => ({
923 ...value,
924 status: value.status === "deleted" ? value.status : ("live" as const),
925 error: Option.none(),
926 }));
927 }
928
929 return {
930 threadId,
931 ...(canResume ? { afterSequence: sequence } : {}),
932 ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}),
933 ...(acceptBoundedSnapshot
934 ? { acceptBoundedSnapshot: true as const, acceptCompactTurnItems: true as const }
935 : {}),
936 };
937 }),
938 {
939 onDefect: () => setStreamError("Could not synchronize the thread."),
940 onExpectedFailure: (cause) => setStreamError(formatThreadError(cause)),
941 retryExpectedFailureAfter: "250 millis",
942 resubscribe: foregroundResubscriptions,
943 },
944 ).pipe(Stream.runForEachArray(applyItems)),
945 );
946
947 return state;
948});
949
950function threadStateChanges(
951 environmentId: EnvironmentIdType,
952 threadId: ThreadIdType,
953 resumeCache?: ThreadResumeCache,
954) {
955 return followStreamInEnvironment(
956 environmentId,
957 Stream.unwrap(
958 makeEnvironmentThreadState(threadId, resumeCache).pipe(Effect.map(SubscriptionRef.changes)),
959 ),
960 );
961}
962
963export function createEnvironmentThreadStateAtoms<R, E>(
964 runtime: Atom.AtomRuntime<
965 | EnvironmentRegistry.EnvironmentRegistry
966 | Persistence.EnvironmentCacheStore
967 | ThreadSnapshotLoader.ThreadSnapshotLoader
968 | R,
969 E
970 >,
971) {
972 // Cache definitions must outlive collectible live-atom definitions. The
973 // registry retains these nodes without retaining environment or RPC scopes.
974 const resumeFamily = Atom.family((key: string) =>
975 Atom.make((): ThreadResumeCache => ({
976 snapshot: undefined,
977 owner: undefined,
978 })).pipe(
979 Atom.setIdleTTL(THREAD_SNAPSHOT_IDLE_TTL_MS),
980 Atom.withLabel(`environment-thread-resume:${key}`),
981 ),
982 );
983 const family = Atom.family((key: string) => {
984 const { environmentId, threadId } = parseThreadKey(key);
985 const resumeAtom = resumeFamily(key);
986 return runtime
987 .atom(
988 (get) => {
989 get.mount(resumeAtom);
990 const resume = get.once(resumeAtom);
991 const live = threadStateChanges(environmentId, threadId, resume);
992 return resume.snapshot === undefined
993 ? live
994 : Stream.concat(Stream.succeed(cachedThreadState(resume.snapshot.state)), live);
995 },
996 {
997 initialValue: EMPTY_ENVIRONMENT_THREAD_STATE,
998 },
999 )
1000 .pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-state:${key}`));
1001 });
1002
1003 return {
1004 stateAtom: (environmentId: EnvironmentIdType, threadId: ThreadIdType) =>
1005 family(threadKey({ environmentId, threadId })),
1006 };
1007}
1008
1009export * from "./archivedThreads.ts";
1010export * from "./checkpointDiff.ts";
1011export * as BoundedThreadSnapshotLoader from "./boundedThreadSnapshotHttp.ts";
1012export * as ThreadHistoryController from "./threadHistoryController.ts";
1013// Flat so consumers' inferred types can name it.
1014export type { ThreadHistoryLoadEarlierResult } from "./threadHistoryController.ts";
1015export * from "./threadHistoryMerge.ts";
1016export * from "./threadSnapshotHttp.ts";
1017export * from "./composerPathSearch.ts";
1018export * from "./threadCommands.ts";
1019export * from "./threadFeedback.ts";
1020export * from "./threadDetail.ts";
1021export * from "./threadShell.ts";
1022export * from "./threadState.ts";