diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 870ab31f00003..8f3cb49dc40e2 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -169,6 +169,7 @@ import type { } from "../../lib/threadActivity"; import { PendingApprovalCard } from "./PendingApprovalCard"; import { ComposerErrorNotice } from "./ComposerErrorNotice"; +import { Notice } from "../../components/ui/Notice"; import { ComposerFeedback } from "./ComposerFeedback"; import { ComposerUsageLimits } from "./ComposerUsageLimits"; import { PendingUserInputCard } from "./PendingUserInputCard"; @@ -250,6 +251,7 @@ export interface ThreadDetailScreenProps { readonly respondingUserInputId: RuntimeRequestId | null; readonly connectionStateLabel: EnvironmentConnectionPhase; readonly threadSyncStatus?: EnvironmentThreadStatus; + readonly threadSyncError?: string | null; readonly historyControls?: ThreadFeedHistoryControls; readonly activeThreadBusy: boolean; readonly canStopThread: boolean; @@ -518,6 +520,7 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { const composerKeyboardOffset = composerBottomInset; const contentPresentationKind = props.contentPresentation.kind; const realThreadSyncLabel = (() => { + if (props.threadSyncError != null) return null; switch (props.threadSyncStatus) { case "empty": case "cached": @@ -1545,6 +1548,17 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { onDismiss={() => props.onDismissFeedback(submission.id)} /> ))} + {props.threadSyncError != null && contentPresentationKind === "ready" ? ( + + + + ) : null} {composerError !== null ? ( ; } - // Queued creation displays its prompt before server creation; temporary not-found is expected. - const contentPresentation = - creationState !== null - ? { kind: "ready" as const } - : projectThreadContentPresentation({ - hasDetail: selectedThreadDetail !== null, - detailError: Option.getOrNull(selectedThreadDetailState.error), - detailDeleted: selectedThreadDetailState.status === "deleted", - connectionState: routeConnectionState, - }); + const threadDetailError = Option.getOrNull(selectedThreadDetailState.error); + const contentPresentation = projectThreadContentPresentation({ + hasDetail: selectedThreadDetail !== null, + detailError: threadDetailError, + detailDeleted: selectedThreadDetailState.status === "deleted", + connectionState: routeConnectionState, + hasPendingCreation: creationState !== null, + }); const serverConfig = routeEnvironmentRuntime?.serverConfig ?? null; const renderThreadRouteBody = () => ( <> @@ -1040,6 +1038,7 @@ function ThreadRouteContent( respondingUserInputId={requests.respondingUserInputId} connectionStateLabel={routeConnectionState} threadSyncStatus={selectedThreadDetailState.status} + threadSyncError={threadDetailError} historyControls={historyControls} activeThreadBusy={composer.activeThreadBusy} canStopThread={awaitingBootstrapTurn || composer.interruptibleRunId !== null} diff --git a/apps/mobile/src/features/threads/threadContentPresentation.test.ts b/apps/mobile/src/features/threads/threadContentPresentation.test.ts index 3990799d85925..94467d02167e4 100644 --- a/apps/mobile/src/features/threads/threadContentPresentation.test.ts +++ b/apps/mobile/src/features/threads/threadContentPresentation.test.ts @@ -28,4 +28,44 @@ describe("thread content presentation", () => { detail: "Reconnect this environment to load the conversation.", }); }); + + it("keeps a pending creation's prompt visible before its thread exists", () => { + expect( + projectThreadContentPresentation({ + hasDetail: false, + detailError: null, + detailDeleted: true, + connectionState: "connected", + hasPendingCreation: true, + }), + ).toEqual({ kind: "ready" }); + }); + + it("surfaces a synchronization failure while creation is still pending", () => { + expect( + projectThreadContentPresentation({ + hasDetail: false, + detailError: "Message synchronization timed out.", + detailDeleted: false, + connectionState: "connected", + hasPendingCreation: true, + }), + ).toEqual({ + kind: "unavailable", + title: "Could not load conversation", + detail: "Message synchronization timed out.", + }); + }); + + it("keeps cached messages visible after a synchronization failure during creation", () => { + expect( + projectThreadContentPresentation({ + hasDetail: true, + detailError: "Message synchronization timed out.", + detailDeleted: false, + connectionState: "connected", + hasPendingCreation: true, + }), + ).toEqual({ kind: "ready" }); + }); }); diff --git a/apps/mobile/src/features/threads/threadContentPresentation.ts b/apps/mobile/src/features/threads/threadContentPresentation.ts index a180ed7594db0..81e722584c24c 100644 --- a/apps/mobile/src/features/threads/threadContentPresentation.ts +++ b/apps/mobile/src/features/threads/threadContentPresentation.ts @@ -14,8 +14,11 @@ export function projectThreadContentPresentation(input: { readonly detailError: string | null; readonly detailDeleted: boolean; readonly connectionState: EnvironmentConnectionPhase; + readonly hasPendingCreation?: boolean; }): ThreadContentPresentation { - if (input.hasDetail) { + // The local prompt remains usable before creation reaches the server. A real + // synchronization failure still needs the unavailable state or a cached-detail notice. + if (input.hasDetail || (input.hasPendingCreation && input.detailError === null)) { return { kind: "ready" }; } if (input.detailDeleted) { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index d2a176a444369..5896e12047dbf 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -248,6 +248,7 @@ import { resolveThreadDetailRef, useEnvironmentSupportsServerBrowser, useProject, + useThreadError, useQuietThreadProjection, useThreadUpdatedAt, useThreadLastVisitedAt, @@ -511,8 +512,10 @@ export default function ChatView(props: ChatViewProps) { ? deriveReportedModelSelection(serverProjection) : null; const threadStatus = useThreadStatus(routeThreadDetailRef); + const threadDetailError = useThreadError(routeThreadDetailRef); const threadSyncPhase = resolveThreadSyncPhase({ detailExists: serverProjection !== null, + detailError: threadDetailError, shellExists: serverThread !== null, status: threadStatus, }); @@ -928,7 +931,7 @@ export default function ChatView(props: ChatViewProps) { title: parentSubagentThread?.title ?? "Parent thread", }; const threadError = isServerThread - ? (localServerError ?? serverRuntime?.lastError ?? null) + ? (threadDetailError ?? localServerError ?? serverRuntime?.lastError ?? null) : localDraftError; const threadErrorBannerKey = getThreadErrorBannerKey(routeThreadKey, threadError); const visibleThreadError = shouldShowThreadErrorBanner( diff --git a/apps/web/src/state/entities.ts b/apps/web/src/state/entities.ts index 0f94280768d4b..09646e3f35b24 100644 --- a/apps/web/src/state/entities.ts +++ b/apps/web/src/state/entities.ts @@ -56,6 +56,9 @@ const EMPTY_THREAD_SHELL_ATOM = Atom.make(null).p const EMPTY_THREAD_PROJECTION_ATOM = Atom.make(null).pipe( Atom.withLabel("web-thread-projection:empty"), ); +const EMPTY_THREAD_ERROR_ATOM = Atom.make(null).pipe( + Atom.withLabel("web-thread-error:empty"), +); const EMPTY_THREAD_STATUS_ATOM = Atom.make("empty").pipe( Atom.withLabel("web-thread-status:empty"), ); @@ -189,6 +192,12 @@ export function useThreadStatus(ref: ScopedThreadRef | null): EnvironmentThreadS ); } +export function useThreadError(ref: ScopedThreadRef | null): string | null { + return useAtomValue( + ref === null ? EMPTY_THREAD_ERROR_ATOM : environmentThreadDetails.errorAtom(ref), + ); +} + export function useThreadHistory(ref: ScopedThreadRef | null): ThreadHistoryMeta { return useAtomValue( ref === null ? EMPTY_THREAD_HISTORY_ATOM : environmentThreadDetails.historyAtom(ref), diff --git a/apps/web/src/threadSync.test.ts b/apps/web/src/threadSync.test.ts new file mode 100644 index 0000000000000..f5dbd8b7c6864 --- /dev/null +++ b/apps/web/src/threadSync.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { resolveThreadSyncPhase } from "./threadSync"; + +describe("resolveThreadSyncPhase", () => { + it("stops syncing cached detail on terminal error and resumes when recovery clears it", () => { + const cachedDetail = { + detailExists: true, + shellExists: true, + status: "cached" as const, + detailError: null, + }; + expect(resolveThreadSyncPhase(cachedDetail)).toBe("syncing"); + + const failedDetail = { + ...cachedDetail, + detailError: + "Message synchronization timed out. Reconnect or return to the app to try again.", + }; + expect.soft(resolveThreadSyncPhase(failedDetail)).toBeNull(); + + const recoveringDetail = { + ...failedDetail, + status: "synchronizing" as const, + detailError: null, + }; + expect(resolveThreadSyncPhase(recoveringDetail)).toBe("syncing"); + }); +}); diff --git a/apps/web/src/threadSync.ts b/apps/web/src/threadSync.ts index a8b5446add2c5..dcc8674278640 100644 --- a/apps/web/src/threadSync.ts +++ b/apps/web/src/threadSync.ts @@ -4,10 +4,11 @@ export type ThreadSyncPhase = "loading" | "syncing"; export function resolveThreadSyncPhase(input: { readonly detailExists: boolean; + readonly detailError?: string | null; readonly shellExists: boolean; readonly status: EnvironmentThreadStatus; }): ThreadSyncPhase | null { - if (!input.shellExists) { + if (!input.shellExists || input.detailError != null) { return null; } diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 2832a4e231556..f194851919ae1 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -236,6 +236,10 @@ export function runStreamGuarded( } interface SubscriptionOptions { + readonly transform?: ( + stream: Stream.Stream, EnvironmentRpcStreamFailure>, + input: EnvironmentRpcInput, + ) => Stream.Stream, EnvironmentRpcStreamFailure>; readonly onDefect?: ( cause: Cause.Cause>, ) => Effect.Effect; @@ -299,7 +303,11 @@ function subscribeDynamicMapped( method: tag, input, }); - const stream = mapStream(session, method(input)).pipe( + const source = method(input); + const stream = mapStream( + session, + options?.transform?.(source, input) ?? source, + ).pipe( Stream.onFirst(() => Effect.sync(() => { expectedFailureRetries = 0; diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index eaf1751a1c2d2..6eb8eb1bf9b3b 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -1,4 +1,5 @@ import { + EnvironmentAuthorizationError, EnvironmentId, EventId, MessageId, @@ -13,6 +14,7 @@ import { type OrchestrationV2TurnItem, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -79,7 +81,11 @@ const BASE_PROJECTION: OrchestrationV2ThreadProjection = { thread: { ...v2Projection.thread, title: "Cached thread" }, }; -type TestThreadInput = OrchestrationV2ThreadStreamItem | Error; +class TestStreamInterrupted { + readonly _tag = "TestStreamInterrupted"; +} + +type TestThreadInput = OrchestrationV2ThreadStreamItem | Error | TestStreamInterrupted; function testSession(client: WsRpcProtocolClient): RpcSession.RpcSession { return { @@ -124,6 +130,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o }) => Effect.Effect<{ readonly item: OrchestrationV2TurnItem | null }>; }) { const inputs = yield* Queue.unbounded(); + const subscribed = yield* Queue.unbounded<{ readonly afterSequence?: number }>(); const observed = yield* Queue.unbounded(); const latest = yield* Ref.make(EMPTY_ENVIRONMENT_THREAD_STATE); const retryCount = yield* Ref.make(0); @@ -142,7 +149,11 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const streamFrom = (queue: Queue.Queue) => Stream.fromQueue(queue).pipe( Stream.mapEffect((input) => - input instanceof Error ? Effect.fail(input) : Effect.succeed(input), + input instanceof Error + ? Effect.fail(input) + : input instanceof TestStreamInterrupted + ? Effect.failCause(Cause.interrupt(9999)) + : Effect.succeed(input), ), ); const client = { @@ -160,6 +171,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o Effect.andThen(Ref.set(lastAcceptBoundedSnapshot, input.acceptBoundedSnapshot)), Effect.andThen(Ref.set(lastAcceptCompactTurnItems, input.acceptCompactTurnItems)), Effect.andThen(SubscriptionRef.update(subscriptionCount, (count) => count + 1)), + Effect.andThen(Queue.offer(subscribed, input)), Effect.as( options?.completionMarker === true ? streamFrom(inputs) @@ -276,6 +288,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o historyController.loadEarlier(TARGET.environmentId, THREAD_ID, throughEntryId), reconcilePendingWork: historyController.reconcilePendingWork(TARGET.environmentId, THREAD_ID), inputs, + subscribed, observed, latest, retryCount, @@ -2082,6 +2095,163 @@ describe("EnvironmentThreads", () => { }), ); + it.effect( + "keeps stream errors visible through expected-failure retries until the completion marker", + () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); + yield* Queue.take(harness.subscribed); + yield* Queue.offerAll(harness.inputs, [snapshot(BASE_PROJECTION), synchronized()]); + yield* awaitThreadState(harness.observed, (state) => state.status === "live"); + + yield* Queue.offer(harness.inputs, new Error("stream failed")); + yield* awaitThreadState(harness.observed, (state) => Option.isSome(state.error)); + yield* TestClock.adjust("250 millis"); + yield* Queue.take(harness.subscribed); + const retrying = yield* SubscriptionRef.get(harness.threadState); + expect(Option.getOrUndefined(retrying.error)).toBe("stream failed"); + expect(Option.getOrThrow(retrying.data)).toEqual(BASE_PROJECTION); + + yield* Queue.offer(harness.inputs, synchronized()); + const recovered = yield* awaitThreadState( + harness.observed, + (state) => state.status === "live" && Option.isNone(state.error), + ); + expect(Option.getOrThrow(recovered.data)).toEqual(BASE_PROJECTION); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(2); + }), + ); + + it.effect("retries a missing completion marker after 15 seconds despite replay traffic", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); + expect((yield* Queue.take(harness.subscribed)).afterSequence).toBe(CACHED_SNAPSHOT_SEQUENCE); + yield* TestClock.adjust("14 seconds"); + yield* Queue.offer(harness.inputs, titleUpdated("Still replaying", 8)); + const replaying = yield* awaitThreadState( + harness.observed, + (state) => Option.getOrNull(state.data)?.thread.title === "Still replaying", + ); + expect(replaying.status).toBe("synchronizing"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(1); + + yield* TestClock.adjust("1 second"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(2); + expect((yield* Queue.take(harness.subscribed)).afterSequence).toBeUndefined(); + const recoveredProjection = { + ...BASE_PROJECTION, + thread: { ...BASE_PROJECTION.thread, title: "Recovered snapshot" }, + }; + yield* Queue.offerAll(harness.inputs, [snapshot(recoveredProjection, 12), synchronized()]); + const live = yield* awaitThreadState(harness.observed, (state) => state.status === "live"); + expect(Option.getOrThrow(live.data)).toEqual(recoveredProjection); + expect(Option.isNone(live.error)).toBe(true); + yield* TestClock.adjust("1 minute"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(2); + + yield* harness.replaceSession; + expect((yield* Queue.take(harness.subscribed)).afterSequence).toBe(12); + }), + ); + + it.effect.each(["session", "foreground"] as const)( + "retains cached messages after both deadlines until a new %s trigger recovers", + (trigger) => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); + yield* Queue.take(harness.subscribed); + yield* TestClock.adjust("15 seconds"); + expect((yield* Queue.take(harness.subscribed)).afterSequence).toBeUndefined(); + yield* TestClock.adjust("15 seconds"); + const failed = yield* awaitThreadState(harness.observed, (state) => + Option.isSome(state.error), + ); + expect(failed.status).toBe("cached"); + expect(Option.getOrThrow(failed.data)).toEqual(BASE_PROJECTION); + expect(Option.getOrThrow(failed.error)).toMatch(/synchronization timed out/i); + yield* TestClock.adjust("10 minutes"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(2); + + if (trigger === "session") yield* harness.replaceSession; + else yield* Queue.offer(harness.wakeups, "application-active"); + expect((yield* Queue.take(harness.subscribed)).afterSequence).toBe( + CACHED_SNAPSHOT_SEQUENCE, + ); + yield* Queue.offerAll(harness.inputs, [snapshot(BASE_PROJECTION, 12), synchronized()]); + const live = yield* awaitThreadState(harness.observed, (state) => state.status === "live"); + expect(Option.isNone(live.error)).toBe(true); + yield* TestClock.adjust("1 minute"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(3); + }), + ); + + it.effect( + "retries a failed replay with one snapshot and retains messages if that also fails", + () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); + yield* Queue.take(harness.subscribed); + yield* Queue.offer(harness.inputs, new Error("Replay could not decode")); + expect((yield* Queue.take(harness.subscribed)).afterSequence).toBeUndefined(); + yield* Queue.offer(harness.inputs, new Error("Snapshot could not load")); + const failed = yield* awaitThreadState(harness.observed, (state) => + Option.isSome(state.error), + ); + expect(failed.status).toBe("cached"); + expect(Option.getOrThrow(failed.data)).toEqual(BASE_PROJECTION); + yield* TestClock.adjust("1 minute"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(2); + }), + ); + + it.effect("recovers when only the remote resumed request is interrupted", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); + yield* Queue.take(harness.subscribed); + yield* Queue.offer(harness.inputs, new TestStreamInterrupted()); + yield* TestClock.adjust("15 seconds"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(2); + expect((yield* Queue.take(harness.subscribed)).afterSequence).toBeUndefined(); + yield* Queue.offerAll(harness.inputs, [snapshot(BASE_PROJECTION, 12), synchronized()]); + yield* awaitThreadState(harness.observed, (state) => state.status === "live"); + }), + ); + + it.effect("does not retry a canceled thread scope after its marker deadline", () => + Effect.gen(function* () { + const scope = yield* Effect.acquireRelease(Scope.make(), (scope) => + Scope.close(scope, Exit.void), + ); + const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }).pipe( + Scope.provide(scope), + ); + yield* Queue.take(harness.subscribed); + yield* Scope.close(scope, Exit.void); + yield* TestClock.adjust("1 minute"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(1); + }), + ); + + it.effect("does not bypass a resume authorization failure with a snapshot retry", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); + yield* Queue.take(harness.subscribed); + yield* Queue.offer( + harness.inputs, + new EnvironmentAuthorizationError({ + message: "Missing scope", + requiredScope: "orchestration:read", + }), + ); + const failed = yield* awaitThreadState(harness.observed, (state) => + Option.isSome(state.error), + ); + expect(Option.getOrThrow(failed.error)).toBe("Missing scope"); + yield* TestClock.adjust("1 minute"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(1); + }), + ); + it.effect("keeps replayed updates synchronizing until the completion marker arrives", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); @@ -2089,6 +2259,7 @@ describe("EnvironmentThreads", () => { harness.observed, (value) => value.status === "synchronizing" && Option.isSome(value.data), ); + yield* Queue.take(harness.subscribed); expect(yield* Ref.get(harness.lastRequestCompletionMarker)).toBe(true); yield* Queue.offer( diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index c7b22d9dcdff0..53f785a3317d9 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -1,4 +1,5 @@ import { + EnvironmentAuthorizationError, isOrchestrationV2WorkActive, ORCHESTRATION_V2_WS_METHODS, type EnvironmentId as EnvironmentIdType, @@ -11,6 +12,7 @@ import { import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot"; import { pendingBackgroundTurnItems } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as Cause from "effect/Cause"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Scope from "effect/Scope"; @@ -18,6 +20,7 @@ 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 Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -32,7 +35,14 @@ import * as ConnectionWakeups from "../connection/wakeups.ts"; import * as Persistence from "../platform/persistence.ts"; import { runCachePersistence } from "./cachePersistence.ts"; import * as ManagedRelay from "../relay/managedRelay.ts"; -import { request, subscribeDynamic } from "../rpc/client.ts"; +import { + isRpcClientError, + request, + subscribeDynamicWithSession, + type EnvironmentRpcStreamFailure, + type EnvironmentRpcStreamValue, +} from "../rpc/client.ts"; +import type { RpcSession } from "../rpc/session.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; @@ -61,6 +71,14 @@ function statusWithoutLiveData( return Option.isSome(data) ? "cached" : "empty"; } +const isEnvironmentAuthorizationError = Schema.is(EnvironmentAuthorizationError); + +class ThreadSynchronizationTimeout extends Schema.TaggedError()( + "ThreadSynchronizationTimeout", + {}, +) {} +const isThreadSynchronizationTimeout = Schema.is(ThreadSynchronizationTimeout); + function formatThreadError(cause: Cause.Cause): string { const error = Cause.squash(cause); return error instanceof Error && error.message.trim().length > 0 @@ -392,6 +410,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make })), ), ); + const clearStreamError = SubscriptionRef.update(state, (current) => + Option.isNone(current.error) ? current : { ...current, error: Option.none() }, + ); const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* ( thread: OrchestrationV2ThreadProjection, @@ -638,10 +659,16 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }); const applyItems = Effect.fn("EnvironmentThreadState.applyItems")(function* ( - items: ReadonlyArray, + frames: ReadonlyArray, ) { yield* applyLock.withPermits(1)( Effect.gen(function* () { + const currentSession = yield* SubscriptionRef.get(supervisor.session); + if (Option.isNone(currentSession)) return; + const items = frames + .filter(([producerSession]) => producerSession === currentSession.value) + .map(([, item]) => item); + if (items.length === 0) return; let events: SequencedItem[] = []; for (const item of items) { if ( @@ -925,18 +952,27 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const resumingLive = yield* Ref.make(initialState.status === "live"); const markSynchronizing = Effect.gen(function* () { if (yield* Ref.get(resumingLive)) return; - yield* SubscriptionRef.update(state, (current) => - current.status === "deleted" - ? current - : { ...current, status: "synchronizing" as const, error: Option.none() }, - ); + yield* setConnecting; }); yield* markSynchronizing; - yield* Effect.forkScoped( - subscribeDynamic( + const subscribeThread = ( + retrySnapshot = false, + ): Stream.Stream< + readonly [ + RpcSession, + EnvironmentRpcStreamValue, + ], + EnvironmentRpcStreamFailure, + EnvironmentSupervisor.EnvironmentSupervisor + > => { + let failedSession: RpcSession | undefined; + let resumeFailed = false; + return subscribeDynamicWithSession( ORCHESTRATION_V2_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { + failedSession = session; + resumeFailed = false; let current = yield* SubscriptionRef.get(state); // A prior definitive miss (or delete event) already cleared this thread. // Park the subscription attempt without opening the socket so we do not @@ -989,6 +1025,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make latestLocalTurnOrdinal: httpResult.history.latestLocalTurnOrdinal ?? null, } : EMPTY_THREAD_HISTORY_META; + yield* clearStreamError; yield* setThread(httpResult.snapshot.projection, { history }); yield* remember; }), @@ -1007,7 +1044,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make } const sequence = yield* SubscriptionRef.get(lastSequence); - const canResume = Option.isSome(current.data); + const canResume = Option.isSome(current.data) && !retrySnapshot; const acceptBoundedSnapshot = yield* Ref.get(acceptsBoundedSocketSnapshots); return { @@ -1026,13 +1063,123 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }; }), { - onDefect: () => setStreamError("Could not synchronize the thread."), + transform: (stream, input) => { + const resumed = input.afterSequence !== undefined; + const snapshotRetry = retrySnapshot; + retrySnapshot = false; + return Stream.unwrap( + Effect.gen(function* () { + const completed = yield* Deferred.make(); + let synchronized = false; + const needsMarker = + (resumed || snapshotRetry) && input.requestCompletionMarker === true; + const observed = stream.pipe( + Stream.mapArrayEffect((items) => { + const complete = items.some((item) => + input.requestCompletionMarker === true + ? item.kind === "synchronized" + : item.kind === "snapshot", + ); + if (!complete) return Effect.succeed(items); + synchronized = true; + // Clear on receipt so applying buffered frames cannot erase a later failure. + return Deferred.succeed(completed, undefined).pipe( + Effect.andThen(clearStreamError), + Effect.as(items), + ); + }), + ); + const watched = needsMarker + ? observed.pipe( + // Remote interruption ends only the request. Local scope cancellation + // also interrupts the watchdog, so it cannot trigger a retry. + Stream.catchCause((cause) => + !synchronized && cause.reasons.every((reason) => reason._tag === "Interrupt") + ? Stream.never + : Stream.failCause(cause), + ), + Stream.concat( + Stream.suspend(() => + synchronized + ? Stream.empty + : Stream.die(new ThreadSynchronizationTimeout()), + ), + ), + // A single deadline bounds all replay traffic until the marker. + Stream.interruptWhen( + Deferred.await(completed).pipe( + Effect.timeoutOrElse({ + duration: "15 seconds", + orElse: () => Effect.die(new ThreadSynchronizationTimeout()), + }), + Effect.andThen(Effect.never), + ), + ), + ) + : observed; + return watched.pipe( + Stream.catchCause((cause) => { + if ( + Cause.hasInterrupts(cause) || + cause.reasons.some( + (reason) => + reason._tag === "Fail" && isEnvironmentAuthorizationError(reason.error), + ) + ) + return Stream.failCause(cause); + const failed = cause.reasons.some( + (reason) => + reason._tag === "Die" || + (reason._tag === "Fail" && + (isRpcClientError(reason.error) + ? reason.error.reason._tag === "RpcClientDefect" + : (needsMarker || snapshotRetry) && !synchronized)), + ); + // Transport and authorization failures keep the RPC subscription's + // existing handling; only a failed catch-up gets a snapshot retry. + if (!failed) return Stream.failCause(cause); + resumeFailed = resumed && !synchronized; + return Stream.die(Cause.squash(cause)); + }), + ); + }), + ); + }, + onDefect: (cause) => + resumeFailed + ? Effect.void + : setStreamError( + cause.reasons.some( + (reason) => + reason._tag === "Die" && isThreadSynchronizationTimeout(reason.defect), + ) + ? "Message synchronization timed out. Reconnect or return to the app to try again." + : "Could not synchronize the thread.", + ), onExpectedFailure: (cause) => setStreamError(formatThreadError(cause)), retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEachArray(applyItems)), - ); + ).pipe( + Stream.catchCause((cause) => { + if (Cause.hasInterrupts(cause)) return Stream.failCause(cause); + if (resumeFailed) return subscribeThread(true); + // Running threads stay mounted. Keep the error until a new session or + // foreground wakeup can start another catch-up attempt. + return Stream.merge( + SubscriptionRef.changes(supervisor.session).pipe( + Stream.filter((session) => Option.isSome(session) && session.value !== failedSession), + ), + foregroundResubscriptions, + ).pipe( + Stream.take(1), + Stream.drain, + Stream.concat(Stream.suspend(() => subscribeThread())), + ); + }), + ); + }; + yield* Effect.forkScoped(subscribeThread().pipe(Stream.runForEachArray(applyItems))); return state; });