From 7f6f1dbe50b54782d209b7c6dbc407f3107839a1 Mon Sep 17 00:00:00 2001
From: Chrrxs <lcchrrxs@gmail.com>
Date: Wed, 15 Jul 2026 09:07:39 -0400
Subject: [PATCH] perf(client): defer active thread cache writes

Active thread updates previously serialized the complete retained history after each 500 ms quiet period. Defer persistence until the session settles so large threads do not repeatedly allocate and write giant snapshots.

Fixes #4005
---
 .../src/state/threads-sync.test.ts            | 46 +++++++++++++++++++
 packages/client-runtime/src/state/threads.ts  | 19 ++++++--
 2 files changed, 60 insertions(+), 5 deletions(-)

diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts
index 8e982723d22..630a5f27d16 100644
--- a/packages/client-runtime/src/state/threads-sync.test.ts
+++ b/packages/client-runtime/src/state/threads-sync.test.ts
@@ -5,6 +5,7 @@ import {
   ProjectId,
   ProviderInstanceId,
   ThreadId,
+  TurnId,
   type OrchestrationThread,
   type OrchestrationThreadDetailSnapshot,
   type OrchestrationThreadStreamItem,
@@ -74,6 +75,26 @@ const BASE_THREAD: OrchestrationThread = {
   checkpoints: [],
   session: null,
 };
+const ACTIVE_THREAD: OrchestrationThread = {
+  ...BASE_THREAD,
+  latestTurn: {
+    turnId: TurnId.make("turn-1"),
+    state: "running",
+    requestedAt: "2026-04-01T00:01:00.000Z",
+    startedAt: "2026-04-01T00:01:00.000Z",
+    completedAt: null,
+    assistantMessageId: null,
+  },
+  session: {
+    threadId: THREAD_ID,
+    status: "running",
+    providerName: "codex",
+    runtimeMode: "full-access",
+    activeTurnId: TurnId.make("turn-1"),
+    lastError: null,
+    updatedAt: "2026-04-01T00:01:00.000Z",
+  },
+};
 
 type TestThreadInput = OrchestrationThreadStreamItem | Error;
 
@@ -308,6 +329,31 @@ describe("EnvironmentThreads", () => {
     }),
   );
 
+  it.effect("does not persist active thread snapshots during streaming or teardown", () =>
+    Effect.gen(function* () {
+      const savedThreads = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const harness = yield* makeHarness({ cached: ACTIVE_THREAD });
+          yield* awaitThreadState(
+            harness.observed,
+            (value) =>
+              value.status === "live" &&
+              Option.isSome(value.data) &&
+              value.data.value.session?.status === "running",
+          );
+
+          yield* TestClock.adjust("500 millis");
+          yield* Effect.yieldNow;
+
+          expect(yield* Ref.get(harness.savedThreads)).toEqual([]);
+          return harness.savedThreads;
+        }),
+      );
+
+      expect(yield* Ref.get(savedThreads)).toEqual([]);
+    }),
+  );
+
   it.effect("seeds the thread from the HTTP snapshot and resumes live events", () =>
     Effect.gen(function* () {
       const httpThread: OrchestrationThread = { ...BASE_THREAD, title: "HTTP title" };
diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts
index fd5b425fa2a..0c9ec340610 100644
--- a/packages/client-runtime/src/state/threads.ts
+++ b/packages/client-runtime/src/state/threads.ts
@@ -41,6 +41,11 @@ function formatThreadError(cause: Cause.Cause<unknown>): string {
     : "Could not synchronize the thread.";
 }
 
+function shouldPersistThread(thread: OrchestrationThread): boolean {
+  const status = thread.session?.status;
+  return status !== "starting" && status !== "running";
+}
+
 export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* (
   threadId: ThreadIdType,
 ) {
@@ -128,10 +133,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
       status: "live",
       error: Option.none(),
     });
-    // Persist the thread together with the sequence it reflects so the next warm
-    // cache can resume from exactly here.
-    const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
-    yield* Queue.offer(persistence, { snapshotSequence, thread });
+    // 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* () {
@@ -246,7 +254,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
       Effect.flatMap(([current, snapshotSequence]) =>
         Option.match(current.data, {
           onNone: () => Effect.void,
-          onSome: (thread) => persist({ snapshotSequence, thread }),
+          onSome: (thread) =>
+            shouldPersistThread(thread) ? persist({ snapshotSequence, thread }) : Effect.void,
         }),
       ),
     ),