fix(mobile): queued messages are delivered by an app service, not a render loop

apps/mobile/src/App.tsx

@@ -1,4 +1,6 @@
 import { installPermissionUpdateNotice } from "./features/connection/permissionUpdateNotice";
+import { installComposerAttachmentUploads } from "./state/composer-attachment-uploads";
+import { installThreadOutboxDelivery } from "./state/threadOutboxDelivery";
 import * as Linking from "expo-linking";
 import * as SplashScreen from "expo-splash-screen";
 import { useEffect } from "react";
@@ -86,11 +88,14 @@ const navigationRef = createNavigationContainerRef();
 
 declare const module: { readonly hot?: FoundationHotModule } | undefined;
 
-const disposePermissionUpdateNotice = installPermissionUpdateNotice();
-disposeOnFoundationReplace(
-  typeof module === "undefined" ? undefined : module.hot,
-  disposePermissionUpdateNotice,
-);
+const appServiceDisposers = [
+  installPermissionUpdateNotice(),
+  installThreadOutboxDelivery(),
+  installComposerAttachmentUploads(),
+];
+disposeOnFoundationReplace(typeof module === "undefined" ? undefined : module.hot, () => {
+  for (const dispose of appServiceDisposers) dispose();
+});
 
 function SplashScreenCoordinator() {
   const { isReady } = useAppearancePreferences();

apps/mobile/src/Stack.tsx

@@ -149,8 +149,6 @@ import { deriveLayout } from "./lib/layout";
 import { useNativeLayoutMetrics } from "./native/native-layout-metrics";
 import { nativeHeaderScrollEdgeEffects } from "./native/StackHeader";
 import { HORIZONTAL_FORM_SHEET_OPTIONS, formSheetDetents } from "./native/sheet-surface";
-import { useThreadOutboxDrain } from "./state/use-thread-outbox-drain";
-import { useComposerAttachmentUploadWorker } from "./state/composer-attachment-uploads";
 import {
   selectWorkspaceDetail,
   workspaceDetailRouter,
@@ -485,13 +483,6 @@ function workspaceLocationFromState(state: NavigationState) {
   };
 }
 
-// Isolate subscriptions so streaming updates cannot re-render RootStackLayout and every screen.
-function ThreadOutboxDrainWorker() {
-  useThreadOutboxDrain();
-  useComposerAttachmentUploadWorker();
-  return null;
-}
-
 function RootStackLayout(props: {
   readonly children: React.ReactNode;
   readonly state: NavigationState;
@@ -546,7 +537,6 @@ function RootStackLayout(props: {
       routePendingTaskId={routePendingTaskId}
     >
       <HardwareKeyboardCommandProvider pathname={pathname}>
-        <ThreadOutboxDrainWorker />
         <ShowcaseCaptureCoordinator pathname={pathname} />
         <SettingsEnvironmentFilterProvider open={settingsOpen}>
           <WorkspaceConnectionStatusProvider>

apps/mobile/src/state/composer-attachment-uploads.test.ts

@@ -19,20 +19,9 @@ const native = vi.hoisted(() => ({
   remove: Symbol("remove"),
   transfer:
     vi.fn<(uri: string, url: string, options: UploadOptions) => Promise<{ status: number }>>(),
-  effects: [] as Array<() => void | (() => void)>,
-  queueRef: {
-    current: null as ReturnType<
-      typeof import("../lib/composerAttachmentUploadQueue").createComposerAttachmentUploadQueue
-    > | null,
-  },
 }));
 
 vi.mock("react-native", () => ({ Alert: { alert: vi.fn() } }));
-vi.mock("react", async (importOriginal) => ({
-  ...(await importOriginal<typeof import("react")>()),
-  useEffect: (effect: () => void | (() => void)) => native.effects.push(effect),
-  useRef: () => native.queueRef,
-}));
 vi.mock("@effect/atom-react", () => ({
   useAtomValue: (atom: Parameters<typeof appAtomRegistry.get>[0]) => appAtomRegistry.get(atom),
 }));
@@ -111,6 +100,8 @@ vi.mock("./entities", () => ({
   useServerConfigs: () => new Map([["environment-1", config]]),
   useThreadShells: () => [],
 }));
+vi.mock("./projects", () => ({ environmentProjects: {} }));
+vi.mock("./workspace", () => ({ workspaceConnections: {} }));
 vi.mock("./server", async () => {
   const { Atom } = await import("effect/reactivity");
   return { serverEnvironment: { configValueAtom: Atom.family(() => Atom.make(null)) } };
@@ -173,25 +164,22 @@ vi.mock("./use-thread-outbox", async () => {
     dispatchingQueuedMessageIdAtom: Atom.make<Record<string, string>>({}).pipe(Atom.keepAlive),
     useThreadOutboxMessages: () =>
       appAtomRegistry.get(threadOutboxManager.queuedMessagesByThreadKeyAtom),
-    useThreadOutboxShellStatuses: () => new Map(),
+    threadOutboxShellStatusesAtom: Atom.make(new Map()),
   };
 });
-vi.mock("./use-remote-environment-registry", () => ({
-  useRemoteConnectionStatus: () => ({
-    connectedEnvironments: [{ environmentId: "environment-1", connectionState: "connected" }],
-  }),
-}));
 
 import {
   composerAttachmentUploadBlockReason,
   composerAttachmentsStillUploading,
   composerAttachmentUploadKey,
 } from "../lib/composerAttachmentUploadQueue";
 import type { DraftComposerAttachment } from "../lib/composerImages";
+import { Atom } from "effect/reactivity";
 import { appAtomRegistry } from "./atom-registry";
 import {
+  composerAttachmentUploadRequests,
   composerAttachmentUploadsAtom,
-  useComposerAttachmentUploadWorker,
+  createComposerAttachmentUploadWorker,
 } from "./composer-attachment-uploads";
 import {
   enqueueThreadOutboxMessage,
@@ -200,7 +188,7 @@ import {
 } from "./thread-outbox";
 import type { QueuedThreadMessage } from "./thread-outbox-model";
 import * as drafts from "./use-composer-drafts";
-import { prepareQueuedMessageAttachments } from "./use-thread-outbox-drain";
+import { prepareQueuedMessageAttachments } from "./threadOutboxDelivery";
 
 const environmentId = EnvironmentId.make("environment-1");
 const draftKey = "environment-1:thread-1";
@@ -217,12 +205,19 @@ const image: DraftComposerAttachment = {
 };
 let disposeWorker: (() => void) | undefined;
 
+const requestsAtom = Atom.make((get) =>
+  composerAttachmentUploadRequests({
+    drafts: get(drafts.composerDraftUploadSourcesAtom),
+    queuedMessagesByThreadKey: get(threadOutboxManager.queuedMessagesByThreadKeyAtom),
+    connectedEnvironmentIds: new Set([environmentId]),
+    operableEnvironmentIds: new Set([environmentId]),
+    serverConfigs: new Map([[environmentId, config]]),
+  }),
+);
+
 function worker() {
-  native.effects = [];
-  useComposerAttachmentUploadWorker();
-  disposeWorker = native.effects[0]!() ?? undefined;
-  const queue = native.queueRef.current!;
-  native.effects[1]!();
+  const { queue, dispose } = createComposerAttachmentUploadWorker(requestsAtom);
+  disposeWorker = dispose;
   return queue;
 }