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