From b623dc23b6bc28eef0a65bbc07cbbaefb60b3eff Mon Sep 17 00:00:00 2001 From: Daniel Vernon Date: Tue, 4 Aug 2026 17:38:49 +0100 Subject: [PATCH 1/2] perf(client): batch large thread sync updates --- .../src/state/threads-sync.test.ts | 54 ++++- packages/client-runtime/src/state/threads.ts | 215 ++++++++++++------ tasks/todo.md | 13 ++ 3 files changed, 211 insertions(+), 71 deletions(-) create mode 100644 tasks/todo.md diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index c2df434e8e7..f0ca0fb0f78 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -133,10 +133,12 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly cached?: OrchestrationThread; readonly httpSnapshot?: Option.Option; readonly completionMarker?: boolean; + readonly eventBatchSize?: number; }) { const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); const latest = yield* Ref.make(EMPTY_ENVIRONMENT_THREAD_STATE); + const statePublicationCount = yield* Ref.make(0); const retryCount = yield* Ref.make(0); const subscriptionCount = yield* Ref.make(0); const loaderCalls = yield* Ref.make(0); @@ -221,7 +223,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); - const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe( + const threadState = yield* makeEnvironmentThreadState(THREAD_ID, { + eventBatchSize: options?.eventBatchSize ?? 1, + }).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ThreadSnapshotLoader, snapshotLoader), @@ -232,7 +236,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o ); yield* SubscriptionRef.changes(threadState).pipe( Stream.runForEach((state) => - Ref.set(latest, state).pipe(Effect.andThen(Queue.offer(observed, state))), + Ref.update(statePublicationCount, (count) => count + 1).pipe( + Effect.andThen(Ref.set(latest, state)), + Effect.andThen(Queue.offer(observed, state)), + ), ), Effect.forkScoped, ); @@ -241,6 +248,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o inputs, observed, latest, + statePublicationCount, retryCount, subscriptionCount, loaderCalls, @@ -347,6 +355,48 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("applies a live burst in order with one state publication", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + cached: BASE_THREAD, + completionMarker: true, + eventBatchSize: 64, + }); + yield* awaitThreadState( + harness.observed, + (value) => value.status === "synchronizing" && Option.isSome(value.data), + ); + const publicationsBeforeBurst = yield* Ref.get(harness.statePublicationCount); + + const finalSequence = CACHED_SNAPSHOT_SEQUENCE + 63; + for (let sequence = CACHED_SNAPSHOT_SEQUENCE + 1; sequence <= finalSequence; sequence += 1) { + yield* Queue.offer( + harness.inputs, + titleUpdated( + sequence === finalSequence + ? "Final title" + : sequence === CACHED_SNAPSHOT_SEQUENCE + 1 + ? "First title" + : "Interim title", + sequence, + ), + ); + } + yield* Queue.offer(harness.inputs, synchronized()); + + const state = yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.title === "Final title", + ); + + expect(Option.getOrThrow(state.data).title).toBe("Final title"); + expect(yield* Ref.get(harness.statePublicationCount)).toBe(publicationsBeforeBurst + 1); + }), + ); + it.effect("reduces live events and persists the latest thread", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 06b5428ca58..f0d39f50a6e 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -7,6 +7,7 @@ import { type ThreadId as ThreadIdType, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; @@ -32,6 +33,111 @@ import { type EnvironmentThreadStatus, } from "./threadState.ts"; +// Coalesce one render-sized burst so web and mobile derive their large thread +// views once instead of once per streamed event. The size cap bounds catch-up +// memory and prevents a quiet connection from waiting indefinitely. +const THREAD_EVENT_BATCH_WINDOW = Duration.millis(16); +const THREAD_EVENT_BATCH_MAX_SIZE = 64; + +interface ThreadStreamBatchReduction { + readonly state: EnvironmentThreadState; + readonly lastSequence: number; + readonly awaitingCompletion: boolean; + readonly threadChanged: boolean; + readonly threadDeleted: boolean; +} + +export interface EnvironmentThreadStateOptions { + readonly eventBatchSize?: number; +} + +function reduceThreadStreamItems( + currentState: EnvironmentThreadState, + currentSequence: number, + currentAwaitingCompletion: boolean, + items: ReadonlyArray, +): ThreadStreamBatchReduction { + let state = currentState; + let lastSequence = currentSequence; + let awaitingCompletion = currentAwaitingCompletion; + let thread = Option.getOrNull(currentState.data); + let threadChanged = false; + let threadDeleted = false; + + for (const item of items) { + if (item.kind === "synchronized") { + awaitingCompletion = false; + if (thread !== null && state.status !== "deleted") { + state = { + data: state.data, + status: "live", + error: Option.none(), + }; + } + continue; + } + + if (item.kind === "snapshot") { + lastSequence = item.snapshot.snapshotSequence; + thread = item.snapshot.thread; + threadChanged = true; + threadDeleted = false; + state = { + data: Option.some(thread), + status: awaitingCompletion ? "synchronizing" : "live", + error: Option.none(), + }; + continue; + } + + if (item.event.sequence <= lastSequence) { + continue; + } + lastSequence = item.event.sequence; + + if (thread === null) { + if (item.event.type === "thread.deleted") { + awaitingCompletion = false; + threadDeleted = true; + state = { + data: Option.none(), + status: "deleted", + error: Option.none(), + }; + } + continue; + } + + const result = applyThreadDetailEvent(thread, item.event); + if (result.kind === "updated") { + thread = result.thread; + threadChanged = true; + state = { + data: Option.some(thread), + status: awaitingCompletion ? "synchronizing" : "live", + error: Option.none(), + }; + } else if (result.kind === "deleted") { + awaitingCompletion = false; + thread = null; + threadDeleted = true; + state = { + data: Option.none(), + status: "deleted", + error: Option.none(), + }; + } + } + + return { + state, + lastSequence, + awaitingCompletion, + threadChanged, + threadDeleted, + }; +} + function statusWithoutLiveData(data: Option.Option): EnvironmentThreadStatus { return Option.isSome(data) ? "cached" : "empty"; } @@ -50,12 +156,14 @@ function shouldPersistThread(thread: OrchestrationThread): boolean { export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* ( threadId: ThreadIdType, + options?: EnvironmentThreadStateOptions, ) { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const snapshotLoader = yield* ThreadSnapshotLoader; const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; + const eventBatchSize = options?.eventBatchSize ?? THREAD_EVENT_BATCH_MAX_SIZE; const cached = yield* cache.loadThread(environmentId, threadId).pipe( Effect.catch((error) => Effect.logWarning("Could not load cached thread.").pipe( @@ -141,81 +249,47 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ), ); - const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* ( - thread: OrchestrationThread, + const applyItems = Effect.fn("EnvironmentThreadState.applyItems")(function* ( + items: ReadonlyArray, ) { - const waiting = yield* Ref.get(awaitingCompletion); - yield* SubscriptionRef.set(state, { - data: Option.some(thread), - status: waiting ? "synchronizing" : "live", - error: Option.none(), - }); - // Active threads can update many times per second and retain large tool - // payloads. The server remains the source of truth while a turn is active; - // persist once it settles so cache encoding stays off the streaming path. - if (shouldPersistThread(thread)) { - const snapshotSequence = yield* SubscriptionRef.get(lastSequence); - yield* Queue.offer(persistence, { snapshotSequence, thread }); - } - }); - - const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { - yield* Ref.set(awaitingCompletion, false); - yield* SubscriptionRef.set(state, { - data: Option.none(), - status: "deleted", - error: Option.none(), - }); - yield* cache.removeThread(environmentId, threadId).pipe( - Effect.catch((error) => - Effect.logWarning("Could not remove the cached thread.").pipe( - Effect.annotateLogs({ - environmentId, - threadId, - error: error.message, - }), - ), - ), + const currentState = yield* SubscriptionRef.get(state); + const reduction = reduceThreadStreamItems( + currentState, + yield* SubscriptionRef.get(lastSequence), + yield* Ref.get(awaitingCompletion), + items, ); - }); - const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( - item: OrchestrationThreadStreamItem, - ) { - if (item.kind === "synchronized") { - yield* Ref.set(awaitingCompletion, false); - yield* SubscriptionRef.update(state, (current) => - Option.isSome(current.data) && current.status !== "deleted" - ? { ...current, status: "live" as const, error: Option.none() } - : current, - ); - return; + yield* SubscriptionRef.set(lastSequence, reduction.lastSequence); + yield* Ref.set(awaitingCompletion, reduction.awaitingCompletion); + if (reduction.state !== currentState) { + yield* SubscriptionRef.set(state, reduction.state); } - if (item.kind === "snapshot") { - yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence); - yield* setThread(item.snapshot.thread); - return; - } - - const sequence = yield* SubscriptionRef.get(lastSequence); - if (item.event.sequence <= sequence) { + if (reduction.threadDeleted) { + yield* cache.removeThread(environmentId, threadId).pipe( + Effect.catch((error) => + Effect.logWarning("Could not remove the cached thread.").pipe( + Effect.annotateLogs({ + environmentId, + threadId, + error: error.message, + }), + ), + ), + ); return; } - yield* SubscriptionRef.set(lastSequence, item.event.sequence); - const current = yield* SubscriptionRef.get(state); - if (Option.isNone(current.data)) { - if (item.event.type === "thread.deleted") { - yield* setDeleted(); - } - return; - } - const result = applyThreadDetailEvent(current.data.value, item.event); - if (result.kind === "updated") { - yield* setThread(result.thread); - } else if (result.kind === "deleted") { - yield* setDeleted(); + const thread = Option.getOrNull(reduction.state.data); + // Active threads can update many times per second and retain large tool + // payloads. The server remains the source of truth while a turn is active; + // persist once it settles so cache encoding stays off the streaming path. + if (reduction.threadChanged && thread !== null && shouldPersistThread(thread)) { + yield* Queue.offer(persistence, { + snapshotSequence: reduction.lastSequence, + thread, + }); } }); @@ -269,7 +343,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); const httpSnapshot = yield* snapshotLoader.load(prepared, threadId); if (Option.isSome(httpSnapshot)) { - yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); + yield* applyItems([{ kind: "snapshot", snapshot: httpSnapshot.value }]); current = yield* SubscriptionRef.get(state); } } @@ -295,7 +369,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEach(applyItem)), + ).pipe( + Stream.groupedWithin(eventBatchSize, THREAD_EVENT_BATCH_WINDOW), + Stream.runForEach(applyItems), + ), ); yield* Effect.addFinalizer(() => diff --git a/tasks/todo.md b/tasks/todo.md new file mode 100644 index 00000000000..c8a32986cc3 --- /dev/null +++ b/tasks/todo.md @@ -0,0 +1,13 @@ +# Large-thread sync optimization + +- [x] Add bounded batching for live thread stream items. +- [x] Add regression coverage for ordered, single-publication batches. +- [x] Run focused client-runtime checks and review the diff. +- [ ] Commit the focused sync fix on `perf/large-thread-sync`. + +## Review/results + +- Client-side batching is implemented in the shared web/mobile runtime. +- Focused sync, reducer, and atom tests pass. +- Targeted typecheck, lint, formatting, and diff checks pass; tsgo reports one unrelated existing suggestion in `src/relay/discovery.ts`. +- Review found no remaining standards issue; server replay/snapshot and per-event reducer costs remain follow-up scope. From 783fd023c69f7ebc9c0ae657ce124cce77d94587 Mon Sep 17 00:00:00 2001 From: Daniel Vernon Date: Tue, 4 Aug 2026 18:03:09 +0100 Subject: [PATCH 2/2] fix(sync): retain settled snapshots across event batches --- .../src/state/threads-sync.test.ts | 75 +++++++++++++++++++ packages/client-runtime/src/state/threads.ts | 28 +++---- tasks/todo.md | 13 ---- 3 files changed, 86 insertions(+), 30 deletions(-) delete mode 100644 tasks/todo.md diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index f0ca0fb0f78..491ab8ef03e 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -8,6 +8,7 @@ import { TurnId, type OrchestrationThread, type OrchestrationThreadDetailSnapshot, + type OrchestrationSession, type OrchestrationThreadStreamItem, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; @@ -281,6 +282,44 @@ const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem => const synchronized = (): OrchestrationThreadStreamItem => ({ kind: "synchronized" }); +const sessionUpdated = ( + status: OrchestrationSession["status"], + sequence: number, + activeTurnId: TurnId | null, +): OrchestrationThreadStreamItem => ({ + kind: "event", + event: { + eventId: EventId.make(`event-session-${sequence}`), + sequence, + occurredAt: + sequence === CACHED_SNAPSHOT_SEQUENCE + 1 + ? "2026-04-01T08:00:00.000Z" + : "2026-04-01T09:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.session-set", + payload: { + threadId: THREAD_ID, + session: { + threadId: THREAD_ID, + status, + providerName: "codex", + runtimeMode: "full-access", + activeTurnId, + lastError: null, + updatedAt: + sequence === CACHED_SNAPSHOT_SEQUENCE + 1 + ? "2026-04-01T08:00:00.000Z" + : "2026-04-01T09:00:00.000Z", + }, + }, + }, +}); + const titleUpdated = (title: string, sequence = 2): OrchestrationThreadStreamItem => ({ kind: "event", event: { @@ -397,6 +436,42 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("persists a settled snapshot before a batched turn starts", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + cached: ACTIVE_THREAD, + eventBatchSize: 2, + }); + + yield* Queue.offer( + harness.inputs, + sessionUpdated("ready", CACHED_SNAPSHOT_SEQUENCE + 1, null), + ); + yield* Queue.offer( + harness.inputs, + sessionUpdated("running", CACHED_SNAPSHOT_SEQUENCE + 2, TurnId.make("turn-2")), + ); + yield* Queue.offer(harness.inputs, synchronized()); + + const state = yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.session?.status === "running" && + value.data.value.session.activeTurnId === TurnId.make("turn-2"), + ); + + expect(Option.getOrThrow(state.data).session?.status).toBe("running"); + yield* TestClock.adjust("500 millis"); + yield* Effect.yieldNow; + + const saved = (yield* Ref.get(harness.savedThreads)).at(-1); + expect(saved?.snapshotSequence).toBe(CACHED_SNAPSHOT_SEQUENCE + 1); + expect(saved?.thread.session?.status).toBe("ready"); + }), + ); + it.effect("reduces live events and persists the latest thread", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index f0d39f50a6e..a554ba7e3ae 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -33,9 +33,6 @@ import { type EnvironmentThreadStatus, } from "./threadState.ts"; -// Coalesce one render-sized burst so web and mobile derive their large thread -// views once instead of once per streamed event. The size cap bounds catch-up -// memory and prevents a quiet connection from waiting indefinitely. const THREAD_EVENT_BATCH_WINDOW = Duration.millis(16); const THREAD_EVENT_BATCH_MAX_SIZE = 64; @@ -43,8 +40,8 @@ interface ThreadStreamBatchReduction { readonly state: EnvironmentThreadState; readonly lastSequence: number; readonly awaitingCompletion: boolean; - readonly threadChanged: boolean; readonly threadDeleted: boolean; + readonly persistableSnapshot: OrchestrationThreadDetailSnapshot | null; } export interface EnvironmentThreadStateOptions { @@ -61,8 +58,8 @@ function reduceThreadStreamItems( let lastSequence = currentSequence; let awaitingCompletion = currentAwaitingCompletion; let thread = Option.getOrNull(currentState.data); - let threadChanged = false; let threadDeleted = false; + let persistableSnapshot: OrchestrationThreadDetailSnapshot | null = null; for (const item of items) { if (item.kind === "synchronized") { @@ -80,8 +77,8 @@ function reduceThreadStreamItems( if (item.kind === "snapshot") { lastSequence = item.snapshot.snapshotSequence; thread = item.snapshot.thread; - threadChanged = true; threadDeleted = false; + persistableSnapshot = shouldPersistThread(thread) ? item.snapshot : null; state = { data: Option.some(thread), status: awaitingCompletion ? "synchronizing" : "live", @@ -99,6 +96,7 @@ function reduceThreadStreamItems( if (item.event.type === "thread.deleted") { awaitingCompletion = false; threadDeleted = true; + persistableSnapshot = null; state = { data: Option.none(), status: "deleted", @@ -111,7 +109,9 @@ function reduceThreadStreamItems( const result = applyThreadDetailEvent(thread, item.event); if (result.kind === "updated") { thread = result.thread; - threadChanged = true; + if (shouldPersistThread(thread)) { + persistableSnapshot = { snapshotSequence: lastSequence, thread }; + } state = { data: Option.some(thread), status: awaitingCompletion ? "synchronizing" : "live", @@ -121,6 +121,7 @@ function reduceThreadStreamItems( awaitingCompletion = false; thread = null; threadDeleted = true; + persistableSnapshot = null; state = { data: Option.none(), status: "deleted", @@ -133,8 +134,8 @@ function reduceThreadStreamItems( state, lastSequence, awaitingCompletion, - threadChanged, threadDeleted, + persistableSnapshot, }; } @@ -281,15 +282,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make return; } - const thread = Option.getOrNull(reduction.state.data); - // Active threads can update many times per second and retain large tool - // payloads. The server remains the source of truth while a turn is active; - // persist once it settles so cache encoding stays off the streaming path. - if (reduction.threadChanged && thread !== null && shouldPersistThread(thread)) { - yield* Queue.offer(persistence, { - snapshotSequence: reduction.lastSequence, - thread, - }); + if (reduction.persistableSnapshot !== null) { + yield* Queue.offer(persistence, reduction.persistableSnapshot); } }); diff --git a/tasks/todo.md b/tasks/todo.md deleted file mode 100644 index c8a32986cc3..00000000000 --- a/tasks/todo.md +++ /dev/null @@ -1,13 +0,0 @@ -# Large-thread sync optimization - -- [x] Add bounded batching for live thread stream items. -- [x] Add regression coverage for ordered, single-publication batches. -- [x] Run focused client-runtime checks and review the diff. -- [ ] Commit the focused sync fix on `perf/large-thread-sync`. - -## Review/results - -- Client-side batching is implemented in the shared web/mobile runtime. -- Focused sync, reducer, and atom tests pass. -- Targeted typecheck, lint, formatting, and diff checks pass; tsgo reports one unrelated existing suggestion in `src/relay/discovery.ts`. -- Review found no remaining standards issue; server replay/snapshot and per-event reducer costs remain follow-up scope.