authorgravatar for 67719167+Chrrxs@users.noreply.github.comChris Michael Guzman <67719167+Chrrxs@users.noreply.github.com> 2026-07-18 07:16:38-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-18 13:16:38+02:00
log765e1b5fc044b899741a11dcba141bb2c49c6c37
tree372b1f3cf9c72702e515b089f410eed1943bf86d
parent03e20c875ccd39e4cedba8f6b713195290ec4652
signature Signed by PGP key B5690EEEBB952194

perf(client): defer active thread cache writes (#4006)

2 files changed, 60 insertions(+), 5 deletions(-)

packages/client-runtime/src/state/threads-sync.test.ts+46
...@@ -5,6 +5,7 @@ import {...@@ -5,6 +5,7 @@ import {
5 ProjectId,5 ProjectId,
6 ProviderInstanceId,6 ProviderInstanceId,
7 ThreadId,7 ThreadId,
8 TurnId,
8 type OrchestrationThread,9 type OrchestrationThread,
9 type OrchestrationThreadDetailSnapshot,10 type OrchestrationThreadDetailSnapshot,
10 type OrchestrationThreadStreamItem,11 type OrchestrationThreadStreamItem,
...@@ -74,6 +75,26 @@ const BASE_THREAD: OrchestrationThread = {...@@ -74,6 +75,26 @@ const BASE_THREAD: OrchestrationThread = {
74 checkpoints: [],75 checkpoints: [],
75 session: null,76 session: null,
76};77};
78const ACTIVE_THREAD: OrchestrationThread = {
79 ...BASE_THREAD,
80 latestTurn: {
81 turnId: TurnId.make("turn-1"),
82 state: "running",
83 requestedAt: "2026-04-01T00:01:00.000Z",
84 startedAt: "2026-04-01T00:01:00.000Z",
85 completedAt: null,
86 assistantMessageId: null,
87 },
88 session: {
89 threadId: THREAD_ID,
90 status: "running",
91 providerName: "codex",
92 runtimeMode: "full-access",
93 activeTurnId: TurnId.make("turn-1"),
94 lastError: null,
95 updatedAt: "2026-04-01T00:01:00.000Z",
96 },
97};
7798
78type TestThreadInput = OrchestrationThreadStreamItem | Error;99type TestThreadInput = OrchestrationThreadStreamItem | Error;
79100
...@@ -308,6 +329,31 @@ describe("EnvironmentThreads", () => {...@@ -308,6 +329,31 @@ describe("EnvironmentThreads", () => {
308 }),329 }),
309 );330 );
310331
332 it.effect("does not persist active thread snapshots during streaming or teardown", () =>
333 Effect.gen(function* () {
334 const savedThreads = yield* Effect.scoped(
335 Effect.gen(function* () {
336 const harness = yield* makeHarness({ cached: ACTIVE_THREAD });
337 yield* awaitThreadState(
338 harness.observed,
339 (value) =>
340 value.status === "live" &&
341 Option.isSome(value.data) &&
342 value.data.value.session?.status === "running",
343 );
344
345 yield* TestClock.adjust("500 millis");
346 yield* Effect.yieldNow;
347
348 expect(yield* Ref.get(harness.savedThreads)).toEqual([]);
349 return harness.savedThreads;
350 }),
351 );
352
353 expect(yield* Ref.get(savedThreads)).toEqual([]);
354 }),
355 );
356
311 it.effect("seeds the thread from the HTTP snapshot and resumes live events", () =>357 it.effect("seeds the thread from the HTTP snapshot and resumes live events", () =>
312 Effect.gen(function* () {358 Effect.gen(function* () {
313 const httpThread: OrchestrationThread = { ...BASE_THREAD, title: "HTTP title" };359 const httpThread: OrchestrationThread = { ...BASE_THREAD, title: "HTTP title" };
packages/client-runtime/src/state/threads.ts+14-5
...@@ -41,6 +41,11 @@ function formatThreadError(cause: Cause.Cause<unknown>): string {...@@ -41,6 +41,11 @@ function formatThreadError(cause: Cause.Cause<unknown>): string {
41 : "Could not synchronize the thread.";41 : "Could not synchronize the thread.";
42}42}
4343
44function shouldPersistThread(thread: OrchestrationThread): boolean {
45 const status = thread.session?.status;
46 return status !== "starting" && status !== "running";
47}
48
44export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* (49export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* (
45 threadId: ThreadIdType,50 threadId: ThreadIdType,
46) {51) {
...@@ -128,10 +133,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -128,10 +133,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
128 status: "live",133 status: "live",
129 error: Option.none(),134 error: Option.none(),
130 });135 });
131 // Persist the thread together with the sequence it reflects so the next warm136 // Active threads can update many times per second and retain large tool
132 // cache can resume from exactly here.137 // payloads. The server remains the source of truth while a turn is active;
133 const snapshotSequence = yield* SubscriptionRef.get(lastSequence);138 // persist once it settles so cache encoding stays off the streaming path.
134 yield* Queue.offer(persistence, { snapshotSequence, thread });139 if (shouldPersistThread(thread)) {
140 const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
141 yield* Queue.offer(persistence, { snapshotSequence, thread });
142 }
135 });143 });
136144
137 const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () {145 const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () {
...@@ -246,7 +254,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -246,7 +254,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
246 Effect.flatMap(([current, snapshotSequence]) =>254 Effect.flatMap(([current, snapshotSequence]) =>
247 Option.match(current.data, {255 Option.match(current.data, {
248 onNone: () => Effect.void,256 onNone: () => Effect.void,
249 onSome: (thread) => persist({ snapshotSequence, thread }),257 onSome: (thread) =>
258 shouldPersistThread(thread) ? persist({ snapshotSequence, thread }) : Effect.void,
250 }),259 }),
251 ),260 ),
252 ),261 ),