fix(client): recover snapshots that race thread creation

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

@@ -112,6 +112,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
     readonly latestLocalTurnOrdinal?: number | null;
   };
   readonly httpSnapshot?: ThreadSnapshotLoadResult;
+  readonly loadSnapshot?: () => Effect.Effect<ThreadSnapshotLoadResult>;
   /** The test sends the completion marker itself instead of the fake server sending it first. */
   readonly completionMarker?: boolean;
   readonly resumeCache?: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]>;
@@ -179,11 +180,15 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
   const snapshotLoader = ThreadSnapshotLoader.ThreadSnapshotLoader.of({
     load: (_prepared, threadId) =>
       Ref.update(loaderCalls, (count) => count + 1).pipe(
-        Effect.as(
-          threadId === THREAD_ID
-            ? (options?.httpSnapshot ??
-                ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult))
-            : ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult),
+        Effect.andThen(
+          options?.loadSnapshot ??
+            (() =>
+              Effect.succeed(
+                threadId === THREAD_ID
+                  ? (options?.httpSnapshot ??
+                      ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult))
+                  : ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult),
+              )),
         ),
       ),
   });
@@ -648,6 +653,75 @@ describe("EnvironmentThreads", () => {
     }),
   );
 
+  it.effect("recovers when the first snapshot precedes thread creation", () =>
+    Effect.gen(function* () {
+      const attempts = yield* Queue.unbounded<void>();
+      let created = false;
+      const harness = yield* makeHarness({
+        loadSnapshot: () =>
+          Queue.offer(attempts, undefined).pipe(
+            Effect.as(
+              created
+                ? {
+                    _tag: "present" as const,
+                    snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION },
+                  }
+                : { _tag: "missing" as const },
+            ),
+          ),
+      });
+      yield* Queue.take(attempts);
+      created = true;
+      yield* TestClock.adjust("500 millis");
+      yield* Queue.take(attempts);
+      yield* awaitThreadState(harness.observed, (state) => state.status === "live");
+      yield* Queue.offer(harness.inputs, titleUpdated("Response arrived", 8));
+      const state = yield* awaitThreadState(
+        harness.observed,
+        (value) => Option.getOrNull(value.data)?.thread.title === "Response arrived",
+      );
+      expect(state.status).toBe("live");
+      expect(yield* Ref.get(harness.removedThreads)).toEqual([]);
+    }),
+  );
+
+  it.effect("bounds missing snapshot retries and recovers on foreground", () =>
+    Effect.gen(function* () {
+      const attempts = yield* Queue.unbounded<void>();
+      let created = false;
+      const harness = yield* makeHarness({
+        loadSnapshot: () =>
+          Queue.offer(attempts, undefined).pipe(
+            Effect.as(
+              created
+                ? {
+                    _tag: "present" as const,
+                    snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION },
+                  }
+                : { _tag: "missing" as const },
+            ),
+          ),
+      });
+      yield* Queue.take(attempts);
+      for (const delay of [500, 1_000, 2_000, 4_000, 8_000, 16_000]) {
+        yield* TestClock.adjust(delay);
+        yield* Queue.take(attempts);
+      }
+      const missing = yield* awaitThreadState(harness.observed, (value) =>
+        Option.isSome(value.error),
+      );
+      expect(missing.status).not.toBe("deleted");
+      yield* TestClock.adjust("1 minute");
+      expect(yield* Ref.get(harness.loaderCalls)).toBe(7);
+      expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(0);
+      created = true;
+      yield* Queue.offer(harness.wakeups, "application-active");
+      yield* awaitThreadState(harness.observed, (value) => value.status === "live");
+      expect(yield* Ref.get(harness.loaderCalls)).toBe(8);
+      expect(yield* Ref.get(harness.removedThreads)).toEqual([]);
+    }),
+  );
+
   it.effect("retains a deletion instead of restoring the old disk snapshot", () =>
     Effect.gen(function* () {
       const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
@@ -681,8 +755,18 @@ describe("EnvironmentThreads", () => {
       };
       yield* Effect.scoped(
         Effect.gen(function* () {
-          const first = yield* makeHarness({ httpSnapshot: { _tag: "missing" }, resumeCache });
-          yield* awaitThreadState(first.observed, (value) => value.status === "deleted");
+          const attempts = yield* Queue.unbounded<void>();
+          const first = yield* makeHarness({
+            resumeCache,
+            loadSnapshot: () =>
+              Queue.offer(attempts, undefined).pipe(Effect.as({ _tag: "missing" as const })),
+          });
+          yield* Queue.take(attempts);
+          for (const delay of [500, 1_000, 2_000, 4_000, 8_000, 16_000]) {
+            yield* TestClock.adjust(delay);
+            yield* Queue.take(attempts);
+          }
+          yield* awaitThreadState(first.observed, (value) => Option.isSome(value.error));
         }),
       );
       const resumed = yield* makeHarness({

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

@@ -406,6 +406,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
     // state when only the projection changes. Bounded installs pass history so
     // projection + cursor persist together in one enqueue.
     const next = yield* SubscriptionRef.updateAndGet(state, (previous) => {
+      if (previous.status === "deleted") return previous;
       const history =
         options?.history !== undefined
           ? options.history
@@ -428,7 +429,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
     // Active projections can update many times per second and retain large tool
     // payloads. Persist once the run settles so cache encoding stays off the
     // streaming path. Progressive meta rides along when the window is incomplete.
-    if (shouldPersistThread(thread, next.history)) {
+    if (next.status !== "deleted" && shouldPersistThread(thread, next.history)) {
       const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
       yield* queuePersist(
         snapshotToPersist(
@@ -938,7 +939,7 @@ 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.
+        // An explicit 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") {
@@ -965,8 +966,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
               }),
             ),
           );
-          const httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult =
+          let httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult =
             yield* snapshotLoader.load(prepared, threadId);
+          // Navigation can precede thread creation; an HTTP miss is not a deletion event.
+          for (const delay of [500, 1_000, 2_000, 4_000, 8_000, 16_000]) {
+            if (httpResult._tag !== "missing") break;
+            yield* Effect.sleep(delay);
+            httpResult = yield* snapshotLoader.load(prepared, threadId);
+          }
           switch (httpResult._tag) {
             case "present": {
               if (canLoadHistory && httpResult.history !== undefined) {
@@ -997,9 +1004,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
               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;
+              yield* setStreamError("Thread not found. Reopen the thread to try again.");
               return yield* Effect.never;
             }
             case "unavailable": {