diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 593e29286b66c..5d4d35bb5c0a0 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/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; /** The test sends the completion marker itself instead of the fake server sending it first. */ readonly completionMarker?: boolean; readonly resumeCache?: NonNullable[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(); + 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(); + 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[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(); + 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({ diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 5c6622b5ad9ae..1715c2af265cb 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/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": {