import {  type AuthSessionRole,  type DesktopSshEnvironmentBootstrap,  type DesktopSshEnvironmentTarget,  type EnvironmentId,  type OrchestrationEvent,  type OrchestrationShellSnapshot,  type OrchestrationShellStreamEvent,  type ServerConfig,  ThreadId,} from "@t3tools/contracts";import {  createWsRpcClient as createBaseWsRpcClient,  type WsProtocolCloseContext,  type WsRpcClient,} from "@t3tools/client-runtime"; import { type QueryClient } from "@tanstack/react-query";import { Throttler } from "@tanstack/react-pacer";import {  createKnownEnvironment,  getKnownEnvironmentWsBaseUrl,  scopedThreadKey,  scopeProjectRef,  scopeThreadRef,} from "@t3tools/client-runtime"; import {  markPromotedDraftThreadByRef,  markPromotedDraftThreadsByRef,  useComposerDraftStore,} from "~/composerDraftStore";import { ensureLocalApi } from "~/localApi";import { collectActiveTerminalUiThreadKeys } from "~/lib/terminalUiStateCleanup";import { deriveOrchestrationBatchEffects } from "~/orchestrationEventEffects";import { getPrimaryKnownEnvironment } from "../primary";import {  bootstrapRemoteBearerSession,  fetchRemoteEnvironmentDescriptor,  fetchRemoteSessionState,  isRemoteEnvironmentAuthHttpError,  resolveRemoteWebSocketConnectionUrl,} from "../remote/api";import { resolveRemotePairingTarget } from "../remote/target";import {  getSavedEnvironmentRecord,  hasSavedEnvironmentRegistryHydrated,  listSavedEnvironmentRecords,  persistSavedEnvironmentRecord,  readSavedEnvironmentBearerToken,  removeSavedEnvironmentBearerToken,  type SavedEnvironmentRecord,  toPersistedSavedEnvironmentRecord,  useSavedEnvironmentRegistryStore,  useSavedEnvironmentRuntimeStore,  waitForSavedEnvironmentRegistryHydration,  writeSavedEnvironmentBearerToken,} from "./catalog";import { createEnvironmentConnection, type EnvironmentConnection } from "./connection";import {  useStore,  selectProjectsAcrossEnvironments,  selectSidebarThreadSummaryByRef,  selectThreadByRef,  selectThreadsAcrossEnvironments,} from "~/store";import { useTerminalUiStateStore } from "~/terminalUiStateStore";import { useUiStateStore } from "~/uiStateStore";import { WsTransport } from "~/rpc/wsTransport";import { getServerConfig } from "../../rpc/serverState";import { appendVersionMismatchHint, resolveServerConfigVersionMismatch } from "../../versionSkew";import {  deriveLogicalProjectKeyFromSettings,  derivePhysicalProjectKey,} from "../../logicalProject";import { subscribeTerminalMetadata, terminalSessionManager } from "../../terminalSessionState";import { resetWsReconnectBackoff } from "~/rpc/wsConnectionState";import { getClientSettings } from "~/hooks/useSettings"; type EnvironmentServiceState = {  readonly queryClient: QueryClient;  readonly queryInvalidationThrottler: Throttler<() => void>;  refCount: number;  stop: () => void;}; type ThreadDetailSubscriptionEntry = {  readonly environmentId: EnvironmentId;  readonly threadId: ThreadId;  unsubscribe: () => void;  unsubscribeConnectionListener: (() => void) | null;  refCount: number;  lastAccessedAt: number;  evictionTimeoutId: ReturnType<typeof setTimeout> | null;}; const environmentConnections = new Map<EnvironmentId, EnvironmentConnection>();class SavedEnvironmentConnectionCancelledError extends Error {  constructor(environmentId: EnvironmentId) {    super(`Saved environment ${environmentId} connection was cancelled.`);    this.name = "SavedEnvironmentConnectionCancelledError";  }} function isSavedEnvironmentConnectionCancelledError(  error: unknown,): error is SavedEnvironmentConnectionCancelledError {  return error instanceof SavedEnvironmentConnectionCancelledError;} interface PendingSavedEnvironmentConnection {  cancelled: boolean;  readonly promise: Promise<EnvironmentConnection>;} const pendingSavedEnvironmentConnections = new Map<  EnvironmentId,  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<  EnvironmentId,  {    readonly sequence: number;    readonly updatedAt: string | null;  }>(); let activeService: EnvironmentServiceState | null = null;let needsProviderInvalidation = false;let lastBrowserHiddenAt: number | null = null;let lastBrowserResumeReconnectAt = Number.NEGATIVE_INFINITY; // TODO(CLIENT-RUNTIME MIGRATION - DO NOT EXPAND THIS WEB-ONLY COPY):// This file still owns web's legacy thread-detail subscription cache. Mobile// uses createThreadDetailManager from @t3tools/client-runtime for the same// retain/reconnect/evict lifecycle. When touching this logic, prefer migrating// web to the shared manager or extracting the missing adapter layer instead of// adding more behavior here.//// Thread detail subscription cache policy:// - Active consumers keep a subscription retained via refCount.// - Released subscriptions stay warm for a longer idle TTL to avoid churn//   while moving around the UI.// - Threads with active work or pending user action are sticky and are never//   evicted while they remain non-idle.// - Capacity eviction only targets idle cached subscriptions.const THREAD_DETAIL_SUBSCRIPTION_IDLE_EVICTION_MS = 15 * 60 * 1000;const MAX_CACHED_THREAD_DETAIL_SUBSCRIPTIONS = 32;const BROWSER_RESUME_RECONNECT_COOLDOWN_MS = 2_000;const INITIAL_SERVER_CONFIG_SNAPSHOT_WAIT_MS = 150;const NOOP = () => undefined;const SSH_HTTP_STATUS_RE = /^\[ssh_http:(\d+)\]\s/u; function createDeferredPromise<T>() {  let resolve: ((value: T) => void) | null = null;  const promise = new Promise<T>((nextResolve) => {    resolve = nextResolve;  });   return {    promise,    resolve: (value: T) => {      resolve?.(value);      resolve = null;    },  };} async function waitForConfigSnapshot(  promise: Promise<ServerConfig>,  timeoutMs: number,): Promise<ServerConfig | null> {  return await new Promise<ServerConfig | null>((resolve) => {    const timeoutId = globalThis.setTimeout(() => resolve(null), timeoutMs);    promise.then(      (config) => {        clearTimeout(timeoutId);        resolve(config);      },      () => {        clearTimeout(timeoutId);        resolve(null);      },    );  });} function createSavedEnvironmentSyncScheduler() {  let activeSync: Promise<void> | null = null;  let queued = false;   const run = async (): Promise<void> => {    do {      queued = false;      await syncSavedEnvironmentConnections(listSavedEnvironmentRecords());    } while (queued);  };   return () => {    if (activeSync) {      queued = true;      return activeSync;    }     activeSync = run()      .catch(() => undefined)      .finally(() => {        activeSync = null;      });     return activeSync;  };}function compareAppliedProjectionVersion(  left: { readonly sequence: number; readonly updatedAt: string | null },  right: { readonly sequence: number; readonly updatedAt: string | null },): number {  if (left.sequence !== right.sequence) {    return left.sequence - right.sequence;  }   const leftUpdatedAt = left.updatedAt ?? "";  const rightUpdatedAt = right.updatedAt ?? "";  if (leftUpdatedAt === rightUpdatedAt) {    return 0;  }   return leftUpdatedAt < rightUpdatedAt ? -1 : 1;} function toAppliedProjectionVersion(  snapshot: Pick<OrchestrationShellSnapshot, "snapshotSequence" | "updatedAt">,): {  readonly sequence: number;  readonly updatedAt: string;} {  return {    sequence: snapshot.snapshotSequence,    updatedAt: snapshot.updatedAt,  };} export function shouldApplyProjectionSnapshot(input: {  readonly current: {    readonly sequence: number;    readonly updatedAt: string | null;  } | null;  readonly next: Pick<OrchestrationShellSnapshot, "snapshotSequence" | "updatedAt">;}): boolean {  if (input.current === null) {    return true;  }   return compareAppliedProjectionVersion(input.current, toAppliedProjectionVersion(input.next)) < 0;} export function shouldApplyProjectionEvent(input: {  readonly current: {    readonly sequence: number;    readonly updatedAt: string | null;  } | null;  readonly sequence: number;}): boolean {  if (input.current === null) {    return true;  }   return input.sequence > input.current.sequence;} function readLastAppliedProjectionVersion(environmentId: EnvironmentId): {  readonly sequence: number;  readonly updatedAt: string | null;} | null {  return lastAppliedProjectionVersionByEnvironment.get(environmentId) ?? null;} function markAppliedProjectionSnapshot(  environmentId: EnvironmentId,  snapshot: Pick<OrchestrationShellSnapshot, "snapshotSequence" | "updatedAt">,): void {  const nextVersion = toAppliedProjectionVersion(snapshot);  const currentVersion = readLastAppliedProjectionVersion(environmentId);  if (    currentVersion !== null &&    compareAppliedProjectionVersion(currentVersion, nextVersion) >= 0  ) {    return;  }   lastAppliedProjectionVersionByEnvironment.set(environmentId, nextVersion);} function markAppliedProjectionEvent(environmentId: EnvironmentId, sequence: number): void {  const currentVersion = readLastAppliedProjectionVersion(environmentId);  if (currentVersion !== null && sequence <= currentVersion.sequence) {    return;  }   lastAppliedProjectionVersionByEnvironment.set(environmentId, {    sequence,    updatedAt: currentVersion?.updatedAt ?? null,  });}function getThreadDetailSubscriptionKey(environmentId: EnvironmentId, threadId: ThreadId): string {  return scopedThreadKey(scopeThreadRef(environmentId, threadId));} function clearThreadDetailSubscriptionEviction(  entry: ThreadDetailSubscriptionEntry,): ThreadDetailSubscriptionEntry {  if (entry.evictionTimeoutId !== null) {    clearTimeout(entry.evictionTimeoutId);    entry.evictionTimeoutId = null;  }  return entry;} function isNonIdleThreadDetailSubscription(entry: ThreadDetailSubscriptionEntry): boolean {  const threadRef = scopeThreadRef(entry.environmentId, entry.threadId);  const state = useStore.getState();  const sidebarThread = selectSidebarThreadSummaryByRef(state, threadRef);   // Prefer shell/sidebar state first because it carries the coarse thread  // readiness flags used throughout the UI (pending approvals/input/plan).  if (sidebarThread) {    if (      sidebarThread.hasPendingApprovals ||      sidebarThread.hasPendingUserInput ||      sidebarThread.hasActionableProposedPlan    ) {      return true;    }     const orchestrationStatus = sidebarThread.session?.orchestrationStatus;    if (      orchestrationStatus &&      orchestrationStatus !== "idle" &&      orchestrationStatus !== "stopped"    ) {      return true;    }     if (sidebarThread.latestTurn?.state === "running") {      return true;    }  }   const thread = selectThreadByRef(state, threadRef);  if (!thread) {    return false;  }   const orchestrationStatus = thread.session?.orchestrationStatus;  return (    Boolean(      orchestrationStatus && orchestrationStatus !== "idle" && orchestrationStatus !== "stopped",    ) ||    thread.latestTurn?.state === "running" ||    thread.pendingSourceProposedPlan !== undefined  );} function shouldEvictThreadDetailSubscription(entry: ThreadDetailSubscriptionEntry): boolean {  return entry.refCount === 0 && !isNonIdleThreadDetailSubscription(entry);} function attachThreadDetailSubscription(entry: ThreadDetailSubscriptionEntry): boolean {  if (entry.unsubscribeConnectionListener !== null) {    entry.unsubscribeConnectionListener();    entry.unsubscribeConnectionListener = null;  }  if (entry.unsubscribe !== NOOP) {    return true;  }   const connection = readEnvironmentConnection(entry.environmentId);  if (!connection) {    return false;  }   entry.unsubscribe = connection.client.orchestration.subscribeThread(    { threadId: entry.threadId },    (item) => {      if (item.kind === "snapshot") {        useStore.getState().syncServerThreadDetail(item.snapshot.thread, entry.environmentId);        return;      }      applyEnvironmentThreadDetailEvent(item.event, entry.environmentId);    },  );  return true;} function watchThreadDetailSubscriptionConnection(entry: ThreadDetailSubscriptionEntry): void {  if (entry.unsubscribeConnectionListener !== null) {    return;  }   entry.unsubscribeConnectionListener = subscribeEnvironmentConnections(() => {    if (attachThreadDetailSubscription(entry)) {      entry.lastAccessedAt = Date.now();    }  });  attachThreadDetailSubscription(entry);} function disposeThreadDetailSubscriptionByKey(key: string): boolean {  const entry = threadDetailSubscriptions.get(key);  if (!entry) {    return false;  }   clearThreadDetailSubscriptionEviction(entry);  entry.unsubscribeConnectionListener?.();  entry.unsubscribeConnectionListener = null;  threadDetailSubscriptions.delete(key);  entry.unsubscribe();  entry.unsubscribe = NOOP;  return true;} function disposeThreadDetailSubscriptionsForEnvironment(environmentId: EnvironmentId): void {  for (const [key, entry] of threadDetailSubscriptions) {    if (entry.environmentId === environmentId) {      disposeThreadDetailSubscriptionByKey(key);    }  }} function detachThreadDetailSubscriptionsForEnvironment(environmentId: EnvironmentId): void {  for (const entry of threadDetailSubscriptions.values()) {    if (entry.environmentId !== environmentId) {      continue;    }    entry.unsubscribe();    entry.unsubscribe = NOOP;    watchThreadDetailSubscriptionConnection(entry);  }} function attachThreadDetailSubscriptionsForEnvironment(environmentId: EnvironmentId): void {  for (const entry of threadDetailSubscriptions.values()) {    if (entry.environmentId === environmentId) {      attachThreadDetailSubscription(entry);    }  }} function reconcileThreadDetailSubscriptionsForEnvironment(  environmentId: EnvironmentId,  threadIds: ReadonlyArray<ThreadId>,): void {  const activeThreadIds = new Set(threadIds);  for (const [key, entry] of threadDetailSubscriptions) {    if (entry.environmentId === environmentId && !activeThreadIds.has(entry.threadId)) {      disposeThreadDetailSubscriptionByKey(key);    }  }} function scheduleThreadDetailSubscriptionEviction(entry: ThreadDetailSubscriptionEntry): void {  clearThreadDetailSubscriptionEviction(entry);  if (!shouldEvictThreadDetailSubscription(entry)) {    return;  }   entry.evictionTimeoutId = setTimeout(() => {    const currentEntry = threadDetailSubscriptions.get(      getThreadDetailSubscriptionKey(entry.environmentId, entry.threadId),    );    if (!currentEntry) {      return;    }     currentEntry.evictionTimeoutId = null;    if (!shouldEvictThreadDetailSubscription(currentEntry)) {      return;    }    disposeThreadDetailSubscriptionByKey(      getThreadDetailSubscriptionKey(entry.environmentId, entry.threadId),    );  }, THREAD_DETAIL_SUBSCRIPTION_IDLE_EVICTION_MS);} function evictIdleThreadDetailSubscriptionsToCapacity(): void {  if (threadDetailSubscriptions.size <= MAX_CACHED_THREAD_DETAIL_SUBSCRIPTIONS) {    return;  }   const idleEntries = [...threadDetailSubscriptions.entries()]    .filter(([, entry]) => shouldEvictThreadDetailSubscription(entry))    .toSorted(([, left], [, right]) => left.lastAccessedAt - right.lastAccessedAt);   for (const [key] of idleEntries) {    if (threadDetailSubscriptions.size <= MAX_CACHED_THREAD_DETAIL_SUBSCRIPTIONS) {

Lines 1–500 of 1848. Load more