diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 3e7eec1449e2..c9e899637d86 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -269,6 +269,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; @@ -567,6 +568,7 @@ function ThreadDetailBody( const composerKeyboardOffset = composerBottomInset; const contentPresentationKind = props.contentPresentation.kind; const realThreadSyncLabel = (() => { + if (props.threadSyncError != null) return null; switch (props.threadSyncStatus) { case "empty": case "cached": @@ -1686,6 +1688,17 @@ function ThreadDetailBody( onDismiss={() => props.onDismissFeedback(submission.id)} /> ))} + {props.threadSyncError != null && contentPresentationKind === "ready" ? ( + + + + ) : null} {freshSessionAvailable ? ( ) : null} diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index 6205fd128179..06ddea17ee32 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -952,16 +952,14 @@ function ThreadRouteContent( return ; } - // 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 3990799d8592..94467d02167e 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 a180ed7594db..81e722584c24 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/server/src/cloud/bootService.test.ts b/apps/server/src/cloud/bootService.test.ts index c9f7d1a77c3d..c522aade836c 100644 --- a/apps/server/src/cloud/bootService.test.ts +++ b/apps/server/src/cloud/bootService.test.ts @@ -360,6 +360,7 @@ const makeHarness = Effect.fn("test.make_boot_service_harness")(function* ( timeouts, control, runtime, + runner, }; }); @@ -680,7 +681,7 @@ it.layer(NodeServices.layer)("boot service install", (it) => { : setting === "manager-only" ? path.join(configRoot, "config root") : undefined; - const { service, makeService, home, statePath, timeouts, runtime, control } = + const { service, makeService, home, statePath, timeouts, runtime, control, runner } = yield* makeHarness("linux", undefined, xdgConfigHome, managerXdgConfigHome); const unitPath = path.join( setting === "relocated" || setting === "manager-only" || setting === "manager-unavailable" @@ -697,6 +698,17 @@ it.layer(NodeServices.layer)("boot service install", (it) => { control.failCommand = "busctl --user --json=short get-property org.freedesktop.systemd1 /org/freedesktop/systemd1 org.freedesktop.systemd1.Manager UnitPath"; } + expect( + yield* BootService.bootServiceUnitPath().pipe( + Effect.provideService(ProcessRunner.ProcessRunner, runner), + Effect.provideService(HostProcess.Platform, "linux"), + Effect.provideService(HostProcess.HomeDirectory, home), + Effect.provideService( + HostProcess.Environment, + xdgConfigHome === undefined ? {} : { XDG_CONFIG_HOME: xdgConfigHome }, + ), + ), + ).toBe(unitPath); const discoveredService = yield* makeService(); expect(parseServiceState(yield* fs.readFileString(statePath))).toEqual({ diff --git a/apps/web/src/components/ChatView.syncErrors.browser.test.tsx b/apps/web/src/components/ChatView.syncErrors.browser.test.tsx new file mode 100644 index 000000000000..f1e70132cd84 --- /dev/null +++ b/apps/web/src/components/ChatView.syncErrors.browser.test.tsx @@ -0,0 +1,172 @@ +import { createMemoryHistory } from "@tanstack/react-router"; +import { + EnvironmentAuthorizationError, + ORCHESTRATION_V2_WS_METHODS, + OrchestrationV2DispatchCommandError, +} from "@t3tools/contracts"; +import { DEFAULT_CLIENT_SETTINGS } from "@t3tools/contracts/settings"; +import type * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Stream from "effect/Stream"; +import { expect, it, vi } from "vite-plus/test"; +import { page } from "vite-plus/test/browser"; + +import { AppRoot } from "~/AppRoot"; +import "~/components/ChatView"; +import { __setClientSettingsForTests } from "~/hooks/useSettings"; +import { getRouter } from "~/router"; +import { appAtomRegistry } from "~/rpc/atomRegistry"; +import { environmentThreadDetails } from "~/state/threads"; +import { + ENVIRONMENT_ID, + fakeDriver, + installFakeEnvironment, + threadWithHistory, + type FakeDriver, +} from "../../test/fakeEnvironment"; +import { act, render } from "../../test/render"; + +vi.mock("../../../../packages/client-runtime/src/connection/driver.ts", async (importOriginal) => { + const [actual, { fakeDriver }, Layer] = await Promise.all([ + importOriginal<{ readonly ConnectionDriver: Context.Key }>(), + import("../../test/fakeEnvironment"), + import("effect/Layer"), + ]); + return { ...actual, layer: Layer.succeed(actual.ConnectionDriver, fakeDriver) }; +}); + +vi.mock(import("~/rpc/requestLatencyState"), async (original) => ({ + ...(await original()), + trackRpcRequestSent: () => {}, +})); + +// The environment session outlives a render, so its failing subscriptions need their own file. +it("keeps synchronization failures separate from local errors and first-load send eligibility", async () => { + const cold = threadWithHistory("cold-sync-failure", 1); + const warm = threadWithHistory("warm-sync-failure", 1); + using _environment = installFakeEnvironment([cold, warm]); + const firstFailure = Deferred.makeUnsafe(); + const nextFailure = Deferred.makeUnsafe(); + let warmSubscriptions = 0; + const connect = fakeDriver.connect; + using _connect = vi.spyOn(fakeDriver, "connect").mockImplementation((entry) => + connect(entry).pipe( + Effect.map((lease) => ({ + ...lease, + session: { + ...lease.session, + client: new Proxy(lease.session.client, { + get(target, method, receiver) { + if (method === ORCHESTRATION_V2_WS_METHODS.subscribeThread) { + return ( + input: Parameters< + (typeof target)[typeof ORCHESTRATION_V2_WS_METHODS.subscribeThread] + >[0], + ) => { + if (input.threadId === cold.id) { + return Stream.fail( + new EnvironmentAuthorizationError({ + message: "Thread history unavailable", + requiredScope: "orchestration:read", + }), + ); + } + warmSubscriptions += 1; + const initial = + warmSubscriptions === 1 + ? target[ORCHESTRATION_V2_WS_METHODS.subscribeThread](input).pipe( + Stream.take(2), + ) + : Stream.empty; + return initial.pipe( + Stream.concat( + Stream.fromEffect( + Deferred.await(warmSubscriptions === 1 ? firstFailure : nextFailure).pipe( + Effect.andThen(Effect.die(new Error("Broken thread stream"))), + ), + ), + ), + ); + }; + } + if (method === ORCHESTRATION_V2_WS_METHODS.dispatchCommand) { + return ( + command: Parameters< + (typeof target)[typeof ORCHESTRATION_V2_WS_METHODS.dispatchCommand] + >[0], + ) => + Effect.fail( + new OrchestrationV2DispatchCommandError({ + commandId: command.commandId, + commandType: command.type, + message: "Local send failed", + }), + ); + } + return Reflect.get(target, method, receiver); + }, + }), + }, + })), + ), + ); + __setClientSettingsForTests({ + ...DEFAULT_CLIENT_SETTINGS, + onboardingCompletedAt: "2026-01-01T00:00:00.000Z", + }); + const router = getRouter( + createMemoryHistory({ initialEntries: [`/${ENVIRONMENT_ID}/${cold.id}`] }), + ); + await router.load(); + await using _app = await render(, { + // AppRoot brings its own router. + router: false, + }); + await expect.element(page.getByText("Thread history unavailable", { exact: true })).toBeVisible(); + await page.getByTestId("composer-editor").fill("Do not send before history arrives"); + await expect + .element(page.getByRole("button", { name: /^(Submit message|Messages loading)$/ })) + .toBeDisabled(); + + await act(() => + router.navigate({ + to: "/$environmentId/$threadId", + params: { environmentId: ENVIRONMENT_ID, threadId: warm.id }, + }), + ); + await expect.element(page.getByText(`Answer 0 for ${warm.id}.`)).toBeVisible(); + await act(() => Deferred.doneUnsafe(firstFailure, Effect.void)); + const syncFailure = page.getByText("Could not synchronize the thread.", { exact: true }); + await expect.element(syncFailure).toBeVisible(); + const threadState = environmentThreadDetails.stateAtom({ + environmentId: ENVIRONMENT_ID, + threadId: warm.id, + }); + const firstSyncFailure = appAtomRegistry.get(threadState).error; + await page.getByTestId("composer-editor").fill("Retry the work"); + await page.getByRole("button", { name: "Submit message", exact: true }).click(); + await expect.element(page.getByTestId("composer-editor")).toHaveTextContent("Retry the work"); + await expect.element(page.getByText("Local send failed", { exact: true })).toBeVisible(); + await page.getByRole("button", { name: "Dismiss error" }).click(); + await expect + .element(page.getByText("Local send failed", { exact: true })) + .not.toBeInTheDocument(); + + await expect.element(syncFailure).toBeVisible(); + await page.getByRole("button", { name: "Dismiss error" }).click(); + await expect.element(syncFailure).not.toBeInTheDocument(); + + document.dispatchEvent(new Event("visibilitychange")); + await vi.waitFor(() => expect(warmSubscriptions).toBe(2)); + await act(() => Deferred.doneUnsafe(nextFailure, Effect.void)); + await vi.waitFor(() => { + const failure = appAtomRegistry.get(threadState).error; + expect(failure).toMatchObject({ _tag: "Some", value: "Could not synchronize the thread." }); + expect(failure).not.toBe(firstSyncFailure); + }); + await expect.element(syncFailure).not.toBeInTheDocument(); + await expect + .element(page.getByText("Local send failed", { exact: true })) + .not.toBeInTheDocument(); +}); diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 906b83e9838c..04661a17a53e 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -258,6 +258,7 @@ import { resolveThreadDetailRef, useEnvironmentSupportsServerBrowser, useProject, + useThreadError, useQuietThreadProjection, useThreadUpdatedAt, useThreadLastVisitedAt, @@ -311,11 +312,7 @@ import { getProviderStatusBannerKey, shouldShowProviderStatusBanner, } from "./chat/ProviderStatusBanner"; -import { - getThreadErrorBannerKey, - isThreadErrorBannerDismissedForSession, - shouldShowThreadErrorBanner, -} from "./chat/ThreadErrorBanner"; +import { resolveThreadErrorBanner } from "./chat/ThreadErrorBanner"; import { QueuedRunsControl, type QueuedRunsControlHandle } from "./chat/QueuedRunsControl"; import { useThreadLinkedPullRequest } from "./pullRequest/useThreadLinkedPullRequest"; import { useProactivePanels } from "../hooks/useProactivePanels"; @@ -520,12 +517,19 @@ 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, }); - const threadDetailLoading = threadSyncPhase === "loading"; + const threadDetailLoading = + resolveThreadSyncPhase({ + detailExists: serverProjection !== null, + shellExists: serverThread !== null, + status: threadStatus, + }) === "loading"; // Held back briefly, so a quick sync does not flash a status line at the end of the timeline. const delayedThreadSyncPhase = useDelayedStatus(routeThreadKey, threadSyncPhase); const activeProviderThread = serverProjection?.providerThreads.find( @@ -941,19 +945,18 @@ export default function ChatView(props: ChatViewProps) { title: parentSubagentThread?.title ?? "Parent thread", }; const threadError = isServerThread - ? (localServerError ?? serverRuntime?.lastError ?? null) + ? (localServerError ?? threadDetailError ?? serverRuntime?.lastError ?? null) : localDraftError; - const threadErrorBannerKey = getThreadErrorBannerKey(routeThreadKey, threadError); - const visibleThreadError = shouldShowThreadErrorBanner( - routeThreadKey, - threadError, - isThreadErrorBannerDismissedForSession(threadErrorBannerKey), - ) - ? threadError - : null; + const threadErrorBanner = resolveThreadErrorBanner({ + threadKey: routeThreadKey, + detailError: isServerThread ? threadDetailError : null, + localError: isServerThread ? localServerError : localDraftError, + runtimeError: isServerThread ? (serverRuntime?.lastError ?? null) : null, + }); + const threadErrorBannerKey = threadErrorBanner?.key ?? null; + const visibleThreadError = threadErrorBanner?.message ?? null; const [, setThreadErrorBannerDismissTick] = useState(0); - const threadErrorIsServerRuntimeError = - localServerError === null && visibleThreadError === serverRuntime?.lastError; + const threadErrorIsServerRuntimeError = threadErrorBanner?.source === "runtime"; const defaultRuntimeMode = resolveProjectSettings(settings, activeThread?.projectId ?? null) .settings.defaultRuntimeMode; const runtimeMode = @@ -3389,6 +3392,7 @@ export default function ChatView(props: ChatViewProps) { threadErrorIsServerRuntimeError ? (serverRuntime?.lastErrorCode ?? null) : null } threadErrorBannerKey={threadErrorBannerKey} + threadErrorIsLocal={threadErrorBanner?.source === "local"} setThreadError={setThreadError} setThreadErrorBannerDismissTick={setThreadErrorBannerDismissTick} /> diff --git a/apps/web/src/components/chat/ChatBannerLane.tsx b/apps/web/src/components/chat/ChatBannerLane.tsx index 8d8a6cb01b83..e36b27396398 100644 --- a/apps/web/src/components/chat/ChatBannerLane.tsx +++ b/apps/web/src/components/chat/ChatBannerLane.tsx @@ -24,6 +24,7 @@ export function ChatBannerLane(props: { threadErrorClass: OrchestrationV2ProviderFailureClass | null; threadErrorCode: string | null; threadErrorBannerKey: string | null; + threadErrorIsLocal: boolean; setThreadError: (threadId: ThreadId | null, error: string | null) => void; setThreadErrorBannerDismissTick: Dispatch>; }) { @@ -37,6 +38,7 @@ export function ChatBannerLane(props: { threadErrorClass, threadErrorCode, threadErrorBannerKey, + threadErrorIsLocal, setThreadError, setThreadErrorBannerDismissTick, } = props; @@ -53,7 +55,7 @@ export function ChatBannerLane(props: { errorClass={threadErrorClass} errorCode={threadErrorCode} onDismiss={() => { - setThreadError(threadId, null); + if (threadErrorIsLocal) setThreadError(threadId, null); dismissThreadErrorBannerForSession(threadErrorBannerKey); setThreadErrorBannerDismissTick((tick) => tick + 1); }} diff --git a/apps/web/src/components/chat/ThreadErrorBanner.tsx b/apps/web/src/components/chat/ThreadErrorBanner.tsx index 7c8f4b7fee78..002e7f2717f2 100644 --- a/apps/web/src/components/chat/ThreadErrorBanner.tsx +++ b/apps/web/src/components/chat/ThreadErrorBanner.tsx @@ -60,15 +60,45 @@ export function shouldShowThreadErrorBanner( const sessionDismissedThreadErrorBannerKeys = new Set(); export function dismissThreadErrorBannerForSession(bannerKey: string | null): void { - if (bannerKey !== null) { - sessionDismissedThreadErrorBannerKeys.add(bannerKey); - } + if (bannerKey !== null) sessionDismissedThreadErrorBannerKeys.add(bannerKey); } export function isThreadErrorBannerDismissedForSession(bannerKey: string | null): boolean { return bannerKey !== null && sessionDismissedThreadErrorBannerKeys.has(bannerKey); } +export function resolveThreadErrorBanner(input: { + readonly threadKey: string; + readonly detailError: string | null; + readonly localError: string | null; + readonly runtimeError: string | null; +}) { + const candidates = [ + { + source: "local", + message: input.localError, + key: getThreadErrorBannerKey(input.threadKey, input.localError), + }, + { + source: "detail", + message: input.detailError, + key: getThreadErrorBannerKey(input.threadKey, input.detailError), + }, + { + source: "runtime", + message: input.runtimeError, + key: getThreadErrorBannerKey(input.threadKey, input.runtimeError), + }, + ]; + return candidates.find((candidate) => + shouldShowThreadErrorBanner( + input.threadKey, + candidate.message, + isThreadErrorBannerDismissedForSession(candidate.key), + ), + ); +} + export const ThreadErrorBanner = memo(function ThreadErrorBanner({ error, onDismiss, diff --git a/apps/web/src/components/files/fileSaveSessions.browser.test.tsx b/apps/web/src/components/files/fileSaveSessions.browser.test.tsx index 70456e2d92bb..2e8004dcdc04 100644 --- a/apps/web/src/components/files/fileSaveSessions.browser.test.tsx +++ b/apps/web/src/components/files/fileSaveSessions.browser.test.tsx @@ -78,7 +78,7 @@ describe("file save sessions in React", () => { ); finishWrite(); getUnsavedFile.mockReturnValue(null); - await vi.runAllTimersAsync(); + await vi.waitFor(() => expect(onPendingChange).toHaveBeenLastCalledWith("README.md", false)); expect(runAtomCommand.mock.calls.map(([, , request]) => request)).toEqual([ { @@ -86,6 +86,5 @@ describe("file save sessions in React", () => { input: { cwd: "/workspace", relativePath: "README.md", contents: "draft" }, }, ]); - expect(onPendingChange).toHaveBeenLastCalledWith("README.md", false); }); }); diff --git a/apps/web/src/state/entities.ts b/apps/web/src/state/entities.ts index 51760393f92d..5568d7804380 100644 --- a/apps/web/src/state/entities.ts +++ b/apps/web/src/state/entities.ts @@ -59,6 +59,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"), ); @@ -209,6 +212,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 000000000000..f5dbd8b7c686 --- /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 a8b5446add2c..dcc867427864 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 93e3f47121b2..28c885e6179c 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -263,6 +263,10 @@ export function runStreamInSession( } interface SubscriptionOptions { + readonly transform?: ( + stream: Stream.Stream, EnvironmentRpcStreamFailure>, + input: EnvironmentRpcInput, + ) => Stream.Stream, EnvironmentRpcStreamFailure>; /** Reports defects; the subscription then waits for the next session or resubscribe. */ readonly onDefect?: ( cause: Cause.Cause>, @@ -327,9 +331,10 @@ function subscribeDynamicMapped( method: tag, input, }); + const source = Stream.suspend(() => method(input)); const stream = mapStream( session, - Stream.suspend(() => method(input)), + options?.transform?.(source, input) ?? source, ).pipe( Stream.onFirst(() => Effect.sync(() => { diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 94440ad744e9..40ab237b40b7 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -14,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"; @@ -80,9 +81,17 @@ const BASE_PROJECTION: OrchestrationV2ThreadProjection = { thread: { ...v2Projection.thread, title: "Cached thread" }, }; +class TestStreamInterrupted { + readonly _tag = "TestStreamInterrupted"; +} + const TEST_DEFECT = { defect: "Malformed snapshot" } as const; -type TestThreadInput = OrchestrationV2ThreadStreamItem | Error | typeof TEST_DEFECT; +type TestThreadInput = + | OrchestrationV2ThreadStreamItem + | Error + | TestStreamInterrupted + | typeof TEST_DEFECT; function testSession( client: WsRpcProtocolClient, @@ -134,6 +143,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); @@ -156,7 +166,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o ? Effect.die(input) : input instanceof Error ? Effect.fail(input) - : Effect.succeed(input), + : input instanceof TestStreamInterrupted + ? Effect.failCause(Cause.interrupt(9999)) + : Effect.succeed(input), ), ); const client = { @@ -174,6 +186,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) @@ -294,6 +307,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, @@ -911,6 +925,8 @@ describe("EnvironmentThreads", () => { ); expect(settled.status).toBe("empty"); expect(configReads).toBe(2); + yield* TestClock.adjust("250 millis"); + expect(configReads).toBe(3); }), ); @@ -2406,11 +2422,171 @@ 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( "resubscribes after a snapshot decode defect ends the thread subscription (#23041)", () => Effect.gen(function* () { - const harness = yield* makeHarness({ httpSnapshot: { _tag: "unavailable" } }); + const harness = yield* makeHarness({ + httpSnapshot: { _tag: "unavailable" }, + completionMarker: true, + }); yield* SubscriptionRef.changes(harness.subscriptionCount).pipe( Stream.filter((count) => count >= 1), Stream.runHead, @@ -2425,13 +2601,13 @@ describe("EnvironmentThreads", () => { yield* TestClock.adjust("60 seconds"); expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(2); - yield* Queue.offer( - harness.inputs, + yield* Queue.offerAll(harness.inputs, [ snapshot({ ...BASE_PROJECTION, thread: { ...BASE_PROJECTION.thread, title: "Recovered thread" }, }), - ); + synchronized(), + ]); const recovered = yield* awaitThreadState( harness.observed, (value) => @@ -2440,6 +2616,31 @@ describe("EnvironmentThreads", () => { value.data.value.thread.title === "Recovered thread", ); expect(Option.isNone(recovered.error)).toBe(true); + + yield* Queue.offer(harness.wakeups, "application-active"); + yield* SubscriptionRef.changes(harness.subscriptionCount).pipe( + Stream.filter((count) => count === 3), + Stream.runHead, + ); + yield* Queue.offer(harness.inputs, TEST_DEFECT); + yield* SubscriptionRef.changes(harness.subscriptionCount).pipe( + Stream.filter((count) => count === 4), + Stream.runHead, + ); + yield* Queue.offer(harness.inputs, TEST_DEFECT); + const failedSnapshot = yield* awaitThreadState(harness.observed, (value) => + Option.isSome(value.error), + ); + expect(Option.getOrThrow(failedSnapshot.data).thread.title).toBe("Recovered thread"); + yield* Queue.offer(harness.wakeups, "application-active"); + yield* SubscriptionRef.changes(harness.subscriptionCount).pipe( + Stream.filter((count) => count === 5), + Stream.runHead, + ); + yield* Queue.offerAll(harness.inputs, [snapshot(BASE_PROJECTION), synchronized()]); + yield* awaitThreadState(harness.observed, (value) => value.status === "live"); + yield* TestClock.adjust("1 minute"); + expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(5); }), ); @@ -2450,6 +2651,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 a3984290e7b6..4a350f5e1257 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"; @@ -34,10 +37,14 @@ import { runCachePersistence } from "./cachePersistence.ts"; import * as ManagedRelay from "../relay/managedRelay.ts"; import { isAuthorizationFailure, + isRpcClientError, request, - subscribeDynamic, + subscribeDynamicWithSession, subscriptionRetryDelay, + 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"; @@ -68,6 +75,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 @@ -435,6 +450,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, @@ -681,10 +699,16 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // Keep this per-batch streaming path untraced for the same reason as applyEventsLocked. const applyItems = Effect.fnUntraced(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 ( @@ -974,36 +998,42 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make (yield* Ref.get(socketOnlyOpen)) && Option.isNone((yield* SubscriptionRef.get(state)).data) ); }); - // A stream defect parks the subscription until something resubscribes it. const resubscribes = yield* Queue.unbounded(); - 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* markSynchronizing; - // A 404 can arrive before the thread exists, so a miss asks HTTP again while mounted. let httpMisses = 0; let defectRetries = 0; - const backOffAfterDefect = Effect.suspend(() => { + const waitAfterDefect = Effect.suspend(() => { const retryDelay = subscriptionRetryDelay(THREAD_SUBSCRIPTION_RETRY_DELAY, defectRetries); defectRetries += 1; - return setStreamError("Could not synchronize the thread.").pipe( - Effect.andThen(Effect.sleep(retryDelay)), - ); + return Effect.sleep(retryDelay); }); - const followThread = (): Effect.Effect< - void, - never, + const backOffAfterDefect = setStreamError("Could not synchronize the thread.").pipe( + Effect.andThen(waitAfterDefect), + ); + const markSynchronizing = Effect.gen(function* () { + if (yield* Ref.get(resumingLive)) return; + yield* setConnecting; + }); + + yield* markSynchronizing; + const subscribeThread = ( + retrySnapshot = false, + ): Stream.Stream< + readonly [ + RpcSession, + EnvironmentRpcStreamValue, + ], + EnvironmentRpcStreamFailure, EnvironmentSupervisor.EnvironmentSupervisor - > => - subscribeDynamic( + > => { + let failedSession: RpcSession | undefined; + let resumeFailed = false; + let retryDefects = true; + return subscribeDynamicWithSession( ORCHESTRATION_V2_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { + failedSession = session; + resumeFailed = false; + retryDefects = true; let current = yield* SubscriptionRef.get(state); // A delete event already cleared this thread. Park without opening the socket. if (current.status === "deleted" && httpMisses === 0) { @@ -1064,6 +1094,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; }), @@ -1088,7 +1119,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 { @@ -1107,19 +1138,94 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }; }), { - // A socket-only open that fails retries over HTTP, which tells a - // missing thread apart from a failure, before showing an error. - onDefect: Effect.fn("EnvironmentThreadState.onDefect")(function* (cause) { - if (yield* failedSocketOnlyOpen) { - yield* Ref.set(socketOnlyOpen, false); - return yield* Queue.offer(resubscribes, undefined); - } - if (!Cause.hasDies(cause)) { - return yield* setStreamError("Could not synchronize the thread."); - } - yield* backOffAfterDefect; - yield* Queue.offer(resubscribes, undefined); - }), + 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); + retryDefects = + Cause.hasDies(cause) && + !cause.reasons.some( + (reason) => + reason._tag === "Die" && isThreadSynchronizationTimeout(reason.defect), + ); + resumeFailed = resumed && !synchronized; + return Stream.die(Cause.squash(cause)); + }), + ); + }), + ); + }, onExpectedFailure: Effect.fn("EnvironmentThreadState.onExpectedFailure")(function* (cause) { const retryOverHttp = !isAuthorizationFailure(cause) && (yield* failedSocketOnlyOpen); if (!retryOverHttp) yield* setStreamError(formatThreadError(cause)); @@ -1128,6 +1234,58 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make resubscribe: Stream.merge(foregroundResubscriptions, Stream.fromQueue(resubscribes)), }, ).pipe( + Stream.catchCause((cause) => + Stream.unwrap( + Effect.gen(function* () { + if (Cause.hasInterrupts(cause)) return Stream.failCause(cause); + if (!isAuthorizationFailure(cause) && (yield* failedSocketOnlyOpen)) { + yield* Ref.set(socketOnlyOpen, false); + return subscribeThread(); + } + if (resumeFailed) return subscribeThread(true); + // A source defect backs off; a failed catch-up waits for a session or foreground wakeup. + return Stream.fromEffect( + 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.", + ), + ).pipe( + Stream.drain, + Stream.concat( + Stream.merge( + SubscriptionRef.changes(supervisor.session).pipe( + Stream.filter( + (session) => Option.isSome(session) && session.value !== failedSession, + ), + ), + Stream.merge( + foregroundResubscriptions, + retryDefects && Cause.hasDies(cause) + ? Stream.fromEffect(waitAfterDefect) + : Stream.never, + ), + ).pipe( + Stream.take(1), + Stream.drain, + Stream.concat(Stream.suspend(() => subscribeThread())), + ), + ), + ); + }), + ), + ), + ); + }; + const followThread = (): Effect.Effect< + void, + never, + EnvironmentSupervisor.EnvironmentSupervisor + > => + subscribeThread().pipe( Stream.runForEachArray((items) => applyItems(items).pipe( Effect.andThen( diff --git a/scripts/scenarios/thread-history-failure-blocks-send.e2e.ts b/scripts/scenarios/thread-history-failure-blocks-send.e2e.ts new file mode 100644 index 000000000000..1db76531c40c --- /dev/null +++ b/scripts/scenarios/thread-history-failure-blocks-send.e2e.ts @@ -0,0 +1,67 @@ +import { expect } from "e2e"; + +import { test, WEB } from "./t3.ts"; + +test( + "A failed first history load keeps sending blocked", + WEB, + async ({ t3, app, screen, browser }) => { + const project = await t3.project("history-failure"); + await t3.thread({ + project, + title: "Existing conversation", + messages: ["Earlier work"], + }); + await browser.addInitScript(() => { + const NativeWebSocket = globalThis.WebSocket; + globalThis.WebSocket = class extends NativeWebSocket { + override send(data: Parameters[0]) { + const request: unknown = typeof data === "string" ? JSON.parse(data) : null; + if ( + typeof request === "object" && + request !== null && + "_tag" in request && + request._tag === "Request" && + "tag" in request && + request.tag === "orchestration.subscribeThread" && + "id" in request + ) { + queueMicrotask(() => + this.dispatchEvent( + new MessageEvent("message", { + data: JSON.stringify({ + _tag: "Exit", + requestId: request.id, + exit: { + _tag: "Failure", + cause: [ + { + _tag: "Fail", + error: { + _tag: "EnvironmentAuthorizationError", + message: "Thread history unavailable", + requiredScope: "orchestration:read", + }, + }, + ], + }, + }), + }), + ), + ); + return; + } + super.send(data); + } + }; + }); + await t3.open(); + await screen.getByRole("button", /^Existing conversation, Done,/).tap(); + await screen.getByRole("button", "Toggle thread details panel").tap(); + await screen.getByRole("button", "Toggle main sidebar").tap(); + await expect(screen.getByText("Thread history unavailable")).toBeVisible(); + await screen.getByRole("textbox").fill("Wait for the existing conversation"); + await app.screenshot("history-unavailable"); + await expect(screen.getByRole("button", /^(Submit message|Messages loading)$/)).toBeDisabled(); + }, +);