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>(); 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 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, });} function setRuntimeConnected(environmentId: EnvironmentId) { const connectedAt = isoNow(); useSavedEnvironmentRuntimeStore.getState().patch(environmentId, { connectionState: "connected", authState: "authenticated", connectedAt, disconnectedAt: null, lastError: null, lastErrorAt: null, }); useSavedEnvironmentRegistryStore.getState().markConnected(environmentId, connectedAt);} function setRuntimeDisconnected(environmentId: EnvironmentId, reason?: string | null) { useSavedEnvironmentRuntimeStore.getState().patch(environmentId, { connectionState: "disconnected", disconnectedAt: isoNow(), ...(reason && reason.trim().length > 0 ? { lastError: reason, lastErrorAt: isoNow(), } : {}), });} function setRuntimeError(environmentId: EnvironmentId, error: unknown) { useSavedEnvironmentRuntimeStore.getState().patch(environmentId, { connectionState: "error", ...getRuntimeErrorFields(error), });} function coalesceOrchestrationUiEvents( events: ReadonlyArray<OrchestrationEvent>,): OrchestrationEvent[] { if (events.length < 2) { return [...events]; } const coalesced: OrchestrationEvent[] = []; for (const event of events) { const previous = coalesced.at(-1); if ( previous?.type === "thread.message-sent" && event.type === "thread.message-sent" && previous.payload.threadId === event.payload.threadId && previous.payload.messageId === event.payload.messageId ) { coalesced[coalesced.length - 1] = { ...event, payload: { ...event.payload, attachments: event.payload.attachments ?? previous.payload.attachments, createdAt: previous.payload.createdAt, text: !event.payload.streaming && event.payload.text.length > 0 ? event.payload.text : previous.payload.text + event.payload.text, }, }; continue; } coalesced.push(event); } return coalesced;} function syncProjectUiFromStore() { const projects = selectProjectsAcrossEnvironments(useStore.getState()); const clientSettings = getClientSettings(); useUiStateStore.getState().syncProjects( projects.map((project) => ({ key: derivePhysicalProjectKey(project), logicalKey: deriveLogicalProjectKeyFromSettings(project, clientSettings), cwd: project.cwd, })), );} function syncThreadUiFromStore() { const threads = selectThreadsAcrossEnvironments(useStore.getState()); useUiStateStore.getState().syncThreads( threads.map((thread) => ({ key: scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), seedVisitedAt: thread.updatedAt ?? thread.createdAt, })), ); markPromotedDraftThreadsByRef( threads.map((thread) => scopeThreadRef(thread.environmentId, thread.id)), );} function reconcileSnapshotDerivedState() { syncProjectUiFromStore(); syncThreadUiFromStore();Lines 1–500 of 1095. Load more
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>(); 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 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, }); } function setRuntimeConnected(environmentId: EnvironmentId) { const connectedAt = isoNow(); useSavedEnvironmentRuntimeStore.getState().patch(environmentId, { connectionState: "connected", authState: "authenticated", connectedAt, disconnectedAt: null, lastError: null, lastErrorAt: null, }); useSavedEnvironmentRegistryStore.getState().markConnected(environmentId, connectedAt); } function setRuntimeDisconnected(environmentId: EnvironmentId, reason?: string | null) { useSavedEnvironmentRuntimeStore.getState().patch(environmentId, { connectionState: "disconnected", disconnectedAt: isoNow(), ...(reason && reason.trim().length > 0 ? { lastError: reason, lastErrorAt: isoNow(), } : {}), }); } function setRuntimeError(environmentId: EnvironmentId, error: unknown) { useSavedEnvironmentRuntimeStore.getState().patch(environmentId, { connectionState: "error", ...getRuntimeErrorFields(error), }); } function coalesceOrchestrationUiEvents( events: ReadonlyArray<OrchestrationEvent>, ): OrchestrationEvent[] { if (events.length < 2) { return [...events]; } const coalesced: OrchestrationEvent[] = []; for (const event of events) { const previous = coalesced.at(-1); if ( previous?.type === "thread.message-sent" && event.type === "thread.message-sent" && previous.payload.threadId === event.payload.threadId && previous.payload.messageId === event.payload.messageId ) { coalesced[coalesced.length - 1] = { ...event, payload: { ...event.payload, attachments: event.payload.attachments ?? previous.payload.attachments, createdAt: previous.payload.createdAt, text: !event.payload.streaming && event.payload.text.length > 0 ? event.payload.text : previous.payload.text + event.payload.text, }, }; continue; } coalesced.push(event); } return coalesced; } function syncProjectUiFromStore() { const projects = selectProjectsAcrossEnvironments(useStore.getState()); const clientSettings = getClientSettings(); useUiStateStore.getState().syncProjects( projects.map((project) => ({ key: derivePhysicalProjectKey(project), logicalKey: deriveLogicalProjectKeyFromSettings(project, clientSettings), cwd: project.cwd, })), ); } function syncThreadUiFromStore() { const threads = selectThreadsAcrossEnvironments(useStore.getState()); useUiStateStore.getState().syncThreads( threads.map((thread) => ({ key: scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), seedVisitedAt: thread.updatedAt ?? thread.createdAt, })), ); markPromotedDraftThreadsByRef( threads.map((thread) => scopeThreadRef(thread.environmentId, thread.id)), ); } function reconcileSnapshotDerivedState() { syncProjectUiFromStore(); syncThreadUiFromStore();