perf(mobile): chats left mid-turn reopen from the phone's cache

apps/mobile/src/state/thread-run-state.test.ts

@@ -0,0 +1,38 @@
+import { describe, expect, it } from "vite-plus/test";
+
+import { selectThreadRunStateSource } from "./thread-run-state";
+
+const at = (updatedAt: string, source: string) => ({ updatedAt, source });
+
+describe("selectThreadRunStateSource", () => {
+  it("reads run state from whichever copy holds the newer event", () => {
+    const retained = at("2026-09-26T10:00:00.000Z", "detail");
+    const settledShell = at("2026-09-26T10:05:00.000Z", "shell");
+    // A detail kept in memory (or on disk) while the turn finished elsewhere.
+    expect(
+      selectThreadRunStateSource({ detail: retained, detailIsLive: true, shell: settledShell }),
+    ).toBe(settledShell);
+    // The open view's stream is ahead of the coalesced thread list.
+    const streamed = at("2026-09-26T10:06:00.000Z", "detail");
+    expect(
+      selectThreadRunStateSource({ detail: streamed, detailIsLive: true, shell: settledShell }),
+    ).toBe(streamed);
+  });
+
+  it("prefers the detail on a tie", () => {
+    const detail = at("2026-09-26T10:00:00.000Z", "detail");
+    const shell = at("2026-09-26T10:00:00.000Z", "shell");
+    expect(selectThreadRunStateSource({ detail, detailIsLive: false, shell })).toBe(detail);
+  });
+
+  it("trusts only a live detail before the thread list has loaded", () => {
+    const detail = at("2026-09-26T10:00:00.000Z", "detail");
+    expect(selectThreadRunStateSource({ detail, detailIsLive: true, shell: null })).toBe(detail);
+    expect(selectThreadRunStateSource({ detail, detailIsLive: false, shell: null })).toBeNull();
+  });
+
+  it("uses the shell when there is no detail", () => {
+    const shell = at("2026-09-26T10:00:00.000Z", "shell");
+    expect(selectThreadRunStateSource({ detail: null, detailIsLive: false, shell })).toBe(shell);
+  });
+});

apps/mobile/src/state/thread-run-state.ts

@@ -0,0 +1,25 @@
+/**
+ * Which copy of a thread decides whether it is working: its detail or its
+ * thread-list shell.
+ *
+ * A detail can be a copy retained in memory or written to disk when a view
+ * closed, even mid-turn, so it can say a turn is running long after it ended.
+ * Both copies stamp `updatedAt` with the time of the latest event they hold,
+ * so the newer one wins; on a tie the detail, which the open view streams.
+ * Before the thread list has loaded there is no shell, and only a live
+ * detail is trusted.
+ */
+export function selectThreadRunStateSource<
+  Detail extends { readonly updatedAt: string },
+  Shell extends { readonly updatedAt: string },
+>(input: {
+  readonly detail: Detail | null;
+  readonly detailIsLive: boolean;
+  /** The thread-list shell only, never one derived from the detail. */
+  readonly shell: Shell | null;
+}): Detail | Shell | null {
+  const { detail, shell } = input;
+  if (shell === null) return input.detailIsLive ? detail : null;
+  if (detail === null) return shell;
+  return Date.parse(detail.updatedAt) >= Date.parse(shell.updatedAt) ? detail : shell;
+}

apps/mobile/src/state/use-thread-composer-state.ts

@@ -2,6 +2,7 @@ import type { ComposerTextPaste } from "../native/T3ComposerEditor.types";
 import { useAtomValue } from "@effect/atom-react";
 import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 import { Alert } from "react-native";
