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