diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index fb4fd3fe1123e..0a51567edc480 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -344,7 +344,7 @@ export interface ThreadFeedProps { readonly onEditPendingMessage: ((message: QueuedThreadMessage) => void) | null; readonly environmentId: EnvironmentId; readonly threadId: ThreadId; - /** Null while the thread's creation is queued: reading detail then caches a not-found as deleted. */ + /** Null while the thread's creation is queued: reading detail then shows a not-found as deleted. */ readonly detailRef: ScopedThreadRef | null; readonly threadTitle: string; readonly workspaceRoot?: string | null; diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index 6ba5c635685a3..fa68a5e2c1a2c 100644 --- a/apps/mobile/src/state/threads.ts +++ b/apps/mobile/src/state/threads.ts @@ -35,7 +35,10 @@ export const threadEnvironment = createThreadEnvironmentAtoms( connectionAtomRuntime, threadFamilies.snapshotAtom, ); -export const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); +export const environmentThreads = createEnvironmentThreadStateAtoms( + connectionAtomRuntime, + threadEnvironment.snapshotAtom, +); export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( environmentThreads.stateAtom, ); diff --git a/apps/web/src/components/ChatView.browser.test.tsx b/apps/web/src/components/ChatView.browser.test.tsx index 8fe98b78da113..c2b3bf9078d04 100644 --- a/apps/web/src/components/ChatView.browser.test.tsx +++ b/apps/web/src/components/ChatView.browser.test.tsx @@ -59,6 +59,6 @@ it("a new draft does not load its reserved thread before the server lists it", a const threadRequests = fetchSpy.mock.calls .map(([input]) => String(input instanceof Request ? input.url : input)) .filter((url) => url.includes(`/api/orchestration/threads/${threadId}/`)); - // A 404 here marks the thread deleted, and the view never loads it once the send creates it. + // A 404 here shows the thread as deleted until the shell lists it. expect(threadRequests).toEqual([]); }); diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index c7919d658ccf6..60808f52ccfc1 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -31,7 +31,10 @@ export const threadEnvironment = createThreadEnvironmentAtoms( connectionAtomRuntime, threadFamilies.snapshotAtom, ); -const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); +const environmentThreads = createEnvironmentThreadStateAtoms( + connectionAtomRuntime, + threadEnvironment.snapshotAtom, +); export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( environmentThreads.stateAtom, ); diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index 23263d432444c..f632b34b6c445 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -3,6 +3,7 @@ import { EventId, ORCHESTRATION_V2_WS_METHODS, TurnItemId, + type OrchestrationV2ShellSnapshot, type OrchestrationV2ThreadHistoryPage, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadDetailSnapshot, @@ -33,7 +34,7 @@ import * as Persistence from "../platform/persistence.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import type { RpcSession } from "../rpc/session.ts"; import { createEnvironmentThreadDetailAtoms } from "./threadDetail.ts"; -import { v2Projection, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; +import { v2Projection, v2ShellSnapshot, v2ThreadId } from "./orchestrationV2TestFixtures.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; import { createEnvironmentThreadStateAtoms, type EnvironmentThreadState } from "./threads.ts"; import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts"; @@ -50,10 +51,14 @@ const THREAD: OrchestrationV2ThreadProjection = { thread: { ...v2Projection.thread, title: "Cached thread" }, }; const SNAPSHOT: OrchestrationV2ThreadDetailSnapshot = { snapshotSequence: 7, projection: THREAD }; +const EMPTY_SHELL: OrchestrationV2ShellSnapshot = { ...v2ShellSnapshot, threads: [] }; const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: { readonly snapshot?: OrchestrationV2ThreadDetailSnapshot; readonly snapshotUnavailable?: boolean; + readonly snapshotMissing?: boolean; + readonly threadNotCreated?: boolean; + readonly removeThreadStalls?: boolean; }) { const subscriptions = yield* Queue.unbounded<{ readonly afterSequence: number | undefined; @@ -66,6 +71,11 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? readonly closed: Deferred.Deferred; }>(); const snapshot = options?.snapshot ?? SNAPSHOT; + const removalInterrupted = yield* Deferred.make(); + let threadCreated = options?.threadNotCreated !== true; + const shellSnapshotAtom = Atom.make( + threadCreated ? v2ShellSnapshot : EMPTY_SHELL, + ); let httpLoads = 0; let diskLoads = 0; let opened = 0; @@ -175,7 +185,12 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? return Option.none(); }), saveThread: () => Effect.void, - removeThread: () => Effect.void, + removeThread: () => + options?.removeThreadStalls === true + ? Effect.never.pipe( + Effect.onInterrupt(() => Deferred.succeed(removalInterrupted, undefined)), + ) + : Effect.void, loadServerConfig: () => Effect.succeedNone, saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeedNone, @@ -191,6 +206,9 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? load: () => Effect.sync(() => { httpLoads += 1; + if (!threadCreated || options?.snapshotMissing === true) { + return { _tag: "missing" as const }; + } if (options?.snapshotUnavailable === true) { return { _tag: "unavailable" as const }; } @@ -212,7 +230,7 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? ), ), ); - const raw = createEnvironmentThreadStateAtoms(runtime); + const raw = createEnvironmentThreadStateAtoms(runtime, () => shellSnapshotAtom); const details = createEnvironmentThreadDetailAtoms(raw.stateAtom); const ref = { environmentId: TARGET.environmentId, threadId: THREAD_ID }; const stateAtom = details.stateAtom(ref); @@ -224,6 +242,15 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? return { registry, + createThread: (archived = false) => { + threadCreated = true; + registry.set( + shellSnapshotAtom, + archived + ? { ...v2ShellSnapshot, threads: [], archivedThreads: v2ShellSnapshot.threads } + : v2ShellSnapshot, + ); + }, makeRegistry, rawAtoms: raw, stateAtom, @@ -231,6 +258,7 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options? ref, subscriptions, olderLoads, + removalInterrupted, loadEarlier: () => historyController.loadEarlier(TARGET.environmentId, THREAD_ID), counts: () => ({ httpLoads, diskLoads, opened, active }), }; @@ -261,6 +289,79 @@ describe("createEnvironmentThreadStateAtoms", () => { vi.restoreAllMocks(); }); + it.effect.each(["mounted", "reopened", "archived"] as const)( + "loads a thread the shell lists after an earlier 404 while %s", + (mode) => + Effect.gen(function* () { + // Routes leave as soon as a thread reads as deleted, mid-way through its cache removal. + const reopened = mode === "reopened"; + const h = yield* makeHarness({ threadNotCreated: true, removeThreadStalls: reopened }); + const unmount = h.registry.mount(h.stateAtom); + yield* observeState(h.registry, h.stateAtom, (state) => state.status === "deleted"); + expect(h.counts()).toMatchObject({ httpLoads: 1, opened: 0 }); + if (reopened) { + unmount(); + yield* Deferred.await(h.removalInterrupted); + } + + h.createThread(mode === "archived"); + const release = reopened ? h.registry.mount(h.stateAtom) : unmount; + const subscription = yield* Queue.take(h.subscriptions); + expect(h.registry.get(h.stateAtom).status).toBe("synchronizing"); + yield* Queue.offer(subscription.events, { kind: "synchronized" }); + yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live"); + expect(currentThread(h.registry, h.stateAtom)).toEqual(THREAD); + expect(h.counts()).toMatchObject({ httpLoads: 2, opened: 1 }); + release(); + yield* Deferred.await(subscription.closed); + }), + ); + + it.effect("lets the socket decide when the shell lists a thread that answers 404", () => + Effect.gen(function* () { + const h = yield* makeHarness({ snapshotMissing: true }); + const unmount = h.registry.mount(h.stateAtom); + const subscription = yield* Queue.take(h.subscriptions); + yield* Queue.offer(subscription.events, { + kind: "snapshot", + snapshotSequence: SNAPSHOT.snapshotSequence, + projection: THREAD, + }); + yield* Queue.offer(subscription.events, { kind: "synchronized" }); + yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live"); + expect(h.counts()).toMatchObject({ httpLoads: 1, opened: 1 }); + unmount(); + yield* Deferred.await(subscription.closed); + }), + ); + + it.effect("keeps a delete event final while the shell still lists the thread", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const unmount = h.registry.mount(h.stateAtom); + const subscription = yield* Queue.take(h.subscriptions); + yield* Queue.offer(subscription.events, { + kind: "event", + sequence: 8, + event: { + id: EventId.make("deleted"), + type: "thread.deleted", + threadId: THREAD_ID, + occurredAt: THREAD.thread.updatedAt, + payload: { ...THREAD.thread, deletedAt: THREAD.thread.updatedAt }, + }, + }); + yield* observeState(h.registry, h.stateAtom, (state) => state.status === "deleted"); + unmount(); + yield* Deferred.await(subscription.closed); + + const release = h.registry.mount(h.stateAtom); + yield* observeState(h.registry, h.stateAtom, (state) => state.status === "deleted"); + expect(h.counts()).toMatchObject({ httpLoads: 1, opened: 1 }); + release(); + }), + ); + it.effect("keeps warm data and resumes a completed cursor without loading another snapshot", () => Effect.gen(function* () { const h = yield* makeHarness(); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 593e29286b66c..224ebb6efc8ac 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -673,6 +673,17 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("asks the server again when the session changes after a 404", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ httpSnapshot: { _tag: "missing" } }); + yield* awaitThreadState(harness.observed, (value) => value.status === "deleted"); + yield* harness.replaceSession; + yield* awaitThreadState(harness.observed, (value) => value.status === "synchronizing"); + yield* awaitThreadState(harness.observed, (value) => value.status === "deleted"); + expect(yield* Ref.get(harness.loaderCalls)).toBe(2); + }), + ); + it.effect("asks the server again after an HTTP miss, since the thread may not exist yet", () => Effect.gen(function* () { const resumeCache: NonNullable[1]> = { diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 5c6622b5ad9ae..a14b72142a78d 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -2,6 +2,7 @@ import { isOrchestrationV2WorkActive, ORCHESTRATION_V2_WS_METHODS, type EnvironmentId as EnvironmentIdType, + type OrchestrationV2ShellSnapshot, type OrchestrationV2ThreadDetailSnapshot, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadStreamItem, @@ -22,7 +23,7 @@ import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { HttpClient } from "effect/http"; -import { Atom } from "effect/reactivity"; +import { Atom, AtomRegistry } from "effect/reactivity"; import * as RemoteEnvironmentAuthorization from "../authorization/service.ts"; import * as EnvironmentRegistry from "../connection/registry.ts"; @@ -55,6 +56,8 @@ import { type EnvironmentThreadStatus, } from "./threadState.ts"; +const NEVER_LISTED = Stream.concat(Stream.make(false), Stream.never); + function statusWithoutLiveData( data: Option.Option, ): EnvironmentThreadStatus { @@ -191,6 +194,7 @@ function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadStat export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* ( threadId: ThreadIdType, resumeCache?: ThreadResumeCache, + listedInShell: Stream.Stream = NEVER_LISTED, ) { const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor; const cache = yield* Persistence.EnvironmentCacheStore; @@ -206,6 +210,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; const retained = resumeCache?.snapshot; + // A delete event is final. A 404 can predate the thread's creation, so it stands only until the + // shell lists the thread, and never outlives this scope. + let notFound = false; const owner = {}; if (resumeCache) resumeCache.owner = owner; const cached = @@ -275,7 +282,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make committed.persisted && matchesThreadSnapshot(committed, Option.getOrNull(current.data), sequence, current.history), }; - if (resumeCache?.owner === owner) resumeCache.snapshot = committed; + if (resumeCache?.owner === owner) resumeCache.snapshot = notFound ? undefined : committed; }); const persistence = yield* Queue.sliding(1); let queuedSnapshot = Option.getOrUndefined(cached); @@ -447,6 +454,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make history: patch(current.history), })); + const isListedInShell = Stream.runHead(listedInShell).pipe( + Effect.map(Option.getOrElse(() => false)), + ); + const reopenNotFound = Effect.suspend(() => { + notFound = false; + return SubscriptionRef.set(state, EMPTY_ENVIRONMENT_THREAD_STATE); + }); + const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.set(state, { @@ -938,12 +953,15 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ORCHESTRATION_V2_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { let current = yield* SubscriptionRef.get(state); - // A prior definitive miss (or delete event) already cleared this thread. - // Park the subscription attempt without opening the socket so we do not - // retry forever against a known-missing id. - if (current.status === "deleted") { + // A delete event already cleared this thread. Park the subscription attempt + // without opening the socket so we do not retry forever against a known-missing id. + if (current.status === "deleted" && !notFound) { return yield* Effect.never; } + if (notFound) { + yield* reopenNotFound; + current = yield* SubscriptionRef.get(state); + } yield* Effect.ignore(session.initialConfig); yield* Ref.set(awaitingCompletion, true); @@ -965,8 +983,17 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }), ), ); - const httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult = + let httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult = yield* snapshotLoader.load(prepared, threadId); + while (httpResult._tag === "missing" && !(yield* isListedInShell)) { + notFound = true; + yield* setDeleted(); + yield* listedInShell.pipe(Stream.filter(Boolean), Stream.runHead); + yield* reopenNotFound; + yield* Ref.set(awaitingCompletion, true); + yield* markSynchronizing; + httpResult = yield* snapshotLoader.load(prepared, threadId); + } switch (httpResult._tag) { case "present": { if (canLoadHistory && httpResult.history !== undefined) { @@ -996,12 +1023,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make current = yield* SubscriptionRef.get(state); break; } - case "missing": { - yield* setDeleted(); - // A 404 also means "not created yet". Only a delete event may outlive this mount. - if (resumeCache?.owner === owner) resumeCache.snapshot = undefined; - return yield* Effect.never; - } + // A thread the shell lists can answer 404 while the two race; the socket decides. + case "missing": case "unavailable": { break; } @@ -1042,12 +1065,15 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make function threadStateChanges( environmentId: EnvironmentIdType, threadId: ThreadIdType, - resumeCache?: ThreadResumeCache, + resumeCache: ThreadResumeCache, + listedInShell: Stream.Stream, ) { return followStreamInEnvironment( environmentId, Stream.unwrap( - makeEnvironmentThreadState(threadId, resumeCache).pipe(Effect.map(SubscriptionRef.changes)), + makeEnvironmentThreadState(threadId, resumeCache, listedInShell).pipe( + Effect.map(SubscriptionRef.changes), + ), ), ); } @@ -1060,6 +1086,9 @@ export function createEnvironmentThreadStateAtoms( | R, E >, + snapshotAtom: ( + environmentId: EnvironmentIdType, + ) => Atom.Atom, ) { // Cache definitions must outlive collectible live-atom definitions. The // registry retains these nodes without retaining environment or RPC scopes. @@ -1075,12 +1104,27 @@ export function createEnvironmentThreadStateAtoms( const family = Atom.family((key: string) => { const { environmentId, threadId } = parseThreadKey(key); const resumeAtom = resumeFamily(key); + const listedAtom = Atom.make((get) => { + const snapshot = get(snapshotAtom(environmentId)); + return ( + snapshot?.threads.some((thread) => thread.id === threadId && thread.deletedAt === null) || + snapshot?.archivedThreads.some( + (thread) => thread.id === threadId && thread.deletedAt === null, + ) || + false + ); + }); return runtime .atom( (get) => { get.mount(resumeAtom); const resume = get.once(resumeAtom); - const live = threadStateChanges(environmentId, threadId, resume); + const live = threadStateChanges( + environmentId, + threadId, + resume, + AtomRegistry.toStream(get.registry, listedAtom), + ); return resume.snapshot === undefined ? live : Stream.concat(Stream.succeed(cachedThreadState(resume.snapshot.state)), live);