[URGENT] fix(dev): reduce instrumentation executor fan-out (phase 3) (#12078)

* fix(dev): reduce instrumentation executor fan-out

* fix(ci): reduce credential refresh complexity

---------

Co-authored-by: backryun <backryun@daonlab.local>

src/instrumentation-node.ts

@@ -319,7 +319,7 @@ export async function registerNodejs(): Promise<void> {
   process.title = renameProcessTitle(process.title);
 
   // Initialize proxy fetch patch FIRST (before any HTTP requests)
-  await import("@omniroute/open-sse/index.ts");
+  await import("@omniroute/open-sse/utils/proxyFetch.ts");
   console.log("[STARTUP] Global fetch proxy patch initialized");
 
   // Register quota fetchers early so combo routing can use real quota-aware

src/lib/providers/validation/anthropicFormat.ts

@@ -9,6 +9,7 @@ import {
   joinClaudeCodeCompatibleUrl,
   joinBaseUrlAndPath,
 } from "@omniroute/open-sse/services/claudeCodeCompatible.ts";
+import { getDefaultExecutor } from "@omniroute/open-sse/executors/defaultResolver.ts";
 import {
   addModelsSuffix,
   normalizeAnthropicBaseUrl,
@@ -137,8 +138,7 @@ export async function validateClaudeOAuthInline({
     typeof override === "string" && override ? override : modelId || "claude-haiku-4-5-20251001";
 
   try {
-    const { getExecutor } = await import("@omniroute/open-sse/executors/index.ts");
-    const executed = await (await getExecutor("claude")).execute({
+    const executed = await getDefaultExecutor("claude").execute({
       model: testModelId,
       body: {
         model: testModelId,

src/lib/services/quotaAutoPing.ts

@@ -22,13 +22,12 @@
 
 import { logger } from "@omniroute/open-sse/utils/logger.ts";
 import { sanitizeErrorMessage } from "@omniroute/open-sse/utils/error.ts";
-import { getExecutor } from "@omniroute/open-sse/executors/index.ts";
 import type { BaseExecutor } from "@omniroute/open-sse/executors/base";
 import { getCodexUsage } from "@omniroute/open-sse/services/usage/codex.ts";
 import { getSettings } from "@/lib/db/settings";
 import { getProviderConnections, updateProviderConnection } from "@/lib/db/providers";
 import { isConnectionUnavailableToAuxiliaryActivity } from "@/lib/exclusiveLeaseIsolation";
-import { refreshAndUpdateCredentials } from "@/lib/usage/providerLimits";
+import { refreshAndUpdateCredentialsWithResolver } from "@/lib/usage/providerLimits/credentialRefresh";
 import { getCircuitBreaker } from "@/shared/utils/circuitBreaker";
 import {
   QUOTA_AUTOPING_FAILURE_COOLDOWN_MS,
@@ -63,8 +62,11 @@ export interface QuotaAutoPingDeps {
   refreshAndUpdateCredentials: (
     connection: QuotaAutoPingConnection
   ) => Promise<{ connection: QuotaAutoPingConnection }>;
-  getCodexUsage: (accessToken?: string, providerSpecificData?: JsonRecord) => Promise<JsonRecord>;
-  getExecutor: (provider: string) => Promise<BaseExecutor>;
+  getCodexUsage: (
+    accessToken?: string,
+    providerSpecificData?: JsonRecord
+  ) => Promise<JsonRecord>;
+  getExecutor: (provider: "codex") => Promise<BaseExecutor>;
   canExecuteProvider: (provider: string) => boolean;
   isConnectionUnavailableToAuxiliaryActivity: (connectionId: string) => Promise<boolean>;
 }
@@ -79,15 +81,33 @@ export function createQuotaAutoPingState(): QuotaAutoPingState {
   return { running: false, resetCache: {}, failureCache: {} };
 }
 
+let codexExecutorPromise: Promise<BaseExecutor> | null = null;
+
+async function loadQuotaAutoPingExecutor(provider: string): Promise<BaseExecutor> {
+  if (provider !== "codex") {
+    throw new Error(`Quota auto-ping does not support provider "${provider}"`);
+  }
+
+  try {
+    codexExecutorPromise ??= import("@omniroute/open-sse/executors/codex.ts").then(
+      ({ CodexExecutor }) => new CodexExecutor()
+    );
+    return await codexExecutorPromise;
+  } catch (error) {
+    codexExecutorPromise = null;
+    throw error;
+  }
+}
+
 export function createDefaultQuotaAutoPingDeps(): QuotaAutoPingDeps {
   return {
     getSettings,
     getProviderConnections,
     updateProviderConnection,
     refreshAndUpdateCredentials: async (connection) =>
-      refreshAndUpdateCredentials(connection as never),
+      refreshAndUpdateCredentialsWithResolver(connection, loadQuotaAutoPingExecutor),
     getCodexUsage,
-    getExecutor,
+    getExecutor: loadQuotaAutoPingExecutor,
     canExecuteProvider: (provider) => getCircuitBreaker(provider).canExecute(),
     isConnectionUnavailableToAuxiliaryActivity,
   };

src/lib/usage/providerLimits.ts

@@ -18,13 +18,10 @@ import { clearRecoveredProviderState } from "@/sse/services/auth";
 import { getMachineId } from "@/shared/utils/machine";
 import { USAGE_SUPPORTED_PROVIDERS } from "@/shared/constants/providers";
 import { mergeProviderLimitsCacheEntry, toProviderLimitsCacheEntry } from "./providerLimitsCache";
-import { getExecutor } from "@omniroute/open-sse/executors/index.ts";
+import { getCredentialRefreshExecutor } from "@omniroute/open-sse/executors/credential.ts";
 import { getUsageForProvider } from "@omniroute/open-sse/services/usage.ts";
 import { cooldownUntilMs } from "@omniroute/open-sse/services/accountFallback.ts";
-import {
-  rotationGroupFor,
-  serializeRefresh,
-} from "@omniroute/open-sse/services/refreshSerializer.ts";
+import { rotationGroupFor } from "@omniroute/open-sse/services/refreshSerializer.ts";
 import {
   extractCodeAssistOnboardTierId,
   extractCodeAssistSubscriptionTier,
@@ -42,29 +39,15 @@ import {
   sanitizeUsageQuotasForProvider,
 } from "./providerLimits/quotaNormalize";
 import { syncInChunksWithSpacing } from "./providerLimits/chunkedSpacingSync";
+import {
+  refreshAndUpdateCredentialsWithResolver,
+  type CredentialRefreshOptions,
+  type ProviderConnectionLike,
+} from "./providerLimits/credentialRefresh";
+export { shouldAttemptRotatingRefresh } from "./providerLimits/credentialRefresh";
 type JsonRecord = Record<string, unknown>;
 type SyncSource = "manual" | "scheduled";
 
-interface ProviderConnectionLike {
-  id: string;
-  provider: string;
-  authType?: string;
-  accessToken?: string;
-  refreshToken?: string;
-  expiresAt?: string;
-  tokenExpiresAt?: string;
-  providerSpecificData?: JsonRecord;
-  testStatus?: string;
-  isActive?: boolean;
-  lastError?: string | null;
-  lastErrorAt?: string | null;
-  lastErrorType?: string | null;
-  lastErrorSource?: string | null;
-  errorCode?: string | number | null;
-  rateLimitedUntil?: string | null;
-  backoffLevel?: number;
-}
-
 const PROVIDER_LIMITS_APIKEY_PROVIDERS = new Set([
   "glm",
   "glm-cn",
@@ -218,122 +201,15 @@ async function syncToCloudIfEnabled() {
   }
 }
 
-/**
- * Whether the quota path may refresh this provider's token. Exported for testing.
- *
- * Rotating-refresh providers (Codex/OpenAI share one Auth0 client_id, etc.) mint a
- * single-use refresh_token on every refresh. The BULK quota-sync path runs many
- * connections concurrently; refreshing sibling accounts in parallel makes Auth0
- * revoke the whole token family (openai/codex#9648) and kills every account but
- * the last (#3019). So the bulk path never refreshes rotating providers
- * (`allowRotatingRefresh` falsy). The on-demand, per-connection path opts in and
- * is made safe by `serializeRefresh` (one token mint at a time per rotation group,
- * so even N concurrent per-account requests can never refresh siblings in
- * parallel). Non-rotating providers are always eligible.
- */
-export function shouldAttemptRotatingRefresh(
-  provider: string,
-  allowRotatingRefresh: boolean | undefined
-): boolean {
-  if (rotationGroupFor(provider) === null) return true;
-  return allowRotatingRefresh === true;
-}
-
 export async function refreshAndUpdateCredentials(
   connection: ProviderConnectionLike,
-  opts: { allowRotatingRefresh?: boolean; force?: boolean } = {}
+  opts: CredentialRefreshOptions = {}
 ) {
-  if (!shouldAttemptRotatingRefresh(connection.provider, opts.allowRotatingRefresh)) {
-    return { connection, refreshed: false };
-  }
-  const executor = await getExecutor(connection.provider);
-  const credentials = {
-    connectionId: connection.id,
-    accessToken: connection.accessToken,
-    refreshToken: connection.refreshToken,
-    expiresAt: connection.tokenExpiresAt || connection.expiresAt || null,
-    providerSpecificData: connection.providerSpecificData,
-    copilotToken: connection.providerSpecificData?.copilotToken,
-    copilotTokenExpiresAt: connection.providerSpecificData?.copilotTokenExpiresAt,
-  };
-
-  // `force` is used ONLY on the reactive 401 recovery path (a usage fetch came
-  // back unauthorized) — it bypasses the proactive `needsRefresh` heuristic so
-  // imported accounts (expiresAt=null, where needsRefresh is always false) can
-  // still re-mint. The mint stays serialized per rotation group; this never
-  // refreshes proactively from the bulk path (#3019 guard above is unchanged).
-  if (!opts.force && !executor.needsRefresh(credentials)) {
-    return { connection, refreshed: false };
-  }
-
-  // Serialize the actual token mint per rotation group so two sibling accounts
-  // never hit Auth0 concurrently (passthrough for non-rotating providers).
-  const refreshResult = (await serializeRefresh(connection.provider, () =>
-    executor.refreshCredentials(credentials, console)
-  )) as
-    | (JsonRecord & {
-        accessToken?: string;
-        refreshToken?: string;
-        expiresIn?: number;
-        expiresAt?: string;
-        copilotToken?: string;
-        copilotTokenExpiresAt?: string;
-      })
-    | null;
-
-  if (!refreshResult) {
-    // Refresh failed but we still have an accessToken — fall back to the
-    // existing token for ANY OAuth provider (graceful degradation) instead of
-    // hard-failing. Previously this was qualified to `provider === "github"`,
-    // which left every other provider stuck on a transient refresh failure even
-    // when a usable access token was still on hand.
-    if (connection.accessToken) {
-      return { connection, refreshed: false };
-    }
-    throw withStatus(
-      new Error("Failed to refresh credentials. Please re-authorize the connection."),
-      401
-    );
-  }
-
-  const updateData: JsonRecord = {
-    updatedAt: new Date().toISOString(),
-  };
-
-  if (refreshResult.accessToken) {
-    updateData.accessToken = refreshResult.accessToken;
-  }
-  if (refreshResult.refreshToken) {
-    updateData.refreshToken = refreshResult.refreshToken;
-  }
-  if (refreshResult.expiresIn) {
-    const expiresAt = new Date(Date.now() + refreshResult.expiresIn * 1000).toISOString();
-    updateData.expiresAt = expiresAt;
-    updateData.tokenExpiresAt = expiresAt;
-  } else if (refreshResult.expiresAt) {
-    updateData.expiresAt = refreshResult.expiresAt;
-    updateData.tokenExpiresAt = refreshResult.expiresAt;
-  }
-  if (refreshResult.copilotToken || refreshResult.copilotTokenExpiresAt) {
-    updateData.providerSpecificData = {
-      ...(connection.providerSpecificData || {}),
-      copilotToken: refreshResult.copilotToken,
-      copilotTokenExpiresAt: refreshResult.copilotTokenExpiresAt,
-    };
-  }
-
-  await updateProviderConnection(connection.id, updateData);
-
-  return {
-    connection: {
-      ...connection,
-      ...updateData,
-      providerSpecificData:
-        (updateData.providerSpecificData as JsonRecord | undefined) ||
-        connection.providerSpecificData,
-    },
-    refreshed: true,
-  };
+  return refreshAndUpdateCredentialsWithResolver(
+    connection,
+    getCredentialRefreshExecutor,
+    opts
+  );
 }
 
 function isUsageAuthError(message: unknown): boolean {

src/lib/usage/providerLimits/credentialRefresh.ts

@@ -0,0 +1,150 @@
+import { updateProviderConnection } from "@/lib/db/providers";
+import type { BaseExecutor } from "@omniroute/open-sse/executors/base";
+import {
+  rotationGroupFor,
+  serializeRefresh,
+} from "@omniroute/open-sse/services/refreshSerializer.ts";
+
+type JsonRecord = Record<string, unknown>;
+
+type CredentialRefreshResult = JsonRecord & {
+  accessToken?: string;
+  refreshToken?: string;
+  expiresIn?: number;
+  expiresAt?: string;
+  copilotToken?: string;
+  copilotTokenExpiresAt?: string;
+};
+
+export interface ProviderConnectionLike {
+  id: string;
+  provider: string;
+  authType?: string;
+  accessToken?: string;
+  refreshToken?: string;
+  expiresAt?: string | null;
+  tokenExpiresAt?: string | null;
+  providerSpecificData?: JsonRecord;
+  testStatus?: string;
+  isActive?: boolean;
+  lastError?: string | null;
+  lastErrorAt?: string | null;
+  lastErrorType?: string | null;
+  lastErrorSource?: string | null;
+  errorCode?: string | number | null;
+  rateLimitedUntil?: string | null;
+  backoffLevel?: number;
+}
+
+export interface CredentialRefreshOptions {
+  allowRotatingRefresh?: boolean;
+  force?: boolean;
+}
+
+export type CredentialExecutorResolver = (provider: string) => Promise<BaseExecutor>;
+
+function withStatus(error: Error, status: number): Error & { status: number } {
+  return Object.assign(error, { status });
+}
+
+/**
+ * Whether the quota path may refresh this provider's token.
+ *
+ * Rotating-refresh providers mint a single-use refresh token on every refresh,
+ * so bulk quota sync must not refresh siblings concurrently. The on-demand path
+ * explicitly opts in and remains serialized per rotation group.
+ */
+export function shouldAttemptRotatingRefresh(
+  provider: string,
+  allowRotatingRefresh: boolean | undefined
+): boolean {
+  if (rotationGroupFor(provider) === null) return true;
+  return allowRotatingRefresh === true;
+}
+
+function buildCredentialUpdateData(
+  connection: ProviderConnectionLike,
+  refreshResult: CredentialRefreshResult
+): JsonRecord {
+  const updateData: JsonRecord = {
+    updatedAt: new Date().toISOString(),
+  };
+
+  if (refreshResult.accessToken) {
+    updateData.accessToken = refreshResult.accessToken;
+  }
+  if (refreshResult.refreshToken) {
+    updateData.refreshToken = refreshResult.refreshToken;
+  }
+  if (refreshResult.expiresIn) {
+    const expiresAt = new Date(Date.now() + refreshResult.expiresIn * 1000).toISOString();
+    updateData.expiresAt = expiresAt;
+    updateData.tokenExpiresAt = expiresAt;
+  } else if (refreshResult.expiresAt) {
+    updateData.expiresAt = refreshResult.expiresAt;
+    updateData.tokenExpiresAt = refreshResult.expiresAt;
+  }
+  if (refreshResult.copilotToken || refreshResult.copilotTokenExpiresAt) {
+    updateData.providerSpecificData = {
+      ...(connection.providerSpecificData || {}),
+      copilotToken: refreshResult.copilotToken,
+      copilotTokenExpiresAt: refreshResult.copilotTokenExpiresAt,
+    };
+  }
+
+  return updateData;
+}
+
+/** Refresh and persist credentials using a caller-supplied executor resolver. */
+export async function refreshAndUpdateCredentialsWithResolver(
+  connection: ProviderConnectionLike,
+  resolveExecutor: CredentialExecutorResolver,
+  opts: CredentialRefreshOptions = {}
+) {
+  if (!shouldAttemptRotatingRefresh(connection.provider, opts.allowRotatingRefresh)) {
+    return { connection, refreshed: false };
+  }
+  const executor = await resolveExecutor(connection.provider);
+  const credentials = {
+    connectionId: connection.id,
+    accessToken: connection.accessToken,
+    refreshToken: connection.refreshToken,
+    expiresAt: connection.tokenExpiresAt || connection.expiresAt || null,
+    providerSpecificData: connection.providerSpecificData,
+    copilotToken: connection.providerSpecificData?.copilotToken,
+    copilotTokenExpiresAt: connection.providerSpecificData?.copilotTokenExpiresAt,
+  };
+
+  if (!opts.force && !executor.needsRefresh(credentials)) {
+    return { connection, refreshed: false };
+  }
+
+  const refreshResult = (await serializeRefresh(connection.provider, () =>
+    executor.refreshCredentials(credentials, console)
+  )) as CredentialRefreshResult | null;
+
+  if (!refreshResult) {
+    if (connection.accessToken) {
+      return { connection, refreshed: false };
+    }
+    throw withStatus(
+      new Error("Failed to refresh credentials. Please re-authorize the connection."),
+      401
+    );
+  }
+
+  const updateData = buildCredentialUpdateData(connection, refreshResult);
+
+  await updateProviderConnection(connection.id, updateData);
+
+  return {
+    connection: {
+      ...connection,
+      ...updateData,
+      providerSpecificData:
+        (updateData.providerSpecificData as JsonRecord | undefined) ||
+        connection.providerSpecificData,
+    },
+    refreshed: true,
+  };
+}