diff --git a/apps/mobile/src/state/thread-run-state.test.ts b/apps/mobile/src/state/thread-run-state.test.ts new file mode 100644 index 000000000000..cc103e1c1c26 --- /dev/null +++ b/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); + }); +}); diff --git a/apps/mobile/src/state/thread-run-state.ts b/apps/mobile/src/state/thread-run-state.ts new file mode 100644 index 000000000000..040b46d7eba4 --- /dev/null +++ b/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; +} diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 051fe543ea2b..32139e7cd804 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/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) { diff --git a/apps/mobile/src/state/use-thread-selection.ts b/apps/mobile/src/state/use-thread-selection.ts index 108afbbc9416..4335a742474b 100644 --- a/apps/mobile/src/state/use-thread-selection.ts +++ b/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, ], ); } diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 848357ec2bef..8353a429f73d 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/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[1]>; readonly loadCached?: Effect.Effect>; readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"]; + readonly registryEntries?: EnvironmentRegistry["Service"]["entries"]; }) { const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); @@ -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>( + 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()); + 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; + 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" }; diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index e0be27d35358..d856cdbfb7a5 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -23,6 +23,7 @@ import { connectionProjectionPhase } from "../connection/model.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import * as ConnectionWakeups from "../connection/wakeups.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; +import { derivePendingRequests } from "../pendingRequests.ts"; import { subscribeDynamic } from "../rpc/client.ts"; import { ThreadSnapshotLoader, type ThreadSnapshotWindow } from "./threadSnapshotHttp.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; @@ -128,7 +129,8 @@ function formatThreadError(cause: Cause.Cause): string { /** * A starting or running session is mid-turn. Its detail can change many times - * per second, so the disk cache waits for it to settle. + * per second, so streamed updates never reach the disk cache; the view writes + * its last state once when it closes (see the scope finalizer). */ export function isThreadSessionRunning(session: OrchestrationThread["session"]): boolean { return session?.status === "starting" || session?.status === "running"; @@ -170,6 +172,11 @@ function matchesThreadSnapshot( // marker the first replayed event moves the status to "synchronizing" on its // own. Downgrading here would flash a sync label on every return to a // recently viewed thread. +// Thread cache writes run one at a time, each re-checking its owner inside +// the permit. A closed view's delayed write can otherwise finish after the +// write of a view reopened meanwhile and leave the older copy on disk. +const threadCacheWrites = Semaphore.makeUnsafe(1); + function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadState { return { ...value, @@ -190,6 +197,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const cache = yield* EnvironmentCacheStore; const snapshotLoader = yield* ThreadSnapshotLoader; const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); + // Optional so harnesses that run this state without a registry still work. + // Read `entries` directly: registry.state() would acquire a supervisor and + // recreate the service scope of an environment that was just removed. + const registry = yield* Effect.serviceOption(EnvironmentRegistry); const environmentId = supervisor.target.environmentId; const retained = resumeCache?.snapshot; const owner = {}; @@ -275,7 +286,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make } | null>(null); const persistence = yield* Queue.sliding(1); - const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( + const persistUnlocked = Effect.fn("EnvironmentThreadState.persist")(function* ( snapshot: OrchestrationThreadDetailSnapshot, ) { if (resumeCache !== undefined && resumeCache.owner !== owner) return; @@ -311,6 +322,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ), ); }); + const persist = (snapshot: OrchestrationThreadDetailSnapshot) => + threadCacheWrites.withPermits(1)(persistUnlocked(snapshot)); yield* Stream.fromQueue(persistence).pipe( Stream.debounce("500 millis"), @@ -878,29 +891,68 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); yield* Effect.addFinalizer(() => Effect.sync(deregister)); + // Closing the view writes its last committed state once, even mid-turn. + // Streamed updates of a running thread never reach the disk cache + // (setThread), so without this every cold open of a Working thread waits + // for the network. The next open renders this snapshot at once and resumes + // after its sequence, like a warm return from the resume cache. Its session + // can be stale by then, so clients take run state from the shell meanwhile. yield* Effect.addFinalizer(() => Effect.suspend(() => { const { state: current, sequence: snapshotSequence } = committed; return Option.match(current.data, { onNone: () => Effect.void, - onSome: (thread) => - shouldPersistThread(thread) - ? persist({ - snapshotSequence, - thread, - ...Option.match(current.page, { - onNone: () => ({}), - onSome: (page) => - ({ - page: { - beforeCursor: page.beforeCursor, - hasMore: page.hasMore, - snapshotSequence, - }, - }) as const, - }), - }) - : Effect.void, + onSome: (thread) => { + const snapshot: OrchestrationThreadDetailSnapshot = { + snapshotSequence, + thread, + ...Option.match(current.page, { + onNone: () => ({}), + onSome: (page) => + ({ + page: { + beforeCursor: page.beforeCursor, + hasMore: page.hasMore, + snapshotSequence, + }, + }) as const, + }), + }; + if (shouldPersistThread(thread)) return persist(snapshot); + // An open approval or question can be answered elsewhere after the + // view closes; a disk copy would bring the answered card back on the + // next cold open, so those threads keep their last settled copy. + const pending = derivePendingRequests(thread.activities); + if (pending.approvals.length > 0 || pending.userInputs.length > 0) return Effect.void; + // Encoding a large running window is heavy: let the pop and the next + // screen's first frames go first. persist drops the write if the + // thread was reopened meanwhile (the owner check). Removing the + // environment also closes this view and clears its cache, and + // re-adding it installs a new target, so the write checks the + // registration itself, not the id, or a removed snapshot could come + // back as an orphan row. Enabling or disabling keeps the target. + const registration: Effect.Effect = Option.match(registry, { + onNone: () => Effect.succeed(null), + onSome: (service) => + SubscriptionRef.get(service.entries).pipe( + Effect.map((entries) => entries.get(environmentId)?.target), + ), + }); + return registration.pipe( + Effect.flatMap((closing) => + closing === undefined + ? Effect.void + : Effect.sleep("750 millis").pipe( + Effect.andThen(registration), + Effect.flatMap((current) => + current === closing ? persist(snapshot) : Effect.void, + ), + Effect.forkDetach, + Effect.asVoid, + ), + ), + ); + }, }); }), );