import {  type AuthSessionRole,  type EnvironmentId,  type OrchestrationEvent,  type OrchestrationShellSnapshot,  type OrchestrationShellStreamEvent,  type ServerConfig,  type TerminalEvent,  ThreadId,} from "@t3tools/contracts";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 { collectActiveTerminalThreadIds } from "~/lib/terminalStateCleanup";import { deriveOrchestrationBatchEffects } from "~/orchestrationEventEffects";import { projectQueryKeys } from "~/lib/projectReactQuery";import { providerQueryKeys } from "~/lib/providerReactQuery";import { getPrimaryKnownEnvironment } from "../primary";import {  bootstrapRemoteBearerSession,  fetchRemoteEnvironmentDescriptor,  fetchRemoteSessionState,  resolveRemoteWebSocketConnectionUrl,} from "../remote/api";import { resolveRemotePairingTarget } from "../remote/target";import {  getSavedEnvironmentRecord,  hasSavedEnvironmentRegistryHydrated,  listSavedEnvironmentRecords,  persistSavedEnvironmentRecord,  readSavedEnvironmentBearerToken,  removeSavedEnvironmentBearerToken,  type SavedEnvironmentRecord,  useSavedEnvironmentRegistryStore,  useSavedEnvironmentRuntimeStore,  waitForSavedEnvironmentRegistryHydration,  writeSavedEnvironmentBearerToken,} from "./catalog";import { createEnvironmentConnection, type EnvironmentConnection } from "./connection";import {  useStore,  selectProjectsAcrossEnvironments,  selectSidebarThreadSummaryByRef,  selectThreadByRef,  selectThreadsAcrossEnvironments,} from "~/store";import { useTerminalStateStore } from "~/terminalStateStore";import { useUiStateStore } from "~/uiStateStore";import { WsTransport } from "../../rpc/wsTransport";import { createWsRpcClient, type WsRpcClient } from "../../rpc/wsRpcClient";import {  deriveLogicalProjectKeyFromSettings,  derivePhysicalProjectKey,} from "../../logicalProject";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>();const environmentConnectionListeners = new Set<() => void>();const threadDetailSubscriptions = new Map<string, ThreadDetailSubscriptionEntry>();const lastAppliedProjectionVersionByEnvironment = new Map<  EnvironmentId,  {    readonly sequence: number;    readonly updatedAt: string | null;  }>(); let activeService: EnvironmentServiceState | null = null;let needsProviderInvalidation = false; // 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 NOOP = () => undefined; 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 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) {      return;    }    disposeThreadDetailSubscriptionByKey(key);  }} function reconcileThreadDetailSubscriptionEvictionState(  entry: ThreadDetailSubscriptionEntry,): void {  clearThreadDetailSubscriptionEviction(entry);  if (!shouldEvictThreadDetailSubscription(entry)) {    return;  }   scheduleThreadDetailSubscriptionEviction(entry);} function reconcileThreadDetailSubscriptionEvictionForThread(  environmentId: EnvironmentId,  threadId: ThreadId,): void {  const entry = threadDetailSubscriptions.get(    getThreadDetailSubscriptionKey(environmentId, threadId),  );  if (!entry) {    return;  }   reconcileThreadDetailSubscriptionEvictionState(entry);} function reconcileThreadDetailSubscriptionEvictionForEnvironment(  environmentId: EnvironmentId,): void {  for (const entry of threadDetailSubscriptions.values()) {    if (entry.environmentId === environmentId) {      reconcileThreadDetailSubscriptionEvictionState(entry);    }  }  evictIdleThreadDetailSubscriptionsToCapacity();} export function retainThreadDetailSubscription(  environmentId: EnvironmentId,  threadId: ThreadId,): () => void {  const key = getThreadDetailSubscriptionKey(environmentId, threadId);  const existing = threadDetailSubscriptions.get(key);  if (existing) {    clearThreadDetailSubscriptionEviction(existing);    existing.refCount += 1;    existing.lastAccessedAt = Date.now();    if (!attachThreadDetailSubscription(existing)) {      watchThreadDetailSubscriptionConnection(existing);    }    let released = false;    return () => {      if (released) {        return;      }      released = true;      existing.refCount = Math.max(0, existing.refCount - 1);      existing.lastAccessedAt = Date.now();      if (existing.refCount === 0) {        reconcileThreadDetailSubscriptionEvictionState(existing);        evictIdleThreadDetailSubscriptionsToCapacity();      }    };  }   const entry: ThreadDetailSubscriptionEntry = {    environmentId,    threadId,    unsubscribe: NOOP,    unsubscribeConnectionListener: null,    refCount: 1,    lastAccessedAt: Date.now(),    evictionTimeoutId: null,  };  threadDetailSubscriptions.set(key, entry);  if (!attachThreadDetailSubscription(entry)) {    watchThreadDetailSubscriptionConnection(entry);  }  evictIdleThreadDetailSubscriptionsToCapacity();   let released = false;  return () => {    if (released) {      return;    }    released = true;    entry.refCount = Math.max(0, entry.refCount - 1);    entry.lastAccessedAt = Date.now();    if (entry.refCount === 0) {      reconcileThreadDetailSubscriptionEvictionState(entry);      evictIdleThreadDetailSubscriptionsToCapacity();    }  };} function emitEnvironmentConnectionRegistryChange() {  for (const listener of environmentConnectionListeners) {    listener();  }} function getRuntimeErrorFields(error: unknown) {  return {    lastError: error instanceof Error ? error.message : String(error),    lastErrorAt: new Date().toISOString(),  } as const;} function isoNow(): string {  return new Date().toISOString();} function setRuntimeConnecting(environmentId: EnvironmentId) {  useSavedEnvironmentRuntimeStore.getState().patch(environmentId, {    connectionState: "connecting",    lastError: null,    lastErrorAt: null,  });}

Lines 1–500 of 1216. Load more