+import * as Option from "effect/Option";
 
 import {
   CommandId,
@@ -65,8 +66,9 @@ import {
   useComposerDraft,
 } from "./use-composer-drafts";
 import { setPendingConnectionError } from "../state/use-remote-environment-registry";
-import { useSelectedThreadDetail } from "../state/use-thread-detail";
+import { useSelectedThreadDetailState } from "../state/use-thread-detail";
 import { useThreadSelection } from "../state/use-thread-selection";
+import { selectThreadRunStateSource } from "./thread-run-state";
 import { enqueueThreadOutboxMessage } from "./thread-outbox";
 import { dispatchingQueuedMessageIdAtom, useThreadOutboxMessages } from "./use-thread-outbox";
 import { threadEnvironment } from "./threads";
@@ -127,9 +129,11 @@ export function useThreadComposerState() {
   const {
     selectedThread: selectedThreadShell,
     selectedThreadCreation,
+    selectedThreadListShell,
     selectedEnvironmentRuntime,
   } = useThreadSelection();
-  const selectedThreadDetail = useSelectedThreadDetail();
+  const selectedThreadDetailState = useSelectedThreadDetailState();
+  const selectedThreadDetail = Option.getOrNull(selectedThreadDetailState.data);
   const composerDrafts = useAtomValue(composerDraftsAtom);
   const acknowledgedMessages = useAtomValue(acknowledgedThreadMessagesAtom);
   const queuedMessagesByThreadKey = useThreadOutboxMessages();
@@ -256,17 +260,24 @@ export function useThreadComposerState() {
       )
     : null;
 
+  // Run state (working or not) comes from whichever copy holds the newer
+  // event: the detail can be a copy retained in memory or written to disk
+  // mid-turn, still running long after that turn ended (thread-run-state.ts).
+  const selectedThreadRunState = selectThreadRunStateSource({
+    detail: selectedThreadDetail,
+    detailIsLive: selectedThreadDetailState.status === "live",
+    shell: selectedThreadListShell,
+  });
   const selectedThreadSessionActivity = useMemo(() => {
-    const selectedThread = selectedThreadDetail ?? selectedThreadShell;
-    if (!selectedThread?.session) {
+    if (!selectedThreadRunState?.session) {
       return null;
     }
 
     return {
-      orchestrationStatus: selectedThread.session.status,
-      activeTurnId: selectedThread.session.activeTurnId ?? undefined,
+      orchestrationStatus: selectedThreadRunState.session.status,
+      activeTurnId: selectedThreadRunState.session.activeTurnId ?? undefined,
     };
-  }, [selectedThreadDetail, selectedThreadShell]);
+  }, [selectedThreadRunState]);
 
   const isCompacting = useMemo(() => {
     const queuedMessage = selectedThreadQueuedMessages.findLast(
@@ -281,12 +292,14 @@ export function useThreadComposerState() {
         message.text.trim().toLowerCase() === "/compact" &&
         !message.attachments?.length,
     );
+    // Run state comes from selectedThreadRunState: a cached detail can carry a
+    // session and turn that finished while the view was closed.
     const compactRequestIsActive =
       latestCompactMessage !== undefined &&
       (latestCompactMessage.createdAt >
-        (selectedThread?.latestTurn?.requestedAt ?? latestCompactMessage.createdAt) ||
-        (selectedThread?.latestTurn?.state === "running" &&
-          latestCompactMessage.createdAt === selectedThread.latestTurn.requestedAt));
+        (selectedThreadRunState?.latestTurn?.requestedAt ?? latestCompactMessage.createdAt) ||
+        (selectedThreadRunState?.latestTurn?.state === "running" &&
+          latestCompactMessage.createdAt === selectedThreadRunState.latestTurn.requestedAt));
     const compactionSettled = selectedThreadDetail?.activities.some((activity) => {
       if (!["context-compaction", "provider.turn.start.failed"].includes(activity.kind))
         return false;
@@ -298,30 +311,29 @@ export function useThreadComposerState() {
     });
     return (
       queuedMessage !== undefined ||
-      ((selectedThread?.session?.status === "starting" ||
-        selectedThread?.session?.status === "running") &&
+      ((selectedThreadRunState?.session?.status === "starting" ||
+        selectedThreadRunState?.session?.status === "running") &&
         compactRequestIsActive &&
         !compactionSettled)
     );
   }, [
     dispatchingQueuedMessageId,
-    selectedThread,
+    selectedThreadRunState,
     selectedThreadDetail,
     selectedThreadQueuedMessages,
   ]);
 
   const activeWorkStartedAt = useMemo(() => {
-    const selectedThread = selectedThreadDetail ?? selectedThreadShell;
-    if (!selectedThread) {
+    if (!selectedThreadRunState) {
       return null;
     }
 
     return deriveActiveWorkStartedAt(
-      selectedThread.latestTurn,
+      selectedThreadRunState.latestTurn,
       selectedThreadSessionActivity,
       null,
     );
-  }, [selectedThreadDetail, selectedThreadSessionActivity, selectedThreadShell]);
+  }, [selectedThreadRunState, selectedThreadSessionActivity]);
 
   const onSendMessage = useCallback(async () => {
     if (!selectedThreadShell) {

apps/mobile/src/state/use-thread-selection.ts

@@ -178,6 +178,8 @@ function useResolvedThreadSelection(params: ThreadSelectionRouteParams | undefin
     () => ({
       selectedThreadRef,
       selectedThread,
+      /** The thread-list shell only; selectedThread can be derived from the detail. */
+      selectedThreadListShell: selectedThreadShell,
       selectedThreadCreation,
       selectedThreadDetailState,
       selectedThreadProject,
@@ -192,6 +194,7 @@ function useResolvedThreadSelection(params: ThreadSelectionRouteParams | undefin
       selectedThreadDetailState,
       selectedThreadProject,
       selectedThreadRef,
+      selectedThreadShell,
     ],
   );
 }

packages/client-runtime/src/state/threads-sync.test.ts

@@ -31,6 +31,7 @@ import {
   type SupervisorConnectionState,
 } from "../connection/model.ts";
 import * as ConnectionWakeups from "../connection/wakeups.ts";
+import { EnvironmentRegistry } from "../connection/registry.ts";
 import * as EnvironmentSupervisor from "../connection/supervisor.ts";
 import * as Persistence from "../platform/persistence.ts";
 import * as RpcSession from "../rpc/session.ts";
@@ -142,6 +143,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
   readonly resumeCache?: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]>;
   readonly loadCached?: Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>>;
   readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"];
+  readonly registryEntries?: EnvironmentRegistry["Service"]["entries"];
 }) {
   const inputs = yield* Queue.unbounded<TestThreadInput>();
   const observed = yield* Queue.unbounded<EnvironmentThreadState>();
@@ -242,7 +244,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
     clearVcsRefs: () => Effect.void,
     clear: () => Effect.void,
   });
-  const threadState = yield* makeEnvironmentThreadState(THREAD_ID, options?.resumeCache).pipe(
+  const makeState = makeEnvironmentThreadState(THREAD_ID, options?.resumeCache).pipe(
     Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
     Effect.provideService(Persistence.EnvironmentCacheStore, cache),
     Effect.provideService(ThreadSnapshotLoader, snapshotLoader),
@@ -251,6 +253,13 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
       ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }),
     ),
   );
