import { ORCHESTRATION_V2_WS_METHODS, type EnvironmentId as EnvironmentIdType, type OrchestrationV2ThreadDetailSnapshot, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadStreamItem, type ThreadId as ThreadIdType,} from "@t3tools/contracts";import * as Cause from "effect/Cause";import * as Effect from "effect/Effect";import * as Fiber from "effect/Fiber";import * as Scope from "effect/Scope";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";import * as Result from "effect/Result";import * as Semaphore from "effect/Semaphore";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import { HttpClient } from "effect/unstable/http";import { Atom } from "effect/unstable/reactivity"; import { RemoteEnvironmentAuthorization } from "../authorization/service.ts";import { EnvironmentRegistry } from "../connection/registry.ts";import { connectionProjectionPhase } from "../connection/model.ts";import { EnvironmentSupervisor } from "../connection/supervisor.ts";import * as ConnectionWakeups from "../connection/wakeups.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts";import { subscribeDynamic } from "../rpc/client.ts";import { parseThreadKey, threadKey } from "./entities.ts";import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts";import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts";import { followStreamInEnvironment } from "./runtime.ts";import { ThreadHistoryController, type ThreadHistoryLoadEarlierResult,} from "./threadHistoryController.ts";import { fetchEnvironmentThreadHistoryPage } from "./threadHistoryHttp.ts";import { applyHistoryPageMeta, clearActiveHistoryLoading, EMPTY_THREAD_HISTORY_META, isActiveHistoryRequestCursor, mergeOlderHistoryIntoProjection, type ThreadHistoryMeta,} from "./threadHistoryMerge.ts";import { ThreadSnapshotLoader, type ThreadSnapshotLoadResult } from "./threadSnapshotHttp.ts";import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState, type EnvironmentThreadStatus,} from "./threadState.ts"; function statusWithoutLiveData( data: Option.Option<OrchestrationV2ThreadProjection>,): EnvironmentThreadStatus { return Option.isSome(data) ? "cached" : "empty";} function formatThreadError(cause: Cause.Cause<unknown>): string { const error = Cause.squash(cause); return error instanceof Error && error.message.trim().length > 0 ? error.message : "Could not synchronize the thread.";} function formatHistoryError(error: unknown): string { if (error instanceof Error && error.message.trim().length > 0) { return error.message; } return "Could not load earlier activity.";} function historyMetaFromCachedSnapshot( snapshot: OrchestrationV2ThreadDetailSnapshot,): ThreadHistoryMeta { const historyCursor = snapshot.historyCursor ?? null; const hasMoreHistory = snapshot.hasMoreHistory ?? false; return { historyCursor, hasMoreHistory, loading: false, error: null, // Cache never stores expanded progressive history (load-earlier growth). expanded: false, latestLocalTurnOrdinal: snapshot.latestLocalTurnOrdinal ?? null, };} function snapshotToPersist( snapshotSequence: number, projection: OrchestrationV2ThreadProjection, history: ThreadHistoryMeta,): OrchestrationV2ThreadDetailSnapshot { // Persist progressive meta with the bounded window so warm resume restores the // history cursor instead of looking like a complete full-projection cache hit. if (history.hasMoreHistory || history.historyCursor !== null) { return { snapshotSequence, projection, historyCursor: history.historyCursor, hasMoreHistory: history.hasMoreHistory, latestLocalTurnOrdinal: history.latestLocalTurnOrdinal, }; } return { snapshotSequence, projection };} function shouldPersistThread( thread: OrchestrationV2ThreadProjection, history: ThreadHistoryMeta,): boolean { // After the user loads older pages the in-memory timeline can grow large. // Keep those expanded projections out of the monolithic cache. if (history.expanded) { return false; } return !thread.runs.some( (run) => run.status === "preparing" || run.status === "starting" || run.status === "running", );} interface ThreadResumeSnapshot { readonly state: EnvironmentThreadState; readonly sequence: number; readonly persisted: boolean;} interface ThreadResumeCache { snapshot: ThreadResumeSnapshot | undefined; owner: object | undefined;} function matchesThreadSnapshot( current: ThreadResumeSnapshot, projection: OrchestrationV2ThreadProjection | null, sequence: number, history: Pick<ThreadHistoryMeta, "historyCursor" | "hasMoreHistory" | "latestLocalTurnOrdinal">,): boolean { if (current.sequence !== sequence || Option.getOrNull(current.state.data) !== projection) return false; const currentHistory = current.state.history; return ( currentHistory.historyCursor === history.historyCursor && currentHistory.hasMoreHistory === history.hasMoreHistory && (!(history.hasMoreHistory || history.historyCursor !== null) || currentHistory.latestLocalTurnOrdinal === history.latestLocalTurnOrdinal) );} function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadState { return { ...value, status: value.status === "deleted" ? "deleted" : statusWithoutLiveData(value.data), error: Option.none(), history: { ...value.history, loading: false, error: null }, };} export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* ( threadId: ThreadIdType, resumeCache?: ThreadResumeCache,) { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const snapshotLoader = yield* ThreadSnapshotLoader; const historyController = yield* Effect.serviceOption(ThreadHistoryController); const httpClient = yield* Effect.serviceOption(HttpClient.HttpClient); const dpopSigner = yield* Effect.serviceOption(ManagedRelayDpopSigner); const remoteAuthorization = yield* Effect.serviceOption(RemoteEnvironmentAuthorization); const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; const retained = resumeCache?.snapshot; const owner = {}; if (resumeCache) resumeCache.owner = owner; const cached = retained === undefined ? yield* cache.loadThread(environmentId, threadId).pipe( Effect.catch((error) => Effect.logWarning("Could not load cached thread.").pipe( Effect.annotateLogs({ environmentId, threadId, error: error.message, }), Effect.as(Option.none<OrchestrationV2ThreadDetailSnapshot>()), ), ), ) : Option.none<OrchestrationV2ThreadDetailSnapshot>(); const cachedThread = Option.map(cached, (snapshot) => snapshot.projection); const initialState: EnvironmentThreadState = retained ? cachedThreadState(retained.state) : { data: cachedThread, status: statusWithoutLiveData(cachedThread), error: Option.none(), history: Option.match(cached, { onNone: () => EMPTY_THREAD_HISTORY_META, onSome: historyMetaFromCachedSnapshot, }), }; const state = yield* SubscriptionRef.make(initialState); // Seed the resume cursor from the cached snapshot so a warm cache can catch up // via `afterSequence` instead of re-downloading the full thread body. const initialSequence = retained?.sequence ?? Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }); const lastSequence = yield* SubscriptionRef.make(initialSequence); let committed: ThreadResumeSnapshot = { state: initialState, sequence: initialSequence, persisted: retained?.persisted ?? Option.isSome(cached), }; if (resumeCache?.owner === owner) resumeCache.snapshot = committed; const awaitingCompletion = yield* Ref.make(false); const applyLock = yield* Semaphore.make(1); // Save only completed data/cursor updates. A canceled scope must not cache // a cursor whose event has not reached the data yet. const remember = Effect.gen(function* () { const current = yield* SubscriptionRef.get(state); const sequence = yield* SubscriptionRef.get(lastSequence); committed = { state: current, sequence, persisted: committed.persisted && matchesThreadSnapshot(committed, Option.getOrNull(current.data), sequence, current.history), }; if (resumeCache?.owner === owner) resumeCache.snapshot = committed; }); const persistence = yield* Queue.sliding<OrchestrationV2ThreadDetailSnapshot>(1); const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( snapshot: OrchestrationV2ThreadDetailSnapshot, ) { if (resumeCache !== undefined && resumeCache.owner !== owner) return; if ( committed.persisted && matchesThreadSnapshot( committed, snapshot.projection, snapshot.snapshotSequence, historyMetaFromCachedSnapshot(snapshot), ) ) return; yield* cache.saveThread(environmentId, snapshot).pipe( Effect.tap(() => Effect.sync(() => { if ( !matchesThreadSnapshot( committed, snapshot.projection, snapshot.snapshotSequence, historyMetaFromCachedSnapshot(snapshot), ) ) return; committed = { ...committed, persisted: true }; if (resumeCache?.owner === owner) resumeCache.snapshot = committed; }), ), Effect.catch((error) => Effect.logWarning("Could not persist the thread cache.").pipe( Effect.annotateLogs({ environmentId, threadId, error: error.message, }), ), ), ); }); yield* Stream.fromQueue(persistence).pipe( Stream.debounce("500 millis"), Stream.runForEach(persist), Effect.forkScoped, ); const setSynchronizing = SubscriptionRef.update(state, (current) => current.status === "deleted" ? current : { ...current, status: "synchronizing" as const, error: Option.none(), }, ); const setReady = SubscriptionRef.update(state, (current) => current.status === "live" || current.status === "deleted" ? current : { ...current, status: "synchronizing" as const, error: Option.none(), }, ); const setDisconnected = Effect.gen(function* () { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.update(state, (current) => ({ ...current, status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), })); }); const setStreamError = (cause: Cause.Cause<unknown>) => Ref.set(awaitingCompletion, false).pipe( Effect.andThen( SubscriptionRef.update(state, (current) => ({ ...current, status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), error: Option.some(formatThreadError(cause)), })), ), ); const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* ( thread: OrchestrationV2ThreadProjection, options?: { /** Socket/full snapshots: drop progressive meta with the new timeline. */ readonly resetHistory?: boolean; /** * Explicit progressive meta installed atomically with the projection * (bounded HTTP). Wins over resetHistory when both are supplied. */ readonly history?: ThreadHistoryMeta; }, ) { const waiting = yield* Ref.get(awaitingCompletion); // Atomic with concurrent history meta updates: never get-then-set the whole // state when only the projection changes. Bounded installs pass history so // projection + cursor persist together in one enqueue. const next = yield* SubscriptionRef.updateAndGet(state, (previous) => { const history = options?.history !== undefined ? options.history : options?.resetHistory === true ? EMPTY_THREAD_HISTORY_META : previous.history; return { ...previous, data: Option.some(thread), status: waiting ? ("synchronizing" as const) : ("live" as const), error: Option.none(), history, }; }); // Active projections can update many times per second and retain large tool // payloads. Persist once the run settles so cache encoding stays off the // streaming path. Progressive meta rides along when the window is incomplete. if (shouldPersistThread(thread, next.history)) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); yield* Queue.offer(persistence, snapshotToPersist(snapshotSequence, thread, next.history)); } }); const patchHistoryMeta = (patch: (history: ThreadHistoryMeta) => ThreadHistoryMeta) => SubscriptionRef.update(state, (current) => ({ ...current, history: patch(current.history), })); const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.set(state, { data: Option.none(), status: "deleted", error: Option.none(), history: EMPTY_THREAD_HISTORY_META, }); yield* remember; if (resumeCache !== undefined && resumeCache.owner !== owner) return; yield* cache.removeThread(environmentId, threadId).pipe( Effect.catch((error) => Effect.logWarning("Could not remove the cached thread.").pipe( Effect.annotateLogs({ environmentId, threadId, error: error.message, }), ), ), ); }); const applyItemLocked = Effect.fn("EnvironmentThreadState.applyItemLocked")(function* ( item: OrchestrationV2ThreadStreamItem, ) { if (item.kind === "synchronized") { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.update(state, (current) => Option.isSome(current.data) && current.status !== "deleted" ? { ...current, status: "live" as const, error: Option.none() } : current, ); return; } if (item.kind === "snapshot") { yield* SubscriptionRef.set(lastSequence, item.snapshotSequence); // True socket/full snapshots replace the full timeline; progressive cursor // state from a prior bounded window must not stick around. Bounded HTTP // installs call setThread with explicit history and never route here. yield* setThread(item.projection, { resetHistory: true }); return; } const sequence = yield* SubscriptionRef.get(lastSequence); if (item.sequence <= sequence) { return; } yield* SubscriptionRef.set(lastSequence, item.sequence); const waiting = yield* Ref.get(awaitingCompletion); // Apply against the latest projection/history in one update so a concurrent // loadEarlier merge (or history-meta patch) cannot be clobbered by a stale // get-then-set rebuild. type EventApplyResult = | { readonly _tag: "noop" } | { readonly _tag: "delete" } | { readonly _tag: "applied"; readonly projection: OrchestrationV2ThreadProjection; readonly history: ThreadHistoryMeta; }; const result = yield* SubscriptionRef.modify( state, (current): readonly [EventApplyResult, EnvironmentThreadState] => { if (current.status === "deleted") { return [{ _tag: "noop" }, current]; } if (Option.isNone(current.data)) { return [ item.event.type === "thread.deleted" ? { _tag: "delete" } : { _tag: "noop" }, current, ]; } if (item.event.type === "thread.deleted") { return [{ _tag: "delete" }, current]; } // Incomplete progressive windows only (hasMore or open cursor). Do not // use expanded: it remains true after the last page as a cache marker. // Full/web and fully-loaded timelines keep default append-on-miss. const partial = current.history.hasMoreHistory || current.history.historyCursor !== null ? { partialTimeline: true as const, latestLocalTurnOrdinal: current.history.latestLocalTurnOrdinal, } : undefined; const next = applyOrchestrationV2ProjectionEvent(current.data.value, item.event, partial); // True no-op when the reducer deliberately returns the current projection // reference (e.g. dropped old partial-timeline turn-item). Do not clear // stream error/status or enqueue persistence. if (next === null || next === current.data.value) { return [{ _tag: "noop" }, current]; } let history = current.history; if (partial !== undefined && item.event.type === "turn-item.updated") { const ordinal = item.event.payload.ordinal; const watermark = history.latestLocalTurnOrdinal; if (watermark === null || ordinal > watermark) { history = { ...history, latestLocalTurnOrdinal: ordinal }; } } const updated: EnvironmentThreadState = { ...current, data: Option.some(next), status: waiting ? "synchronizing" : "live", error: Option.none(), history, }; return [{ _tag: "applied", projection: next, history: updated.history }, updated]; }, ); if (result._tag === "delete") { yield* setDeleted(); return; } if (result._tag === "applied" && shouldPersistThread(result.projection, result.history)) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); yield* Queue.offer( persistence, snapshotToPersist(snapshotSequence, result.projection, result.history), ); } }); const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( item: OrchestrationV2ThreadStreamItem, ) { yield* applyLock.withPermits(1)(applyItemLocked(item).pipe(Effect.andThen(remember)));Lines 1–500 of 856. Load more
import { ORCHESTRATION_V2_WS_METHODS, type EnvironmentId as EnvironmentIdType, type OrchestrationV2ThreadDetailSnapshot, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadStreamItem, type ThreadId as ThreadIdType, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Scope from "effect/Scope"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { HttpClient } from "effect/unstable/http"; import { Atom } from "effect/unstable/reactivity"; import { RemoteEnvironmentAuthorization } from "../authorization/service.ts"; import { EnvironmentRegistry } from "../connection/registry.ts"; import { connectionProjectionPhase } from "../connection/model.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import * as ConnectionWakeups from "../connection/wakeups.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts"; import { subscribeDynamic } from "../rpc/client.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; import { followStreamInEnvironment } from "./runtime.ts"; import { ThreadHistoryController, type ThreadHistoryLoadEarlierResult, } from "./threadHistoryController.ts"; import { fetchEnvironmentThreadHistoryPage } from "./threadHistoryHttp.ts"; import { applyHistoryPageMeta, clearActiveHistoryLoading, EMPTY_THREAD_HISTORY_META, isActiveHistoryRequestCursor, mergeOlderHistoryIntoProjection, type ThreadHistoryMeta, } from "./threadHistoryMerge.ts"; import { ThreadSnapshotLoader, type ThreadSnapshotLoadResult } from "./threadSnapshotHttp.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState, type EnvironmentThreadStatus, } from "./threadState.ts"; function statusWithoutLiveData( data: Option.Option<OrchestrationV2ThreadProjection>, ): EnvironmentThreadStatus { return Option.isSome(data) ? "cached" : "empty"; } function formatThreadError(cause: Cause.Cause<unknown>): string { const error = Cause.squash(cause); return error instanceof Error && error.message.trim().length > 0 ? error.message : "Could not synchronize the thread."; } function formatHistoryError(error: unknown): string { if (error instanceof Error && error.message.trim().length > 0) { return error.message; } return "Could not load earlier activity."; } function historyMetaFromCachedSnapshot( snapshot: OrchestrationV2ThreadDetailSnapshot, ): ThreadHistoryMeta { const historyCursor = snapshot.historyCursor ?? null; const hasMoreHistory = snapshot.hasMoreHistory ?? false; return { historyCursor, hasMoreHistory, loading: false, error: null, // Cache never stores expanded progressive history (load-earlier growth). expanded: false, latestLocalTurnOrdinal: snapshot.latestLocalTurnOrdinal ?? null, }; } function snapshotToPersist( snapshotSequence: number, projection: OrchestrationV2ThreadProjection, history: ThreadHistoryMeta, ): OrchestrationV2ThreadDetailSnapshot { // Persist progressive meta with the bounded window so warm resume restores the // history cursor instead of looking like a complete full-projection cache hit. if (history.hasMoreHistory || history.historyCursor !== null) { return { snapshotSequence, projection, historyCursor: history.historyCursor, hasMoreHistory: history.hasMoreHistory, latestLocalTurnOrdinal: history.latestLocalTurnOrdinal, }; } return { snapshotSequence, projection }; } function shouldPersistThread( thread: OrchestrationV2ThreadProjection, history: ThreadHistoryMeta, ): boolean { // After the user loads older pages the in-memory timeline can grow large. // Keep those expanded projections out of the monolithic cache. if (history.expanded) { return false; } return !thread.runs.some( (run) => run.status === "preparing" || run.status === "starting" || run.status === "running", ); } interface ThreadResumeSnapshot { readonly state: EnvironmentThreadState; readonly sequence: number; readonly persisted: boolean; } interface ThreadResumeCache { snapshot: ThreadResumeSnapshot | undefined; owner: object | undefined; } function matchesThreadSnapshot( current: ThreadResumeSnapshot, projection: OrchestrationV2ThreadProjection | null, sequence: number, history: Pick<ThreadHistoryMeta, "historyCursor" | "hasMoreHistory" | "latestLocalTurnOrdinal">, ): boolean { if (current.sequence !== sequence || Option.getOrNull(current.state.data) !== projection) return false; const currentHistory = current.state.history; return ( currentHistory.historyCursor === history.historyCursor && currentHistory.hasMoreHistory === history.hasMoreHistory && (!(history.hasMoreHistory || history.historyCursor !== null) || currentHistory.latestLocalTurnOrdinal === history.latestLocalTurnOrdinal) ); } function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadState { return { ...value, status: value.status === "deleted" ? "deleted" : statusWithoutLiveData(value.data), error: Option.none(), history: { ...value.history, loading: false, error: null }, }; } export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* ( threadId: ThreadIdType, resumeCache?: ThreadResumeCache, ) { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const snapshotLoader = yield* ThreadSnapshotLoader; const historyController = yield* Effect.serviceOption(ThreadHistoryController); const httpClient = yield* Effect.serviceOption(HttpClient.HttpClient); const dpopSigner = yield* Effect.serviceOption(ManagedRelayDpopSigner); const remoteAuthorization = yield* Effect.serviceOption(RemoteEnvironmentAuthorization); const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; const retained = resumeCache?.snapshot; const owner = {}; if (resumeCache) resumeCache.owner = owner; const cached = retained === undefined ? yield* cache.loadThread(environmentId, threadId).pipe( Effect.catch((error) => Effect.logWarning("Could not load cached thread.").pipe( Effect.annotateLogs({ environmentId, threadId, error: error.message, }), Effect.as(Option.none<OrchestrationV2ThreadDetailSnapshot>()), ), ), ) : Option.none<OrchestrationV2ThreadDetailSnapshot>(); const cachedThread = Option.map(cached, (snapshot) => snapshot.projection); const initialState: EnvironmentThreadState = retained ? cachedThreadState(retained.state) : { data: cachedThread, status: statusWithoutLiveData(cachedThread), error: Option.none(), history: Option.match(cached, { onNone: () => EMPTY_THREAD_HISTORY_META, onSome: historyMetaFromCachedSnapshot, }), }; const state = yield* SubscriptionRef.make(initialState); // Seed the resume cursor from the cached snapshot so a warm cache can catch up // via `afterSequence` instead of re-downloading the full thread body. const initialSequence = retained?.sequence ?? Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }); const lastSequence = yield* SubscriptionRef.make(initialSequence); let committed: ThreadResumeSnapshot = { state: initialState, sequence: initialSequence, persisted: retained?.persisted ?? Option.isSome(cached), }; if (resumeCache?.owner === owner) resumeCache.snapshot = committed; const awaitingCompletion = yield* Ref.make(false); const applyLock = yield* Semaphore.make(1); // Save only completed data/cursor updates. A canceled scope must not cache // a cursor whose event has not reached the data yet. const remember = Effect.gen(function* () { const current = yield* SubscriptionRef.get(state); const sequence = yield* SubscriptionRef.get(lastSequence); committed = { state: current, sequence, persisted: committed.persisted && matchesThreadSnapshot(committed, Option.getOrNull(current.data), sequence, current.history), }; if (resumeCache?.owner === owner) resumeCache.snapshot = committed; }); const persistence = yield* Queue.sliding<OrchestrationV2ThreadDetailSnapshot>(1); const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( snapshot: OrchestrationV2ThreadDetailSnapshot, ) { if (resumeCache !== undefined && resumeCache.owner !== owner) return; if ( committed.persisted && matchesThreadSnapshot( committed, snapshot.projection, snapshot.snapshotSequence, historyMetaFromCachedSnapshot(snapshot), ) ) return; yield* cache.saveThread(environmentId, snapshot).pipe( Effect.tap(() => Effect.sync(() => { if ( !matchesThreadSnapshot( committed, snapshot.projection, snapshot.snapshotSequence, historyMetaFromCachedSnapshot(snapshot), ) ) return; committed = { ...committed, persisted: true }; if (resumeCache?.owner === owner) resumeCache.snapshot = committed; }), ), Effect.catch((error) => Effect.logWarning("Could not persist the thread cache.").pipe( Effect.annotateLogs({ environmentId, threadId, error: error.message, }), ), ), ); }); yield* Stream.fromQueue(persistence).pipe( Stream.debounce("500 millis"), Stream.runForEach(persist), Effect.forkScoped, ); const setSynchronizing = SubscriptionRef.update(state, (current) => current.status === "deleted" ? current : { ...current, status: "synchronizing" as const, error: Option.none(), }, ); const setReady = SubscriptionRef.update(state, (current) => current.status === "live" || current.status === "deleted" ? current : { ...current, status: "synchronizing" as const, error: Option.none(), }, ); const setDisconnected = Effect.gen(function* () { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.update(state, (current) => ({ ...current, status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), })); }); const setStreamError = (cause: Cause.Cause<unknown>) => Ref.set(awaitingCompletion, false).pipe( Effect.andThen( SubscriptionRef.update(state, (current) => ({ ...current, status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), error: Option.some(formatThreadError(cause)), })), ), ); const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* ( thread: OrchestrationV2ThreadProjection, options?: { /** Socket/full snapshots: drop progressive meta with the new timeline. */ readonly resetHistory?: boolean; /** * Explicit progressive meta installed atomically with the projection * (bounded HTTP). Wins over resetHistory when both are supplied. */ readonly history?: ThreadHistoryMeta; }, ) { const waiting = yield* Ref.get(awaitingCompletion); // Atomic with concurrent history meta updates: never get-then-set the whole // state when only the projection changes. Bounded installs pass history so // projection + cursor persist together in one enqueue. const next = yield* SubscriptionRef.updateAndGet(state, (previous) => { const history = options?.history !== undefined ? options.history : options?.resetHistory === true ? EMPTY_THREAD_HISTORY_META : previous.history; return { ...previous, data: Option.some(thread), status: waiting ? ("synchronizing" as const) : ("live" as const), error: Option.none(), history, }; }); // Active projections can update many times per second and retain large tool // payloads. Persist once the run settles so cache encoding stays off the // streaming path. Progressive meta rides along when the window is incomplete. if (shouldPersistThread(thread, next.history)) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); yield* Queue.offer(persistence, snapshotToPersist(snapshotSequence, thread, next.history)); } }); const patchHistoryMeta = (patch: (history: ThreadHistoryMeta) => ThreadHistoryMeta) => SubscriptionRef.update(state, (current) => ({ ...current, history: patch(current.history), })); const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.set(state, { data: Option.none(), status: "deleted", error: Option.none(), history: EMPTY_THREAD_HISTORY_META, }); yield* remember; if (resumeCache !== undefined && resumeCache.owner !== owner) return; yield* cache.removeThread(environmentId, threadId).pipe( Effect.catch((error) => Effect.logWarning("Could not remove the cached thread.").pipe( Effect.annotateLogs({ environmentId, threadId, error: error.message, }), ), ), ); }); const applyItemLocked = Effect.fn("EnvironmentThreadState.applyItemLocked")(function* ( item: OrchestrationV2ThreadStreamItem, ) { if (item.kind === "synchronized") { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.update(state, (current) => Option.isSome(current.data) && current.status !== "deleted" ? { ...current, status: "live" as const, error: Option.none() } : current, ); return; } if (item.kind === "snapshot") { yield* SubscriptionRef.set(lastSequence, item.snapshotSequence); // True socket/full snapshots replace the full timeline; progressive cursor // state from a prior bounded window must not stick around. Bounded HTTP // installs call setThread with explicit history and never route here. yield* setThread(item.projection, { resetHistory: true }); return; } const sequence = yield* SubscriptionRef.get(lastSequence); if (item.sequence <= sequence) { return; } yield* SubscriptionRef.set(lastSequence, item.sequence); const waiting = yield* Ref.get(awaitingCompletion); // Apply against the latest projection/history in one update so a concurrent // loadEarlier merge (or history-meta patch) cannot be clobbered by a stale // get-then-set rebuild. type EventApplyResult = | { readonly _tag: "noop" } | { readonly _tag: "delete" } | { readonly _tag: "applied"; readonly projection: OrchestrationV2ThreadProjection; readonly history: ThreadHistoryMeta; }; const result = yield* SubscriptionRef.modify( state, (current): readonly [EventApplyResult, EnvironmentThreadState] => { if (current.status === "deleted") { return [{ _tag: "noop" }, current]; } if (Option.isNone(current.data)) { return [ item.event.type === "thread.deleted" ? { _tag: "delete" } : { _tag: "noop" }, current, ]; } if (item.event.type === "thread.deleted") { return [{ _tag: "delete" }, current]; } // Incomplete progressive windows only (hasMore or open cursor). Do not // use expanded: it remains true after the last page as a cache marker. // Full/web and fully-loaded timelines keep default append-on-miss. const partial = current.history.hasMoreHistory || current.history.historyCursor !== null ? { partialTimeline: true as const, latestLocalTurnOrdinal: current.history.latestLocalTurnOrdinal, } : undefined; const next = applyOrchestrationV2ProjectionEvent(current.data.value, item.event, partial); // True no-op when the reducer deliberately returns the current projection // reference (e.g. dropped old partial-timeline turn-item). Do not clear // stream error/status or enqueue persistence. if (next === null || next === current.data.value) { return [{ _tag: "noop" }, current]; } let history = current.history; if (partial !== undefined && item.event.type === "turn-item.updated") { const ordinal = item.event.payload.ordinal; const watermark = history.latestLocalTurnOrdinal; if (watermark === null || ordinal > watermark) { history = { ...history, latestLocalTurnOrdinal: ordinal }; } } const updated: EnvironmentThreadState = { ...current, data: Option.some(next), status: waiting ? "synchronizing" : "live", error: Option.none(), history, }; return [{ _tag: "applied", projection: next, history: updated.history }, updated]; }, ); if (result._tag === "delete") { yield* setDeleted(); return; } if (result._tag === "applied" && shouldPersistThread(result.projection, result.history)) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); yield* Queue.offer( persistence, snapshotToPersist(snapshotSequence, result.projection, result.history), ); } }); const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( item: OrchestrationV2ThreadStreamItem, ) { yield* applyLock.withPermits(1)(applyItemLocked(item).pipe(Effect.andThen(remember)));