From ac13741f1700f066d890096d8bed784bb19d8c0d Mon Sep 17 00:00:00 2001 From: ahalekelly <7078138+ahalekelly@users.noreply.github.com> Date: Fri, 2 Oct 2026 01:41:22 -0700 Subject: [PATCH 1/5] fix(client-runtime): apply a thread's resume catch-up in one update A resumed thread subscription applied each replayed frame separately, so the thread screen re-rendered through stale intermediate states until the completion marker arrived. Hold items until the marker and reduce them in one state update; live events after it still apply as they arrive. Co-Authored-By: Claude Opus 5.5 (1M context) (cherry picked from commit 96021ba267d97bab6619f59ea5c8c2ed1ea14cf6) --- .../src/state/threads-sync.test.ts | 51 ++++++++++--------- packages/client-runtime/src/state/threads.ts | 39 +++++++++++--- 2 files changed, 59 insertions(+), 31 deletions(-) diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index eaf1751a1c2d..54e8239e0599 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -294,6 +294,15 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o }; }); +const awaitSubscriptionCount = ( + harness: { readonly subscriptionCount: SubscriptionRef.SubscriptionRef }, + count: number, +) => + SubscriptionRef.changes(harness.subscriptionCount).pipe( + Stream.filter((current) => current >= count), + Stream.runHead, + ); + const snapshot = ( projection: OrchestrationV2ThreadProjection, snapshotSequence = 1, @@ -2082,34 +2091,28 @@ describe("EnvironmentThreads", () => { }), ); - it.effect("keeps replayed updates synchronizing until the completion marker arrives", () => + it.effect("publishes a 20-event catch-up as one state change at its marker", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); - yield* awaitThreadState( - harness.observed, - (value) => value.status === "synchronizing" && Option.isSome(value.data), - ); + yield* awaitSubscriptionCount(harness, 1); expect(yield* Ref.get(harness.lastRequestCompletionMarker)).toBe(true); - - yield* Queue.offer( - harness.inputs, - titleUpdated("Caught-up title", CACHED_SNAPSHOT_SEQUENCE + 1), - ); - const catchingUp = yield* awaitThreadState( - harness.observed, - (value) => - value.status === "synchronizing" && - Option.isSome(value.data) && - value.data.value.thread.title === "Caught-up title", - ); - expect(catchingUp.status).toBe("synchronizing"); - + yield* Queue.clear(harness.observed); + // Separate frames, as the server sends a replay. + for (let index = 1; index <= 20; index += 1) { + yield* Queue.offer( + harness.inputs, + titleUpdated(`Title ${index}`, CACHED_SNAPSHOT_SEQUENCE + index), + ); + yield* Effect.yieldNow; + } yield* Queue.offer(harness.inputs, synchronized()); - const live = yield* awaitThreadState( - harness.observed, - (value) => value.status === "live" && Option.isSome(value.data), - ); - expect(Option.getOrThrow(live.data).thread.title).toBe("Caught-up title"); + const live = yield* Queue.take(harness.observed); + expect(live.status).toBe("live"); + expect(Option.getOrThrow(live.data).thread.title).toBe("Title 20"); + + yield* Queue.offer(harness.inputs, titleUpdated("Live title", CACHED_SNAPSHOT_SEQUENCE + 21)); + const next = yield* Queue.take(harness.observed); + expect(Option.getOrThrow(next.data).thread.title).toBe("Live title"); }), ); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index c7b22d9dcdff..ce5e0575dabd 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -602,10 +602,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ) { if (item.kind === "synchronized") { yield* Ref.set(awaitingCompletion, false); - yield* SubscriptionRef.update(state, (current) => - Option.isSome(current.data) && current.status !== "deleted" && Option.isNone(current.error) - ? { ...current, status: "live" as const, error: Option.none() } - : current, + yield* SubscriptionRef.updateSome(state, (current) => + Option.isSome(current.data) && + current.status !== "deleted" && + current.status !== "live" && + Option.isNone(current.error) + ? Option.some({ ...current, status: "live" as const }) + : Option.none(), ); return; } @@ -637,11 +640,30 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* applyEventsLocked([item]); }); - const applyItems = Effect.fn("EnvironmentThreadState.applyItems")(function* ( - items: ReadonlyArray, + // Items received before the server's completion marker apply together when it + // arrives, so a replayed resume renders its final state once. A restarted + // attempt keeps them: its replay repeats them from the unmoved cursor, and + // already applied sequences are skipped. + let catchUp: OrchestrationV2ThreadStreamItem[] = []; + const receiveItems = Effect.fn("EnvironmentThreadState.receiveItems")(function* ( + received: ReadonlyArray, ) { yield* applyLock.withPermits(1)( Effect.gen(function* () { + catchUp.push(...received); + const completes = received.some((item) => item.kind === "synchronized"); + if (catchUp.length === 0) return; + if (!completes && (yield* Ref.get(awaitingCompletion))) { + // A warm resume keeps its "live" status until it learns it is behind. + yield* SubscriptionRef.updateSome(state, (current) => + current.status === "live" + ? Option.some({ ...current, status: "synchronizing" as const }) + : Option.none(), + ); + return; + } + const items = catchUp; + catchUp = []; let events: SequencedItem[] = []; for (const item of items) { if ( @@ -651,6 +673,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make events.push(item); continue; } + // Clearing the flag first lets the events before the marker publish + // "live", so a catch-up ending in the marker is one state change. + if (item.kind === "synchronized") yield* Ref.set(awaitingCompletion, false); yield* applyEventsLocked(events); events = []; yield* applyItemLocked(item); @@ -1031,7 +1056,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEachArray(applyItems)), + ).pipe(Stream.runForEachArray(receiveItems)), ); return state; From 20753c0067d4adf03375afb5a6e98379d4e5de5c Mon Sep 17 00:00:00 2001 From: ahalekelly <7078138+ahalekelly@users.noreply.github.com> Date: Thu, 29 Oct 2026 22:38:54 -0700 Subject: [PATCH 2/5] fix(client-runtime): a new thread attempt drops the previous attempt's held catch-up Each retry before the completion marker replays the same items from the unmoved cursor, so keeping the held items grew the buffer by one replay per retry. Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/state/threads.ts | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index ce5e0575dabd..2932e2e6757b 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -641,10 +641,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }); // Items received before the server's completion marker apply together when it - // arrives, so a replayed resume renders its final state once. A restarted - // attempt keeps them: its replay repeats them from the unmoved cursor, and - // already applied sequences are skipped. + // arrives, so a replayed resume renders its final state once. A new attempt + // drops them: they were never applied, so its cursor replays them. let catchUp: OrchestrationV2ThreadStreamItem[] = []; + const startAttempt = applyLock.withPermits(1)( + Effect.sync(() => { + catchUp = []; + }), + ); const receiveItems = Effect.fn("EnvironmentThreadState.receiveItems")(function* ( received: ReadonlyArray, ) { @@ -962,6 +966,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make subscribeDynamic( ORCHESTRATION_V2_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { + yield* startAttempt; 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 From cd452beaf17f5f4b4640ce5d616bdec9c242064d Mon Sep 17 00:00:00 2001 From: ahalekelly <7078138+ahalekelly@users.noreply.github.com> Date: Fri, 2 Oct 2026 03:18:55 -0700 Subject: [PATCH 3/5] fix(client-runtime): ignore a replaced thread subscription's late items Stream.switchMap still delivers values an old attempt buffered after the next attempt's makeInput ran, and a foreground resubscribe reuses the session. Since a new attempt clears the held catch-up, a late event and completion marker from the replaced attempt could then apply without the dropped prefix and advance the cursor past it, so the next attempt skipped those events. A late snapshot could also regress the cursor. Tag thread stream values with their attempt's input (subscribeDynamicWithInput) and drop items from any other attempt. Starting an attempt and receiving items both run under applyLock, so a batch lands wholly before or after an attempt starts. Also take the observer's replayed initial state before clearing the queue in the 20-event catch-up test, so it cannot be read as the first change. Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/rpc/client.ts | 29 +++- .../src/state/threads-sync.test.ts | 132 ++++++++++++++++-- packages/client-runtime/src/state/threads.ts | 26 ++-- 3 files changed, 163 insertions(+), 24 deletions(-) diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 2832a4e23155..1a56989a600a 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -251,6 +251,7 @@ function subscribeDynamicMapped( makeInput: (session: RpcSession) => Effect.Effect>, mapStream: ( session: RpcSession, + input: EnvironmentRpcInput, stream: Stream.Stream, EnvironmentRpcStreamFailure>, ) => Stream.Stream>, options?: SubscriptionOptions, @@ -299,7 +300,7 @@ function subscribeDynamicMapped( method: tag, input, }); - const stream = mapStream(session, method(input)).pipe( + const stream = mapStream(session, input, method(input)).pipe( Stream.onFirst(() => Effect.sync(() => { expectedFailureRetries = 0; @@ -397,7 +398,7 @@ export function subscribeDynamic( EnvironmentRpcStreamFailure, EnvironmentSupervisor.EnvironmentSupervisor > { - return subscribeDynamicMapped(tag, makeInput, (_session, stream) => stream, options); + return subscribeDynamicMapped(tag, makeInput, (_session, _input, stream) => stream, options); } /** Tags each value before `switchMap` can buffer it across a session change. */ @@ -413,7 +414,29 @@ export function subscribeDynamicWithSession stream.pipe(Stream.map((value) => [session, value] as const)), + (session, _input, stream) => stream.pipe(Stream.map((value) => [session, value] as const)), + options, + ); +} + +/** + * Tags each value with the input of the attempt that produced it. Every + * `makeInput` call starts an attempt, including a resubscribe on the same + * session; `makeInput` must return a fresh object so the input identifies it. + */ +export function subscribeDynamicWithInput( + tag: TTag, + makeInput: (session: RpcSession) => Effect.Effect>, + options?: SubscriptionOptions, +): Stream.Stream< + readonly [input: EnvironmentRpcInput, value: EnvironmentRpcStreamValue], + EnvironmentRpcStreamFailure, + EnvironmentSupervisor.EnvironmentSupervisor +> { + return subscribeDynamicMapped( + tag, + makeInput, + (_session, input, stream) => stream.pipe(Stream.map((value) => [input, value] as const)), options, ); } diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 54e8239e0599..47cdfe0b3227 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -21,8 +21,8 @@ import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; -import * as Scope from "effect/Scope"; import * as Scheduler from "effect/Scheduler"; +import * as Scope from "effect/Scope"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -122,6 +122,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly getTurnItem?: (input: { readonly itemId: TurnItemId; }) => Effect.Effect<{ readonly item: OrchestrationV2TurnItem | null }>; + readonly streamFor?: ( + subscription: number, + ) => Stream.Stream | undefined; }) { const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); @@ -153,17 +156,19 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly acceptCompactTurnItems?: true; }) => Stream.unwrap( - Ref.set(lastSubscribeAfterSequence, input.afterSequence).pipe( - Effect.andThen( - Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker === true), - ), - Effect.andThen(Ref.set(lastAcceptBoundedSnapshot, input.acceptBoundedSnapshot)), - Effect.andThen(Ref.set(lastAcceptCompactTurnItems, input.acceptCompactTurnItems)), - Effect.andThen(SubscriptionRef.update(subscriptionCount, (count) => count + 1)), - Effect.as( - options?.completionMarker === true - ? streamFrom(inputs) - : Stream.concat(Stream.make(synchronized()), streamFrom(inputs)), + Effect.all([ + Ref.set(lastSubscribeAfterSequence, input.afterSequence), + Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker === true), + Ref.set(lastAcceptBoundedSnapshot, input.acceptBoundedSnapshot), + Ref.set(lastAcceptCompactTurnItems, input.acceptCompactTurnItems), + ]).pipe( + Effect.andThen(SubscriptionRef.updateAndGet(subscriptionCount, (count) => count + 1)), + Effect.map( + (count) => + options?.streamFor?.(count) ?? + (options?.completionMarker === true + ? streamFrom(inputs) + : Stream.concat(Stream.make(synchronized()), streamFrom(inputs))), ), ), ), @@ -2096,6 +2101,8 @@ describe("EnvironmentThreads", () => { const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); yield* awaitSubscriptionCount(harness, 1); expect(yield* Ref.get(harness.lastRequestCompletionMarker)).toBe(true); + // The observer replays the current state when it starts; take it so clearing drops it. + yield* Queue.take(harness.observed); yield* Queue.clear(harness.observed); // Separate frames, as the server sends a replay. for (let index = 1; index <= 20; index += 1) { @@ -2147,6 +2154,107 @@ describe("EnvironmentThreads", () => { }), ); + // A long batch yields to the scheduler while it applies, so the next attempt + // can start while the replaced attempt's batch and items are still landing. + const LONG_BATCH = 1000; + + it.effect( + "does not apply a replaced attempt's snapshot after the next attempt reads its cursor", + () => + Effect.gen(function* () { + const sent = yield* Deferred.make(); + const harness = yield* makeHarness({ + cached: BASE_PROJECTION, + streamFor: (subscription) => + subscription === 1 + ? Stream.fromArray([ + synchronized(), + snapshot(BASE_PROJECTION, 8), + ...Array.from({ length: LONG_BATCH }, (_, index) => + titleUpdated(`Title ${index + 9}`, index + 9), + ), + snapshot({ + ...BASE_PROJECTION, + thread: { ...BASE_PROJECTION.thread, title: "Stale" }, + }), + ]).pipe( + Stream.rechunk(1), + Stream.concat( + Stream.fromEffect(Deferred.succeed(sent, undefined)).pipe(Stream.drain), + ), + Stream.concat(Stream.never), + ) + : undefined, + }); + yield* Deferred.await(sent); + yield* Queue.offer(harness.wakeups, "application-active"); + yield* awaitSubscriptionCount(harness, 2); + yield* Queue.offer(harness.inputs, titleUpdated("Live", LONG_BATCH + 9)); + const titles: Array = []; + yield* awaitThreadState(harness.observed, (value) => { + titles.push(Option.getOrNull(value.data)?.thread.title); + return Option.getOrNull(value.data)?.thread.title === "Live"; + }); + // The next attempt started with the snapshot still buffered: its cursor + // is past the replaced snapshot's sequence but short of the batch end. + const cursor = yield* Ref.get(harness.lastSubscribeAfterSequence); + expect(cursor).toBeGreaterThan(8); + expect(cursor).toBeLessThan(LONG_BATCH + 8); + expect(titles).not.toContain("Stale"); + }).pipe( + // A small budget lets the next attempt start between the consumer's + // batches, before it takes the queued snapshot. + Effect.provideService(Scheduler.MaxOpsBeforeYield, 50), + ), + ); + + it.effect("does not apply a replaced attempt's late marker without its dropped prefix", () => + Effect.gen(function* () { + const sent = yield* Deferred.make(); + const replay = [ + questionResolved(CACHED_SNAPSHOT_SEQUENCE + 1), + ...Array.from({ length: 40 }, (_, index) => + titleUpdated(`Title ${index}`, CACHED_SNAPSHOT_SEQUENCE + 2 + index), + ), + synchronized(), + ]; + const harness = yield* makeHarness({ + cached: { ...BASE_PROJECTION, runtimeRequests: [PENDING_QUESTION] }, + completionMarker: true, + streamFor: (subscription) => + subscription === 1 + ? Stream.fromArray(replay).pipe( + Stream.rechunk(1), + Stream.concat( + Stream.fromEffect(Deferred.succeed(sent, undefined)).pipe(Stream.drain), + ), + Stream.concat(Stream.never), + ) + : undefined, + }); + yield* Deferred.await(sent); + yield* Queue.offer(harness.wakeups, "application-active"); + yield* awaitSubscriptionCount(harness, 2); + // The next attempt started before the replaced attempt's buffered marker landed. + expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE); + yield* Queue.offerAll(harness.inputs, [ + ...replay, + titleUpdated("Live", CACHED_SNAPSHOT_SEQUENCE + 42), + ]); + const live = yield* awaitThreadState( + harness.observed, + (value) => Option.getOrNull(value.data)?.thread.title === "Live", + ); + expect(Option.getOrThrow(live.data).runtimeRequests.map((request) => request.status)).toEqual( + ["resolved"], + ); + }).pipe( + // The replaced attempt's items fill the switch buffer, and a small budget + // lets the next attempt start while the consumer still drains them. + Effect.provideService(Scheduler.MaxOpsBeforeYield, 10), + ), + ); + it.effect("skips an unknown event type and resumes after it", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 2932e2e6757b..00fbab54a280 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -32,7 +32,7 @@ import * as ConnectionWakeups from "../connection/wakeups.ts"; import * as Persistence from "../platform/persistence.ts"; import { runCachePersistence } from "./cachePersistence.ts"; import * as ManagedRelay from "../relay/managedRelay.ts"; -import { request, subscribeDynamic } from "../rpc/client.ts"; +import { request, subscribeDynamicWithInput } from "../rpc/client.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts"; import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; @@ -640,22 +640,28 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* applyEventsLocked([item]); }); - // Items received before the server's completion marker apply together when it - // arrives, so a replayed resume renders its final state once. A new attempt - // drops them: they were never applied, so its cursor replays them. + // Items held until the completion marker apply together, so a replay-only resume renders once. + // A new attempt drops the previous attempt's held and late items (unapplied, so its cursor + // replays them). Both run under applyLock: each batch lands wholly before or after an attempt. + let attempt: unknown; let catchUp: OrchestrationV2ThreadStreamItem[] = []; const startAttempt = applyLock.withPermits(1)( Effect.sync(() => { + attempt = undefined; catchUp = []; }), ); const receiveItems = Effect.fn("EnvironmentThreadState.receiveItems")(function* ( - received: ReadonlyArray, + received: ReadonlyArray, ) { yield* applyLock.withPermits(1)( Effect.gen(function* () { - catchUp.push(...received); - const completes = received.some((item) => item.kind === "synchronized"); + let completes = false; + for (const [input, item] of received) { + if (input !== attempt) continue; + catchUp.push(item); + if (item.kind === "synchronized") completes = true; + } if (catchUp.length === 0) return; if (!completes && (yield* Ref.get(awaitingCompletion))) { // A warm resume keeps its "live" status until it learns it is behind. @@ -963,7 +969,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* markSynchronizing; yield* Effect.forkScoped( - subscribeDynamic( + subscribeDynamicWithInput( ORCHESTRATION_V2_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { yield* startAttempt; @@ -1040,7 +1046,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const canResume = Option.isSome(current.data); const acceptBoundedSnapshot = yield* Ref.get(acceptsBoundedSocketSnapshots); - return { + const input = { threadId, ...(canResume ? { afterSequence: sequence } : {}), requestCompletionMarker: true as const, @@ -1054,6 +1060,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make : {}), acceptDedupedBodies: true as const, }; + attempt = input; + return input; }), { onDefect: () => setStreamError("Could not synchronize the thread."), From 53e821e1bcd718b1d019580f9b04af584290ed47 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 6 Nov 2026 08:42:06 +0000 Subject: [PATCH 4/5] test(client-runtime): keep the late-marker race test valid after main's catch-up changes --- packages/client-runtime/src/state/threads-sync.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 63ae7379e160..70983ce9880c 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -2326,7 +2326,7 @@ describe("EnvironmentThreads", () => { synchronized(), ]; const harness = yield* makeHarness({ - cached: { ...BASE_PROJECTION, runtimeRequests: [PENDING_QUESTION] }, + cached: PROJECTION_WITH_QUESTION, completionMarker: true, streamFor: (subscription) => subscription === 1 @@ -2358,7 +2358,7 @@ describe("EnvironmentThreads", () => { }).pipe( // The replaced attempt's items fill the switch buffer, and a small budget // lets the next attempt start while the consumer still drains them. - Effect.provideService(Scheduler.MaxOpsBeforeYield, 10), + Effect.provideService(Scheduler.MaxOpsBeforeYield, 5), ), ); From 1b642a6226e9762dca3bfdd84b42e4ad41cd2300 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 10 Nov 2026 05:09:54 +0000 Subject: [PATCH 5/5] test(client-runtime): finish the socket snapshot before loading older history A fresh subscribe now holds items until the completion marker, so the older-page cancellation test sends the marker the server sends after a snapshot. --- packages/client-runtime/src/state/threads-atoms.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index 460f7e593990..9aaa605998a5 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -304,6 +304,8 @@ describe("createEnvironmentThreadStateAtoms", () => { }); const unmount = h.registry.mount(h.stateAtom); const first = yield* Queue.take(h.subscriptions); + yield* Queue.offer(first.events, { kind: "synchronized" }); + yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live"); const loading = yield* h.loadEarlier().pipe(Effect.forkScoped); const older = yield* Queue.take(h.olderLoads); expect(h.registry.get(h.stateAtom).history.loading).toBe(true);