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

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" ? (
+                      <View className="shrink-0 px-4 pb-3">
+                        <Notice
+                          message={props.threadSyncError}
+                          icon="exclamationmark.circle"
+                          tone="danger"
+                          iconTone="danger"
+                          floating
+                        />
+                      </View>
+                    ) : null}
                     {freshSessionAvailable ? (
                       <FreshSessionNotice threadKey={selectedThreadKey} />
                     ) : null}

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

@@ -952,16 +952,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 = () => (
     <>
@@ -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}

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/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({

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<unknown, FakeDriver> }>(),
+    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<void>();
+  const nextFailure = Deferred.makeUnsafe<void>();
+  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 router={router} />, {
+    // 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();
+});