perf(client): defer active thread cache writes

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" };

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,
         }),
       ),
     ),