@@ -133,10 +133,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);
@@ -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 });
 
@@ -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<OrchestrationThreadStreamItem>,
+): 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<OrchestrationThread>): 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<OrchestrationThreadStreamItem>,
   ) {
-    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(() =>
 
@@ -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.