fix(client-runtime): new threads no longer stay blank when opened before they exist

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

@@ -129,6 +129,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
   const retryCount = yield* Ref.make(0);
   const subscriptionCount = yield* SubscriptionRef.make(0);
   const loaderCalls = yield* Ref.make(0);
+  const snapshotLoads = yield* Queue.unbounded<void>();
+  const httpSnapshot = yield* Ref.make<ThreadSnapshotLoadResult>(
+    options?.httpSnapshot ?? { _tag: "unavailable" },
+  );
   const lastSubscribeAfterSequence = yield* Ref.make<number | undefined>(undefined);
   const lastRequestCompletionMarker = yield* Ref.make(false);
   const lastAcceptBoundedSnapshot = yield* Ref.make<true | undefined>(undefined);
@@ -179,11 +183,11 @@ 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(
+        Effect.andThen(Queue.offer(snapshotLoads, undefined)),
+        Effect.andThen(
           threadId === THREAD_ID
-            ? (options?.httpSnapshot ??
-                ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult))
-            : ({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult),
+            ? Ref.get(httpSnapshot)
+            : Effect.succeed({ _tag: "unavailable" } satisfies ThreadSnapshotLoadResult),
         ),
       ),
   });
@@ -281,6 +285,8 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
     retryCount,
     subscriptionCount,
     loaderCalls,
+    snapshotLoads,
+    httpSnapshot,
     lastSubscribeAfterSequence,
     lastRequestCompletionMarker,
     lastAcceptBoundedSnapshot,
@@ -2013,6 +2019,39 @@ describe("EnvironmentThreads", () => {
     }),
   );
 
+  it.effect.each(["same session", "replaced session"] as const)(
+    "loads a new thread whose first snapshot request ran before it existed (%s)",
+    (session) =>
+      Effect.gen(function* () {
+        const harness = yield* makeHarness({
+          httpSnapshot: { _tag: "missing" },
+          completionMarker: true,
+        });
+        yield* Queue.take(harness.snapshotLoads);
+        yield* awaitThreadState(harness.observed, (value) => value.status === "deleted");
+
+        if (session === "replaced session") {
+          yield* harness.replaceSession;
+          yield* Queue.take(harness.snapshotLoads);
+        }
+        yield* Ref.set(harness.httpSnapshot, {
+          _tag: "present",
+          snapshot: { snapshotSequence: 11, projection: BASE_PROJECTION },
+        });
+        yield* TestClock.adjust("250 millis");
+
+        yield* awaitThreadState(
+          harness.observed,
+          (value) => value.status === "synchronizing" && Option.isSome(value.data),
+        );
+        yield* Queue.offer(harness.inputs, synchronized());
+        const state = yield* awaitThreadState(harness.observed, (value) => value.status === "live");
+        expect(Option.getOrThrow(state.data)).toEqual(BASE_PROJECTION);
+        expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(11);
+        expect(yield* SubscriptionRef.get(harness.subscriptionCount)).toBe(1);
+      }),
+  );
+
   it.effect("does not recreate a deleted cache from a queued save or finalizer", () =>
     Effect.gen(function* () {
       const scope = yield* Scope.make();

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

@@ -152,6 +152,7 @@ interface ThreadResumeSnapshot {
   readonly sequence: number;
   readonly persisted: boolean;
   readonly acceptsBoundedSnapshots?: boolean;
+  readonly deletionConfirmed?: boolean;
 }
 
 interface ThreadResumeCache {
@@ -245,11 +246,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
   const canLoadHistory = Option.isSome(httpClient) && Option.isSome(historyController);
   const acceptsBoundedSocketSnapshots = yield* Ref.make(canLoadHistory);
   const lastSequence = yield* SubscriptionRef.make(initialSequence);
+  // Only a delete event proves deletion. A snapshot miss can precede a new thread's creation.
+  let deletionConfirmed = retained?.deletionConfirmed === true;
   let committed: ThreadResumeSnapshot = {
     state: initialState,
     sequence: initialSequence,
     persisted: retained?.persisted ?? Option.isSome(cached),
     acceptsBoundedSnapshots: canLoadHistory,
+    deletionConfirmed,
   };
   if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
   const awaitingCompletion = yield* Ref.make(false);
@@ -271,6 +275,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
       state: current,
       sequence,
       acceptsBoundedSnapshots: yield* Ref.get(acceptsBoundedSocketSnapshots),
+      deletionConfirmed,
       persisted:
         committed.persisted &&
         matchesThreadSnapshot(committed, Option.getOrNull(current.data), sequence, current.history),
@@ -581,6 +586,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
     );
 
     if (result._tag === "delete") {
+      deletionConfirmed = true;
       yield* setDeleted();
       return;
     }
@@ -938,10 +944,8 @@ 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") {
+        // Park without opening the socket so we do not retry forever against a deleted id.
+        if (current.status === "deleted" && deletionConfirmed) {
           return yield* Effect.never;
         }
 
@@ -965,8 +969,17 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
               }),
             ),
           );
-          const httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult =
+          let httpResult: ThreadSnapshotLoader.ThreadSnapshotLoadResult =
             yield* snapshotLoader.load(prepared, threadId);
+          for (let attempt = 0; httpResult._tag === "missing"; attempt++) {
+            if (current.status !== "deleted") {
+              yield* setDeleted();
+              yield* Ref.set(awaitingCompletion, true);
+              current = yield* SubscriptionRef.get(state);
+            }
+            yield* Effect.sleep(Math.min(250 * 2 ** attempt, 5_000));
+            httpResult = yield* snapshotLoader.load(prepared, threadId);
+          }
           switch (httpResult._tag) {
             case "present": {
               if (canLoadHistory && httpResult.history !== undefined) {
@@ -996,10 +1009,6 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
               current = yield* SubscriptionRef.get(state);
               break;
             }
-            case "missing": {
-              yield* setDeleted();
-              return yield* Effect.never;
-            }
             case "unavailable": {
               break;
             }