Refactor client runtime state refresh handling

- Move archived threads state into shared client-runtime
- Add stale-aware refresh/invalidation for path search, discovery, and VCS refs
- Wire provider invalidations through the web runtime

Author: juliusmarminge

apps/mobile/src/state/use-composer-path-search.ts

@@ -17,10 +17,13 @@ import {
   subscribeEnvironmentConnections,
 } from "./environment-session-registry";
 
+const COMPOSER_PATH_SEARCH_STALE_TIME_MS = 15_000;
+
 const composerPathSearchManager = createComposerPathSearchManager({
   getRegistry: () => appAtomRegistry,
   getClient: (environmentId) => getEnvironmentClient(environmentId)?.projects ?? null,
   subscribeClientChanges: subscribeEnvironmentConnections,
+  staleTimeMs: COMPOSER_PATH_SEARCH_STALE_TIME_MS,
 });
 
 export function useComposerPathSearch(target: ComposerPathSearchTarget): ComposerPathSearchState {

apps/web/src/environments/runtime/index.ts

@@ -27,4 +27,5 @@ export {
   resetEnvironmentServiceForTests,
   startEnvironmentConnectionService,
   subscribeEnvironmentConnections,
+  subscribeProviderInvalidations,
 } from "./service";

apps/web/src/environments/runtime/service.ts

@@ -119,6 +119,7 @@ const pendingSavedEnvironmentConnections = new Map<
   PendingSavedEnvironmentConnection
 >();
 const environmentConnectionListeners = new Set<() => void>();
+const providerInvalidationListeners = new Set<() => void>();
 const threadDetailSubscriptions = new Map<string, ThreadDetailSubscriptionEntry>();
 const terminalMetadataSubscriptions = new Map<EnvironmentId, () => void>();
 const lastAppliedProjectionVersionByEnvironment = new Map<
@@ -604,6 +605,12 @@ function emitEnvironmentConnectionRegistryChange() {
   }
 }
 
+function emitProviderInvalidation() {
+  for (const listener of providerInvalidationListeners) {
+    listener();
+  }
+}
+
 function getRuntimeErrorFields(error: unknown) {
   return {
     lastError: error instanceof Error ? error.message : String(error),
@@ -1543,6 +1550,13 @@ export function subscribeEnvironmentConnections(listener: () => void): () => voi
   };
 }
 
+export function subscribeProviderInvalidations(listener: () => void): () => void {
+  providerInvalidationListeners.add(listener);
+  return () => {
+    providerInvalidationListeners.delete(listener);
+  };
+}
+
 export function listEnvironmentConnections(): ReadonlyArray<EnvironmentConnection> {
   return [...environmentConnections.values()];
 }
@@ -1769,6 +1783,7 @@ export function startEnvironmentConnectionService(queryClient: QueryClient): ()
       }
       needsProviderInvalidation = false;
       void queryClient.invalidateQueries({ queryKey: providerQueryKeys.all });
