fix(client-runtime): recover threads when resume never completes

apps/mobile/src/features/threads/ThreadDetailScreen.tsx

@@ -174,6 +174,7 @@ import type {
 } from "../../lib/threadActivity";
 import { PendingApprovalCard } from "./PendingApprovalCard";
 import { ComposerErrorNotice } from "./ComposerErrorNotice";
+import { Notice } from "../../components/ui/Notice";
 import { ManagedRuntimeModeNotice } from "./ManagedRuntimeModeNotice";
 import { ComposerFeedback } from "./ComposerFeedback";
 import { ComposerUsageLimits } from "./ComposerUsageLimits";
@@ -256,6 +257,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;
@@ -538,6 +540,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":
@@ -1600,6 +1603,17 @@ function ThreadDetailBody(
                         onDismiss={() => props.onDismissFeedback(submission.id)}
                       />
                     ))}
+                    {props.threadSyncError != null && contentPresentationKind === "ready" ? (
+                      <View className="shrink-0 px-4 pb-3">
+                        <Notice
+                          message={props.threadSyncError}
+                          icon="exclamationmark.circle"
+                          tone="danger"
+                          iconTone="danger"
+                          floating
+                        />
+                      </View>
+                    ) : null}
                     {props.canOperateThread ? (
                       <ManagedRuntimeModeNotice
                         threadKey={selectedThreadKey}

apps/mobile/src/features/threads/ThreadRouteScreen.tsx

@@ -986,16 +986,14 @@ function ThreadRouteContent(
     return <OpeningThreadLoadingScreen />;
   }
 
-  // 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 = () => (
     <>
@@ -1078,6 +1076,7 @@ function ThreadRouteContent(
           respondingUserInputId={requests.respondingUserInputId}
           connectionStateLabel={routeConnectionState}
           threadSyncStatus={selectedThreadDetailState.status}
+          threadSyncError={threadDetailError}
           historyControls={historyControls}
           activeThreadBusy={composer.activeThreadBusy}
           canStopThread={awaitingBootstrapTurn || composer.interruptibleRunId !== null}

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" });
+  });
 });

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) {

apps/web/src/components/ChatView.tsx

@@ -246,6 +246,7 @@ import {
   resolveThreadDetailRef,
   useEnvironmentSupportsServerBrowser,
   useProject,
+  useThreadError,
   useQuietThreadProjection,
   useThreadUpdatedAt,
   useThreadLastVisitedAt,
@@ -509,8 +510,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,
   });
@@ -920,7 +923,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(

apps/web/src/state/entities.ts

@@ -56,6 +56,9 @@ const EMPTY_THREAD_SHELL_ATOM = Atom.make<EnvironmentThreadShell | null>(null).p
 const EMPTY_THREAD_PROJECTION_ATOM = Atom.make<EnvironmentThread | null>(null).pipe(
   Atom.withLabel("web-thread-projection:empty"),
 );
+const EMPTY_THREAD_ERROR_ATOM = Atom.make<string | null>(null).pipe(
+  Atom.withLabel("web-thread-error:empty"),
+);
 const EMPTY_THREAD_STATUS_ATOM = Atom.make<EnvironmentThreadStatus>("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),

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");
+  });
+});

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;
   }
 

packages/client-runtime/src/rpc/client.ts

@@ -236,6 +236,10 @@ export function runStreamGuarded<TTag extends EnvironmentStreamCommandRpcTag>(
 }
 
 interface SubscriptionOptions<TTag extends EnvironmentSubscriptionRpcTag> {
+  readonly transform?: (
+    stream: Stream.Stream<EnvironmentRpcStreamValue<TTag>, EnvironmentRpcStreamFailure<TTag>>,
+    input: EnvironmentRpcInput<TTag>,
+  ) => Stream.Stream<EnvironmentRpcStreamValue<TTag>, EnvironmentRpcStreamFailure<TTag>>;
   readonly onDefect?: (
     cause: Cause.Cause<EnvironmentRpcStreamFailure<TTag>>,
   ) => Effect.Effect<void, never, never>;
@@ -299,7 +303,11 @@ function subscribeDynamicMapped<TTag extends EnvironmentSubscriptionRpcTag, A>(
                         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;