perf(client): batch large thread sync updates (upstream pingdotgg#5344) (#328)

Imported from pingdotgg#5344 at source SHA
783fd02 (commits b623dc2 + 783fd02
squashed into one provenance commit).

Imported behavior:
- `reduceThreadStreamItems`, a pure reducer that folds a batch of thread
  stream items into one state and one persistable snapshot.
- `Stream.groupedWithin(64, 16ms)` on the live subscription so a burst of
  thread events publishes the `SubscriptionRef` once instead of per event,
  and web/mobile stop rebuilding large thread views per streamed event.
- `eventBatchSize` on `EnvironmentThreadStateOptions`, plus the upstream
  regression tests for ordered single-publication bursts and for persisting
  a settled snapshot when a batch ends with a non-persistable turn start.

Local adaptations:
- Kept our `httpSnapshotLoadAttempted` guard around the HTTP snapshot
  fallback; the call now goes through `applyItems([...])`.
- Restored `setDeleted` (removed upstream) for the terminal
  `thread-deleted` subscription failure, which never reaches the item
  stream and so cannot go through the batch reducer. Cache removal is
  shared with the reducer path via `removeCachedThread`.

Excluded:
- `tasks/todo.md`, the author's scratch checklist.

Follow-up (fork/changes, not this layer): our `reload-required` branch and
`reloadFromServer` are built on the deleted `setThread`, so rebasing
fork/changes onto this layer must re-express them against the reducer
(split the batch at the reload point, then re-enter `applyItems` with the
remainder).

Co-authored-by: T3 Code PR Stack <41898282+github-actions[bot]@users.noreply.github.com>

Author: patroza

.github/upstream-candidates.json

@@ -36,6 +36,12 @@
       "sourceSha": "a27510d060645809ae1472bba4dbb248dc624e25",
       "status": "active",
       "purpose": "Refresh open workspace file previews when disk contents change"
+    },
+    {
+      "upstreamPr": 5344,
+      "sourceSha": "783fd023c69f7ebc9c0ae657ce124cce77d94587",
+      "status": "active",
+      "purpose": "Batch live thread stream items so large-thread sync publishes client state once per batch"
     }
   ]
 }

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

@@ -9,6 +9,7 @@ import {
   TurnId,
   type OrchestrationThread,
   type OrchestrationThreadDetailSnapshot,
+  type OrchestrationSession,
   type OrchestrationThreadStreamItem,
 } from "@t3tools/contracts";
 import { describe, expect, it } from "@effect/vitest";
@@ -136,10 +137,12 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
   readonly cached?: OrchestrationThread;
   readonly httpSnapshot?: Option.Option<OrchestrationThreadDetailSnapshot>;
   readonly completionMarker?: boolean;
+  readonly eventBatchSize?: number;
 }) {
   const inputs = yield* Queue.unbounded<TestThreadInput>();
   const observed = yield* Queue.unbounded<EnvironmentThreadState>();
   const latest = yield* Ref.make<EnvironmentThreadState>(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);
@@ -224,7 +227,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),
@@ -235,7 +240,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,
   );
@@ -244,6 +252,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
     inputs,
     observed,
     latest,
+    statePublicationCount,
     retryCount,
     subscriptionCount,
     loaderCalls,
@@ -276,6 +285,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: {
@@ -350,6 +397,84 @@ 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("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 });