+      emitProviderInvalidation();
     },
     {
       wait: 100,

apps/web/src/lib/archivedThreadsState.ts

@@ -1,83 +1,23 @@
 import { useAtomValue } from "@effect/atom-react";
-import { EnvironmentId, type OrchestrationShellSnapshot } from "@t3tools/contracts";
-import { AsyncResult, Atom } from "effect/unstable/reactivity";
+import {
+  type ArchivedSnapshotEntry,
+  createArchivedThreadsManager,
+  makeArchivedThreadsEnvironmentKey,
+  readArchivedThreadsSnapshotState,
+} from "@t3tools/client-runtime";
+import type { EnvironmentId } from "@t3tools/contracts";
 import { useCallback, useMemo } from "react";
-import * as Cause from "effect/Cause";
-import * as Effect from "effect/Effect";
-import * as Option from "effect/Option";
+
 import { readEnvironmentApi } from "../environmentApi";
 import { appAtomRegistry } from "../rpc/atomRegistry";
 
-const ARCHIVED_THREADS_STALE_TIME_MS = 5_000;
-const ARCHIVED_THREADS_IDLE_TTL_MS = 5 * 60_000;
-const ARCHIVED_THREADS_ENVIRONMENT_KEY_SEPARATOR = "\u001f";
-
-export type ArchivedSnapshotEntry = {
-  readonly environmentId: EnvironmentId;
-  readonly snapshot: OrchestrationShellSnapshot;
-};
-
-const knownArchivedThreadEnvironmentKeys = new Set<string>();
-
-function makeArchivedThreadsEnvironmentKey(environmentIds: ReadonlyArray<EnvironmentId>): string {
-  return environmentIds.toSorted().join(ARCHIVED_THREADS_ENVIRONMENT_KEY_SEPARATOR);
-}
-
-function parseArchivedThreadsEnvironmentKey(key: string): ReadonlyArray<EnvironmentId> {
-  if (key.length === 0) {
-    return [];
-  }
-  return key
-    .split(ARCHIVED_THREADS_ENVIRONMENT_KEY_SEPARATOR)
-    .map((environmentId) => EnvironmentId.make(environmentId));
-}
-
-const archivedThreadSnapshotsAtom = Atom.family((environmentKey: string) => {
-  knownArchivedThreadEnvironmentKeys.add(environmentKey);
-  return Atom.make(
-    Effect.promise(async (): Promise<ReadonlyArray<ArchivedSnapshotEntry>> => {
-      const environmentIds = parseArchivedThreadsEnvironmentKey(environmentKey);
-      const snapshots = await Promise.all(
-        environmentIds.map(async (environmentId) => {
-          const api = readEnvironmentApi(environmentId);
-          if (!api) {
-            return null;
-          }
-          return {
-            environmentId,
-            snapshot: await api.orchestration.getArchivedShellSnapshot(),
-          };
-        }),
-      );
-      return snapshots.filter((snapshot) => snapshot !== null);
-    }),
-  ).pipe(
-    Atom.swr({
-      staleTime: ARCHIVED_THREADS_STALE_TIME_MS,
-      revalidateOnMount: true,
-    }),
-    Atom.setIdleTTL(ARCHIVED_THREADS_IDLE_TTL_MS),
-    Atom.withLabel(`archived-thread-snapshots:${environmentKey}`),
-  );
+const archivedThreadsManager = createArchivedThreadsManager({
+  getRegistry: () => appAtomRegistry,
+  getClient: (environmentId) => readEnvironmentApi(environmentId)?.orchestration ?? null,
 });
 
-function readArchivedThreadsError(
-  result: AsyncResult.AsyncResult<ReadonlyArray<ArchivedSnapshotEntry>, unknown>,
-): string | null {
-  if (result._tag !== "Failure") {
-    return null;
-  }
-
-  const error = Cause.squash(result.cause);
-  return error instanceof Error ? error.message : "Failed to load archived threads.";
-}
-
 export function refreshArchivedThreadsForEnvironment(environmentId: EnvironmentId): void {
-  for (const key of knownArchivedThreadEnvironmentKeys) {
-    if (parseArchivedThreadsEnvironmentKey(key).includes(environmentId)) {
-      appAtomRegistry.refresh(archivedThreadSnapshotsAtom(key));
-    }
-  }
+  archivedThreadsManager.refreshForEnvironment(environmentId);
 }
 
 export function useArchivedThreadSnapshots(environmentIds: ReadonlyArray<EnvironmentId>): {
@@ -90,17 +30,14 @@ export function useArchivedThreadSnapshots(environmentIds: ReadonlyArray<Environ
     () => makeArchivedThreadsEnvironmentKey(environmentIds),
     [environmentIds],
   );
-  const atom = archivedThreadSnapshotsAtom(environmentKey);
+  const atom = archivedThreadsManager.getAtom(environmentKey);
   const result = useAtomValue(atom);
-  const snapshots = Option.getOrElse(AsyncResult.value(result), () => []);
   const refresh = useCallback(() => {
-    appAtomRegistry.refresh(atom);
-  }, [atom]);
+    archivedThreadsManager.refresh(environmentIds);
+  }, [environmentIds]);
 
   return {
-    snapshots,
-    error: readArchivedThreadsError(result),
-    isLoading: result.waiting,
+    ...readArchivedThreadsSnapshotState(result),
     refresh,
   };
 }

apps/web/src/lib/composerPathSearchState.ts

@@ -14,20 +14,29 @@ import { useEffect, useMemo } from "react";
 import {
   readEnvironmentConnection,
   subscribeEnvironmentConnections,
+  subscribeProviderInvalidations,
 } from "../environments/runtime";
 import { appAtomRegistry } from "../rpc/atomRegistry";
 
 const COMPOSER_PATH_SEARCH_LIMIT = 80;
 const COMPOSER_PATH_SEARCH_DEBOUNCE_MS = 120;
+const COMPOSER_PATH_SEARCH_STALE_TIME_MS = 15_000;
 
 const composerPathSearchManager = createComposerPathSearchManager({
   getRegistry: () => appAtomRegistry,
   getClient: (environmentId) => readEnvironmentConnection(environmentId)?.client.projects ?? null,
   subscribeClientChanges: subscribeEnvironmentConnections,
   limit: COMPOSER_PATH_SEARCH_LIMIT,
   debounceMs: COMPOSER_PATH_SEARCH_DEBOUNCE_MS,
+  staleTimeMs: COMPOSER_PATH_SEARCH_STALE_TIME_MS,
 });
 
+export function invalidateComposerPathSearches(): void {
+  composerPathSearchManager.invalidate();
+}
+
+subscribeProviderInvalidations(invalidateComposerPathSearches);
+
 export function useComposerPathSearch(target: ComposerPathSearchTarget): ComposerPathSearchState {
   const stableTarget = useMemo(
     () => ({

apps/web/src/lib/sourceControlDiscoveryState.ts

@@ -7,11 +7,13 @@ import {
   sourceControlDiscoveryStateAtom,
 } from "@t3tools/client-runtime";
 import { EnvironmentId, type SourceControlDiscoveryResult } from "@t3tools/contracts";
-import * as Effect from "effect/Effect";
-import { Atom } from "effect/unstable/reactivity";
+import { useEffect } from "react";
 
 import { readPrimaryEnvironmentDescriptor } from "../environments/primary";
-import { readEnvironmentConnection } from "../environments/runtime";
+import {
+  readEnvironmentConnection,
+  subscribeEnvironmentConnections,
+} from "../environments/runtime";
 import { readLocalApi } from "../localApi";
 import { appAtomRegistry } from "../rpc/atomRegistry";
 
@@ -59,21 +61,11 @@ const sourceControlDiscoveryManager = createSourceControlDiscoveryManager({
     }
     return null;
   },
+  subscribeClientChanges: subscribeEnvironmentConnections,
+  staleTimeMs: SOURCE_CONTROL_DISCOVERY_STALE_TIME_MS,
+  idleTtlMs: SOURCE_CONTROL_DISCOVERY_IDLE_TTL_MS,
 });
 
-const sourceControlDiscoveryAutoRefreshAtom = Atom.family((targetKey: string) =>
-  Atom.make(() =>
-    Effect.promise(() => sourceControlDiscoveryManager.refresh({ key: targetKey })),
-  ).pipe(
-    Atom.swr({
-      staleTime: SOURCE_CONTROL_DISCOVERY_STALE_TIME_MS,
-      revalidateOnMount: true,
-    }),
-    Atom.setIdleTTL(SOURCE_CONTROL_DISCOVERY_IDLE_TTL_MS),
-    Atom.withLabel(`source-control-discovery:auto-refresh:${targetKey}`),
-  ),
-);
-
 export function refreshSourceControlDiscovery(
   input?: SourceControlDiscoveryTargetInput,
 ): Promise<SourceControlDiscoveryResult | null> {
@@ -97,7 +89,7 @@ export function useSourceControlDiscovery(
     getSourceControlDiscoveryTargetKey(sourceControlDiscoveryTarget(input)) ??
     SOURCE_CONTROL_DISCOVERY_TARGET.key;
 
-  useAtomValue(sourceControlDiscoveryAutoRefreshAtom(targetKey));
+  useEffect(() => sourceControlDiscoveryManager.watch({ key: targetKey }), [targetKey]);
 
   return useAtomValue(sourceControlDiscoveryStateAtom(targetKey));
 }

packages/client-runtime/src/archivedThreadsState.test.ts

@@ -0,0 +1,96 @@
+import { EnvironmentId, type OrchestrationShellSnapshot } from "@t3tools/contracts";
+import { AtomRegistry } from "effect/unstable/reactivity";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import {
+  type ArchivedThreadsClient,
+  createArchivedThreadsManager,
+  makeArchivedThreadsEnvironmentKey,
+  parseArchivedThreadsEnvironmentKey,
+  readArchivedThreadsSnapshotState,
+} from "./archivedThreadsState.ts";
+
+let registry = AtomRegistry.make();
+
+function resetAtomRegistry() {
+  registry.dispose();
+  registry = AtomRegistry.make();
+}
+
+function createSnapshot(id: string): OrchestrationShellSnapshot {
+  return {
+    snapshotSequence: 1,
+    projects: [],
+    threads: [],
+    updatedAt: `2026-05-08T00:00:00.000Z`,
+    id,
+  } as OrchestrationShellSnapshot;
+}
+
+describe("createArchivedThreadsManager", () => {
+  afterEach(() => {
+    resetAtomRegistry();
+  });
+
+  it("loads archived snapshots for configured environment clients", async () => {
+    const envA = EnvironmentId.make("env-a");
+    const envB = EnvironmentId.make("env-b");
+    const clients = new Map<EnvironmentId, ArchivedThreadsClient>([
+      [
+        envA,
+        {
+          getArchivedShellSnapshot: vi.fn(async () => createSnapshot("a")),
+        },
+      ],
+      [
+        envB,
+        {
+          getArchivedShellSnapshot: vi.fn(async () => createSnapshot("b")),
+        },
+      ],
+    ]);
+    const manager = createArchivedThreadsManager({
+      getRegistry: () => registry,
+      getClient: (environmentId) => clients.get(environmentId) ?? null,
+    });
+
+    const result = registry.get(manager.getAtom(makeArchivedThreadsEnvironmentKey([envB, envA])));
+
+    await vi.waitFor(() => {
+      const state = readArchivedThreadsSnapshotState(
+        registry.get(manager.getAtom(makeArchivedThreadsEnvironmentKey([envA, envB]))),
+      );
+      expect(state.snapshots.map((snapshot) => snapshot.environmentId)).toEqual([envA, envB]);
+    });
+    expect(readArchivedThreadsSnapshotState(result).isLoading).toBe(true);
+  });
+
+  it("refreshes known snapshot groups that include an environment", async () => {
+    const envA = EnvironmentId.make("env-a");
+    const envB = EnvironmentId.make("env-b");
+    const getArchivedShellSnapshot = vi.fn(async () =>
+      createSnapshot(`a-${getArchivedShellSnapshot.mock.calls.length}`),
+    );
+    const manager = createArchivedThreadsManager({
+      getRegistry: () => registry,
+      getClient: (environmentId) => (environmentId === envA ? { getArchivedShellSnapshot } : null),
+      staleTimeMs: 60_000,
+    });
+
+    const atom = manager.getAtom(makeArchivedThreadsEnvironmentKey([envA, envB]));
+    registry.get(atom);
+    await vi.waitFor(() => expect(getArchivedShellSnapshot).toHaveBeenCalledTimes(1));
+
+    manager.refreshForEnvironment(envA);
+
+    await vi.waitFor(() => expect(getArchivedShellSnapshot).toHaveBeenCalledTimes(2));
+  });
+
+  it("round-trips environment keys in sorted order", () => {
+    const envA = EnvironmentId.make("env-a");
+    const envB = EnvironmentId.make("env-b");
+    const key = makeArchivedThreadsEnvironmentKey([envB, envA]);
+
+    expect(parseArchivedThreadsEnvironmentKey(key)).toEqual([envA, envB]);
+  });
+});