+  const threadState = yield* options?.registryEntries === undefined
+    ? makeState
+    : makeState.pipe(
+        Effect.provideService(EnvironmentRegistry, {
+          entries: options.registryEntries,
+        } as unknown as EnvironmentRegistry["Service"]),
+      );
   yield* SubscriptionRef.changes(threadState).pipe(
     Stream.runForEach((state) =>
       Ref.update(stateChangeCount, (count) => count + 1).pipe(
@@ -692,7 +701,7 @@ describe("EnvironmentThreads", () => {
     }),
   );
 
-  it.effect("does not persist active thread snapshots during streaming or teardown", () =>
+  it.effect("does not persist active threads while streaming or rewrite them on teardown", () =>
     Effect.gen(function* () {
       const savedThreads = yield* Effect.scoped(
         Effect.gen(function* () {
@@ -717,6 +726,128 @@ describe("EnvironmentThreads", () => {
     }),
   );
 
+  it.effect("writes a running thread once, when its view closes", () =>
+    Effect.gen(function* () {
+      const savedThreads = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const harness = yield* makeHarness({
+            httpSnapshot: Option.some({ snapshotSequence: 7, thread: ACTIVE_THREAD }),
+          });
+          yield* awaitThreadState(harness.observed, (value) => value.status === "live");
+          yield* Queue.offer(harness.inputs, titleUpdated("Streaming", 8));
+          yield* awaitThreadState(
+            harness.observed,
+            (value) => Option.getOrNull(value.data)?.title === "Streaming",
+          );
+          yield* TestClock.adjust("500 millis");
+          yield* Effect.yieldNow;
+
+          // Streamed updates of a running thread never encode it.
+          expect(yield* Ref.get(harness.savedThreads)).toEqual([]);
+          return harness.savedThreads;
+        }),
+      );
+
+      // Closing the view writes the committed state once, after the pop has
+      // settled, under the sequence it reflects, so the next cold open renders
+      // it and resumes after 8.
+      expect(yield* Ref.get(savedThreads)).toEqual([]);
+      yield* TestClock.adjust("750 millis");
+      yield* Effect.yieldNow;
+      expect(
+        (yield* Ref.get(savedThreads)).map((saved) => [
+          saved.snapshotSequence,
+          saved.thread.title,
+          saved.thread.session?.status,
+        ]),
+      ).toEqual([[8, "Streaming", "running"]]);
+    }),
+  );
+
+  it.effect("keeps a running thread waiting on an approval off the disk", () =>
+    Effect.gen(function* () {
+      const waiting: OrchestrationThread = {
+        ...ACTIVE_THREAD,
+        activities: [
+          {
+            id: EventId.make("approval-open"),
+            createdAt: "2026-02-23T00:00:01.000Z",
+            kind: "approval.requested",
+            summary: "Command approval requested",
+            tone: "approval",
+            payload: { requestId: "req-1", requestKind: "command", detail: "bun run lint" },
+            turnId: TurnId.make("turn-1"),
+          },
+        ],
+      };
+      const savedThreads = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const harness = yield* makeHarness({
+            httpSnapshot: Option.some({ snapshotSequence: 7, thread: waiting }),
+          });
+          yield* awaitThreadState(harness.observed, (value) => value.status === "live");
+          return harness.savedThreads;
+        }),
+      );
+
+      // The approval can be answered elsewhere after the view closes; a disk
+      // copy would bring the answered card back on the next cold open.
+      yield* TestClock.adjust("750 millis");
+      yield* Effect.yieldNow;
+      expect(yield* Ref.get(savedThreads)).toEqual([]);
+    }),
+  );
+
+  it.effect("drops the close-time write when the environment was removed meanwhile", () =>
+    Effect.gen(function* () {
+      const entries = yield* SubscriptionRef.make<ReadonlyMap<EnvironmentId, never>>(
+        new Map([[TARGET.environmentId, undefined as never]]),
+      );
+      const savedThreads = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const harness = yield* makeHarness({
+            httpSnapshot: Option.some({ snapshotSequence: 7, thread: ACTIVE_THREAD }),
+            registryEntries: entries as unknown as EnvironmentRegistry["Service"]["entries"],
+          });
+          yield* awaitThreadState(harness.observed, (value) => value.status === "live");
+          return harness.savedThreads;
+        }),
+      );
+
+      // remove() drops the entry and then clears the cache; a late write
+      // would leave an orphan row for a re-added environment to cold-open.
+      yield* SubscriptionRef.set(entries, new Map<EnvironmentId, never>());
+      yield* TestClock.adjust("750 millis");
+      yield* Effect.yieldNow;
+      expect(yield* Ref.get(savedThreads)).toEqual([]);
+    }),
+  );
+
+  it.effect("drops the close-time write when the environment was re-added meanwhile", () =>
+    Effect.gen(function* () {
+      const registered = (target: unknown) =>
+        new Map([[TARGET.environmentId, { target } as never]]) as ReadonlyMap<EnvironmentId, never>;
+      const entries = yield* SubscriptionRef.make(registered({}));
+      const savedThreads = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const harness = yield* makeHarness({
+            httpSnapshot: Option.some({ snapshotSequence: 7, thread: ACTIVE_THREAD }),
+            registryEntries: entries as unknown as EnvironmentRegistry["Service"]["entries"],
+          });
+          yield* awaitThreadState(harness.observed, (value) => value.status === "live");
+          return harness.savedThreads;
+        }),
+      );
+
+      // Same environment id, new registration: the removed snapshot must not
+      // be written back into the fresh cache.
+      yield* SubscriptionRef.set(entries, registered({}));
+      yield* TestClock.adjust("750 millis");
+      yield* Effect.yieldNow;
+      expect(yield* Ref.get(savedThreads)).toEqual([]);
+    }),
+  );
+
   it.effect("seeds the thread from the HTTP snapshot and resumes live events", () =>
     Effect.gen(function* () {
       const httpThread: OrchestrationThread = { ...BASE_THREAD, title: "HTTP title" };