Share thread state idle TTL across client atoms (#3163)

Author: juliusmarminge

packages/client-runtime/src/state/threadDetail.ts

@@ -15,12 +15,12 @@ import type { EnvironmentThread, EnvironmentThreadShell } from "./models.ts";
 import { scopeThread } from "./models.ts";
 import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threads.ts";
 import { parseThreadKey, threadKey } from "./entities.ts";
+import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
 
 const EMPTY_MESSAGES: ReadonlyArray<OrchestrationMessage> = Object.freeze([]);
 const EMPTY_ACTIVITIES: ReadonlyArray<OrchestrationThreadActivity> = Object.freeze([]);
 const EMPTY_PROPOSED_PLANS: ReadonlyArray<OrchestrationProposedPlan> = Object.freeze([]);
 const EMPTY_CHECKPOINTS: ReadonlyArray<OrchestrationCheckpointSummary> = Object.freeze([]);
-const THREAD_DETAIL_IDLE_TTL_MS = 5 * 60_000;
 
 /**
  * Combine detail-only collections with the shell's authoritative thread metadata.
@@ -75,7 +75,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
         () => EMPTY_ENVIRONMENT_THREAD_STATE,
       ),
     ).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-state-value:${key}`),
     );
   });
@@ -93,21 +93,21 @@ export function createEnvironmentThreadDetailAtoms<E>(
       previousValue = source === null ? null : scopeThread(ref.environmentId, source);
       return previousValue;
     }).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-detail:${key}`),
     );
   });
 
   const threadStatusAtomFamily = Atom.family((key: string) =>
     Atom.make((get) => get(threadStateValueAtomFamily(key)).status).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-status:${key}`),
     ),
   );
 
   const threadErrorAtomFamily = Atom.family((key: string) =>
     Atom.make((get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).error)).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-error:${key}`),
     ),
   );
@@ -117,7 +117,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
       (get): ReadonlyArray<OrchestrationMessage> =>
         get(threadDetailAtomFamily(key))?.messages ?? EMPTY_MESSAGES,
     ).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-messages:${key}`),
     ),
   );
@@ -127,7 +127,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
       (get): ReadonlyArray<OrchestrationThreadActivity> =>
         get(threadDetailAtomFamily(key))?.activities ?? EMPTY_ACTIVITIES,
     ).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-activities:${key}`),
     ),
   );
@@ -137,7 +137,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
       (get): ReadonlyArray<OrchestrationProposedPlan> =>
         get(threadDetailAtomFamily(key))?.proposedPlans ?? EMPTY_PROPOSED_PLANS,
     ).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-proposed-plans:${key}`),
     ),
   );
@@ -147,7 +147,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
       (get): ReadonlyArray<OrchestrationCheckpointSummary> =>
         get(threadDetailAtomFamily(key))?.checkpoints ?? EMPTY_CHECKPOINTS,
     ).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-checkpoints:${key}`),
     ),
   );
@@ -156,7 +156,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
     Atom.make(
       (get): OrchestrationSession | null => get(threadDetailAtomFamily(key))?.session ?? null,
     ).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-session:${key}`),
     ),
   );
@@ -165,7 +165,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
     Atom.make(
       (get): OrchestrationLatestTurn | null => get(threadDetailAtomFamily(key))?.latestTurn ?? null,
     ).pipe(
-      Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),
+      Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
       Atom.withLabel(`environment-thread-latest-turn:${key}`),
     ),
   );

packages/client-runtime/src/state/threadRetention.ts

@@ -0,0 +1,3 @@
+// Mobile thread routes unmount during back navigation. Retain the stream-backed
+// state across short subscriber gaps without keeping every opened thread alive.
+export const THREAD_STATE_IDLE_TTL_MS = 5 * 60_000;

packages/client-runtime/src/state/threads-atoms.test.ts

@@ -0,0 +1,26 @@
+import { EnvironmentId, ThreadId } from "@t3tools/contracts";
+import { describe, expect, it } from "@effect/vitest";
+import * as Layer from "effect/Layer";
+import { Atom } from "effect/unstable/reactivity";
+
+import type { EnvironmentRegistry } from "../connection/registry.ts";
+import type { EnvironmentCacheStore } from "../platform/persistence.ts";
+import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
+import { createEnvironmentThreadStateAtoms } from "./threads.ts";
+
+describe("createEnvironmentThreadStateAtoms", () => {
+  it("retains thread state across short subscriber gaps", () => {
+    const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<
+      EnvironmentRegistry | EnvironmentCacheStore,
+      never
+    >;
+    const threads = createEnvironmentThreadStateAtoms(runtime);
+    const environmentId = EnvironmentId.make("environment-1");
+    const threadId = ThreadId.make("thread-1");
+    const atom = threads.stateAtom(environmentId, threadId);
+
+    expect(atom.idleTTL).toBe(THREAD_STATE_IDLE_TTL_MS);
+    expect(threads.stateAtom(environmentId, threadId)).toBe(atom);
+    expect(threads.stateAtom(environmentId, ThreadId.make("thread-2"))).not.toBe(atom);
+  });
+});

packages/client-runtime/src/state/threads.ts

@@ -21,6 +21,7 @@ import { EnvironmentSupervisor } from "../connection/supervisor.ts";
 import { EnvironmentCacheStore } from "../platform/persistence.ts";
 import { subscribe } from "../rpc/client.ts";
 import { applyThreadDetailEvent } from "./threadReducer.ts";
+import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
 import { followStreamInEnvironment } from "./runtime.ts";
 
 export type EnvironmentThreadStatus = "empty" | "cached" | "synchronizing" | "live" | "deleted";
@@ -249,9 +250,14 @@ export function createEnvironmentThreadStateAtoms<R, E>(
 ) {
   const family = Atom.family((key: string) => {
     const { environmentId, threadId } = parseThreadAtomKey(key);
-    return runtime.atom(threadStateChanges(environmentId, threadId), {
-      initialValue: EMPTY_ENVIRONMENT_THREAD_STATE,
-    });
+    return runtime
+      .atom(threadStateChanges(environmentId, threadId), {
+        initialValue: EMPTY_ENVIRONMENT_THREAD_STATE,
+      })
+      .pipe(
+        Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
+        Atom.withLabel(`environment-thread-state:${key}`),
+      );
   });
 
   return {