From 565eb281a7b8ffdd234bf1422505c09669bab67a Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Thu, 18 Jun 2026 19:12:07 -0300 Subject: [PATCH 1/4] refactor(combo): extract reset-aware quota block to combo/quotaStrategies.ts (QG v2 Fase 9 T5 D7b) (#4204) --- config/quality/file-size-baseline.json | 3 +- open-sse/services/combo.ts | 849 +-------------------- open-sse/services/combo/quotaScoring.ts | 311 ++++++++ open-sse/services/combo/quotaStrategies.ts | 568 ++++++++++++++ 4 files changed, 905 insertions(+), 826 deletions(-) create mode 100644 open-sse/services/combo/quotaScoring.ts create mode 100644 open-sse/services/combo/quotaStrategies.ts diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json index 8ab2b2c98d..49abbf7b93 100644 --- a/config/quality/file-size-baseline.json +++ b/config/quality/file-size-baseline.json @@ -12,6 +12,7 @@ "_rebaseline_2026_06_18_qg9_combo_split_d6": "QG v2 Fase 9 T5 D6: combo.ts 4430->3819 — combo structure resolution extracted byte-identically to the new open-sse/services/combo/comboStructure.ts (638, 3432 — auto-strategy scoring/intent/tag-routing extracted byte-identically to the new open-sse/services/combo/autoStrategy.ts (434, autoStrategy import cycle (a later task can move the reset-window block out of combo.ts without breaking autoStrategy). autoStrategy.ts never imports from the combo barrel. dedupeTargetsByExecutionKey was already in comboData.ts (D6) and is NOT re-moved. QUOTA_SOFT_DEPRIORITIZE_FACTOR + setCandidateQuotaSoftPenalty stay re-exported from combo.ts for chatCore.ts's dynamic import(\"../services/combo\"); scoreAutoTargets/expandAutoComboCandidatePool keep their public re-export too. Orphaned combo.ts imports dropped where moved-out symbols stopped being referenced; ProviderCandidate/AutoProviderCandidate/HistoricalLatencyStatsEntry types are imported into combo.ts for buildAutoCandidates. Pure move, no logic change.", "_rebaseline_2026_06_18_qg9_combo_split_d7a": "QG v2 Fase 9 T5 D7a: combo.ts 3440->3398 (wc -l 3397 + 1, stacked on #4194 which added the passthrough-quota guards) — round-robin sticky state extracted byte-identically to the new open-sse/services/combo/rrState.ts (71, ) + rrStickyTargets, and the three helpers clampStickyRoundRobinTargetLimit/getStickyRoundRobinStartIndex/recordStickyRoundRobinSuccess. STATE COHESION: rrCounters and rrStickyTargets remain SINGLE instances defined once in rrState.ts; combo.ts imports the same references back and keeps mutating them directly in orderTargetsByResetAwareQuota/orderTargetsByResetWindow/handleRoundRobinCombo (no Map duplicated — sticky-3 round-robin behavior preserved, guarded by combo-routing-e2e.test.ts). The quota state left behind (MAX_RESET_AWARE_CACHE/resetAwareConnectionCache/resetAwareQuotaCache) stays in combo.ts for a later task (D7b). None of the 6 were ever public, so combo.ts imports all six back (no re-export). rrState.ts never imports from the combo barrel; it imports only ResolvedComboTarget from ./types.ts. Pure move, no logic change.", + "_rebaseline_2026_06_18_qg9_combo_split_d7b": "QG v2 Fase 9 T5 D7b: combo.ts 3398->2597 (wc -l 2596 + 1) — the reset-aware / reset-window quota block extracted byte-identically into two new leaves under open-sse/services/combo/ (both 800 cap): quotaScoring.ts (311) holds the PURE half (config consts/resolvers + window-math/scoring helpers, no state, no async); quotaStrategies.ts (568) holds the STATEFUL/async half — the two mutable caches resetAwareConnectionCache + resetAwareQuotaCache (new Map) + MAX_RESET_AWARE_CACHE kept as SINGLE instances next to their only readers/writers getQuotaAwareConnectionsForTarget + fetchResetAwareQuotaWithCache (STATE COHESION; grep 'resetAware*Cache = new Map' in combo.ts == 0), plus normalizeConnectionIds/filterAllowedConnectionIds/getTargetConnectionIds/mapWithConcurrency/preScreenTargets/orderTargetsByResetAwareQuota/orderTargetsByResetWindow. quotaStrategies imports rrCounters + MAX_RR_COUNTERS from ./rrState.ts (D7a) and the pure helpers from ./quotaScoring.ts; neither leaf imports the combo barrel. combo.ts imports back the 3 reset-window helpers buildAutoCandidates needs (resolveResetWindowConfig/fetchResetAwareQuotaWithCache/calculateResetWindowAffinity) + resolveSlaRoutingPolicy + the 3 orderers used by the orchestrator; preScreenTargets stays re-exported. Orphaned imports dropped (clamp01/PRE_SCREEN_CONCURRENCY/ProviderProfile/SlaRoutingPolicy/IsModelAvailable); hasPerModelQuota KEPT (used by the #4194 passthrough-quota guards that stay). buildAutoCandidates + calculateTargetContextAffinity/getBootstrapLatencyMs stay (D8). Pure move, no logic change.", "_rebaseline_2026_06_18_8_2_sliding_window": "Fase 8.2 own growth: rateLimitManager.ts 1017->1022 (+5 = one import + one `await awaitProviderDefaultSlot(...)` call + a 2-line comment at the existing withRateLimit chokepoint). All sliding-window logic was extracted to the new open-sse/services/providerDefaultRateLimit.ts + open-sse/services/slidingWindowLimiter.ts (both 1440 (+5 = appendNoThinkingVariants(finalModels) call + comment at the existing finalModels chokepoint) and chat.ts 1458->1471 (+13 = applyNoThinkingAlias(body) call + comment right after body.model is read, before model resolution). All real logic lives in the new open-sse/utils/noThinkingAlias.ts (6009 (+29 at the existing streaming-return chokepoint = capture streamRecovery.continueMidStream alongside .enabled; refactor the early-retry reopen thunk into a shared runUpstreamStream(body) helper — net DRY — and add the gated continueStream(assistantSoFar) thunk that re-runs the upstream with makeContinuationBody(bodyToSend, …), plus the onContinue log). All continuation logic (scanOpenAiSseText, makeContinuationBody, trimContinuationOverlap, the createRecoverableStream continuation path) lives in open-sse/services/streamRecovery.ts (; - -const MAX_RESET_AWARE_CACHE = 200; - -const resetAwareConnectionCache = new Map< - string, - { fetchedAt: number; connections: Array> } ->(); -const resetAwareQuotaCache = new Map< - string, - { fetchedAt: number; quota: unknown; refreshPromise: Promise | null } ->(); - -function finiteNumberOrNull(value: unknown): number | null { - const numericValue = Number(value); - return Number.isFinite(numericValue) ? numericValue : null; -} - -function getPercentConfig(value: unknown, fallback: number): number { - const numericValue = finiteNumberOrNull(value); - if (numericValue === null) return fallback; - return Math.max(0, Math.min(100, numericValue)); -} - -function getWeightConfig(value: unknown, fallback: number): number { - const numericValue = finiteNumberOrNull(value); - if (numericValue === null || numericValue < 0) return fallback; - return numericValue; -} - -function getDurationConfig(value: unknown, fallback: number, max: number): number { - const numericValue = finiteNumberOrNull(value); - if (numericValue === null || numericValue < 0) return fallback; - return Math.min(max, Math.floor(numericValue)); -} - -function resolveResetAwareConfig(config: Record | null | undefined) { - const sessionWeight = getWeightConfig( - config?.resetAwareSessionWeight, - RESET_AWARE_DEFAULTS.sessionWeight - ); - const weeklyWeight = getWeightConfig( - config?.resetAwareWeeklyWeight, - RESET_AWARE_DEFAULTS.weeklyWeight - ); - const totalWeight = sessionWeight + weeklyWeight; - const normalizedSessionWeight = - totalWeight > 0 ? sessionWeight / totalWeight : RESET_AWARE_DEFAULTS.sessionWeight; - - return { - sessionWeight: normalizedSessionWeight, - weeklyWeight: 1 - normalizedSessionWeight, - tieBand: - getPercentConfig(config?.resetAwareTieBandPercent, RESET_AWARE_DEFAULTS.tieBandPercent) / 100, - exhaustionGuard: - getPercentConfig( - config?.resetAwareExhaustionGuardPercent, - RESET_AWARE_DEFAULTS.exhaustionGuardPercent - ) / 100, - quotaCacheTtlMs: getDurationConfig(config?.resetAwareQuotaCacheTtlMs, 0, 300_000), - quotaCacheMaxStaleMs: getDurationConfig(config?.resetAwareQuotaCacheMaxStaleMs, 0, 3_600_000), - }; -} - -function resolveResetWindowConfig(config: Record | null | undefined) { - const rawWindows = Array.isArray(config?.resetWindowWindows) ? config.resetWindowWindows : null; - const windows = rawWindows - ?.filter((windowName): windowName is ResetWindowName => - (RESET_WINDOW_NAMES as readonly string[]).includes(String(windowName)) - ) - .filter((windowName, index, array) => array.indexOf(windowName) === index); - - const effectiveWindows = - windows && windows.length > 0 - ? windows - : config?.resetWindowIncludeSession === true - ? (["weekly", "session"] as ResetWindowName[]) - : (["weekly"] as ResetWindowName[]); - - return { - windows: effectiveWindows, - tieBandMs: Math.max( - 0, - finiteNumberOrNull(config?.resetWindowTieBandMs) ?? RESET_WINDOW_DEFAULT_TIE_BAND_MS - ), - quotaCacheTtlMs: getDurationConfig(config?.resetWindowQuotaCacheTtlMs, 0, 300_000), - quotaCacheMaxStaleMs: getDurationConfig(config?.resetWindowQuotaCacheMaxStaleMs, 0, 3_600_000), - }; -} - -function resolveSlaRoutingPolicy( - config: Record | null | undefined -): SlaRoutingPolicy | undefined { - if (!config) return undefined; - const nestedSla = isRecord(config.sla) ? config.sla : {}; - const targetP95Ms = finiteNumberOrNull(config.slaTargetP95Ms ?? nestedSla.targetP95Ms); - const maxErrorRate = finiteNumberOrNull(config.slaMaxErrorRate ?? nestedSla.maxErrorRate); - const maxCostPer1MTokens = finiteNumberOrNull( - config.slaMaxCostPer1MTokens ?? nestedSla.maxCostPer1MTokens - ); - const hardConstraints = config.slaHardConstraints ?? nestedSla.hardConstraints; - - const policy: SlaRoutingPolicy = {}; - if (targetP95Ms !== null && targetP95Ms > 0) policy.targetP95Ms = targetP95Ms; - if (maxErrorRate !== null && maxErrorRate >= 0) policy.maxErrorRate = clamp01(maxErrorRate); - if (maxCostPer1MTokens !== null && maxCostPer1MTokens > 0) { - policy.maxCostPer1MTokens = maxCostPer1MTokens; - } - if (typeof hardConstraints === "boolean") policy.hardConstraints = hardConstraints; - - return Object.keys(policy).length > 0 ? policy : undefined; -} - -function getResetAwareProvider(target: ResolvedComboTarget): string | null { - const provider = (target.providerId || target.provider || "").toLowerCase(); - return provider || null; -} - -function normalizeResetAt(value: unknown): string | null { - if (typeof value === "string" && value.trim().length > 0) return value.trim(); - if (typeof value === "number" && Number.isFinite(value)) return String(value); - return null; -} - -function parseResetTimeMs(resetAt: string | null | undefined): number { - if (!resetAt) return NaN; - const resetTime = Date.parse(resetAt); - if (Number.isFinite(resetTime)) return resetTime; - - if (!/^\d+(?:\.\d+)?$/.test(resetAt)) return NaN; - const numericResetAt = Number(resetAt); - if (!Number.isFinite(numericResetAt)) return NaN; - return numericResetAt < 10_000_000_000 ? numericResetAt * 1000 : numericResetAt; -} - -function getQuotaWindow( - quota: unknown, - key: "window5h" | "window7d" | "windowWeekly" | "windowMonthly" -): { percentUsed: number | null; resetAt: string | null } | null { - if (!isRecord(quota)) return null; - const window = quota[key]; - if (!isRecord(window)) return null; - const percentUsed = finiteNumberOrNull(window.percentUsed); - const resetAt = normalizeResetAt(window.resetAt); - return { percentUsed, resetAt }; -} - -function normalizeWindowPercentUsed(value: unknown): number | null { - const numericValue = finiteNumberOrNull(value); - if (numericValue === null) return null; - if (numericValue > 1) return clamp01(numericValue / 100); - return clamp01(numericValue); -} - -function getNamedQuotaWindow( - quota: unknown, - windowName: ResetWindowName -): { percentUsed: number | null; resetAt: string | null } | null { - if (!quota || !isRecord(quota)) return null; - - if (windowName === "session") return getQuotaWindow(quota, "window5h"); - if (windowName === "weekly") { - return getQuotaWindow(quota, "window7d") || getQuotaWindow(quota, "windowWeekly"); - } - if (windowName === "monthly") return getQuotaWindow(quota, "windowMonthly"); - - return null; -} - -function getWindowsMapQuotaWindow( - quota: unknown, - windowName: ResetWindowName -): { percentUsed: number | null; resetAt: string | null } | null { - if (!quota || !isRecord(quota) || !isRecord(quota.windows)) return null; - const candidates = Object.entries(quota.windows) - .map(([key, value]) => ({ key: key.toLowerCase(), value })) - .filter(({ key }) => key === windowName || key.startsWith(`${windowName} `)); - - if (candidates.length === 0) return null; - candidates.sort((a, b) => a.key.localeCompare(b.key)); - const window = candidates[0].value; - if (!isRecord(window)) return null; - - return { - percentUsed: normalizeWindowPercentUsed(window.percentUsed), - resetAt: normalizeResetAt(window.resetAt), - }; -} - -function resolveQuotaWindowByName( - quota: unknown, - windowName: ResetWindowName -): { percentUsed: number | null; resetAt: string | null } | null { - return getNamedQuotaWindow(quota, windowName) || getWindowsMapQuotaWindow(quota, windowName); -} - -function getResetUrgency(resetAt: string | null | undefined, windowMs: number): number { - if (!resetAt) return 0.5; - const resetTime = parseResetTimeMs(resetAt); - if (!Number.isFinite(resetTime)) return 0.5; - const msUntilReset = resetTime - Date.now(); - if (msUntilReset <= 0) return 1; - return clamp01(1 - msUntilReset / windowMs); -} - -function scoreQuotaWindow( - remaining: number, - resetAt: string | null | undefined, - windowMs: number, - remainingWeight: number, - resetPressureWeight: number -): number { - const normalizedRemaining = clamp01(remaining); - const resetUrgency = getResetUrgency(resetAt, windowMs); - const resetPressure = resetUrgency * (1 - normalizedRemaining); - return remainingWeight * normalizedRemaining + resetPressureWeight * resetPressure; -} - -function scoreResetAwareQuota(quota: unknown, config: ReturnType) { - if (!quota || !isRecord(quota)) return { score: 0.5 }; - if (quota.limitReached === true) return { score: -Infinity }; - - const overallPercentUsed = clamp01(finiteNumberOrNull(quota.percentUsed) ?? 0.5); - const sessionWindow = getQuotaWindow(quota, "window5h"); - const weeklyWindow = getQuotaWindow(quota, "window7d") || getQuotaWindow(quota, "windowWeekly"); - const sessionRemaining = clamp01(1 - (sessionWindow?.percentUsed ?? overallPercentUsed)); - const weeklyRemaining = clamp01(1 - (weeklyWindow?.percentUsed ?? overallPercentUsed)); - const sessionScore = scoreQuotaWindow( - sessionRemaining, - sessionWindow?.resetAt, - RESET_AWARE_SESSION_WINDOW_MS, - RESET_AWARE_SESSION_REMAINING_WEIGHT, - RESET_AWARE_SESSION_RESET_PRESSURE_WEIGHT - ); - const weeklyScore = scoreQuotaWindow( - weeklyRemaining, - weeklyWindow?.resetAt ?? normalizeResetAt(quota.resetAt), - RESET_AWARE_WEEKLY_WINDOW_MS, - RESET_AWARE_WEEKLY_REMAINING_WEIGHT, - RESET_AWARE_WEEKLY_RESET_PRESSURE_WEIGHT - ); - let score = config.sessionWeight * sessionScore + config.weeklyWeight * weeklyScore; - - if (config.exhaustionGuard > 0 && sessionRemaining < config.exhaustionGuard) { - score *= Math.max(0.05, sessionRemaining / config.exhaustionGuard); - } - - return { score }; -} - -async function getQuotaAwareConnectionsForTarget( - target: ResolvedComboTarget, - connectionCache: Map>>, - connectionLoadPromises: Map>>>, - comboName: string, - log: { warn?: (...args: unknown[]) => void } -) { - const provider = getResetAwareProvider(target); - if (!provider || !getQuotaFetcher(provider)) return []; - if (!connectionCache.has(provider)) { - const cached = resetAwareConnectionCache.get(provider); - if (cached && Date.now() - cached.fetchedAt < RESET_AWARE_CONNECTION_CACHE_TTL_MS) { - connectionCache.set(provider, cached.connections); - return cached.connections; - } - - if (!connectionLoadPromises.has(provider)) { - connectionLoadPromises.set( - provider, - (async () => { - try { - const connections = await getProviderConnections({ provider, isActive: true }); - const activeConnections = Array.isArray(connections) - ? (connections as Array>) - : []; - if ( - !resetAwareConnectionCache.has(provider) && - resetAwareConnectionCache.size >= MAX_RESET_AWARE_CACHE - ) { - const oldest = resetAwareConnectionCache.keys().next().value; - if (oldest !== undefined) resetAwareConnectionCache.delete(oldest); - } - resetAwareConnectionCache.set(provider, { - connections: activeConnections, - fetchedAt: Date.now(), - }); - return activeConnections; - } catch (error) { - log.warn?.("COMBO", "Reset-aware failed to load quota-aware connections.", { - comboName, - err: error, - operation: "getProviderConnections", - provider, - }); - return []; - } - })() - ); - } - - const connections = await connectionLoadPromises.get(provider)!; - connectionCache.set(provider, connections); - } - return connectionCache.get(provider) || []; -} - -function normalizeConnectionIds(value: unknown): string[] | null { - if (!Array.isArray(value)) return null; - const ids = value.filter( - (connectionId): connectionId is string => - typeof connectionId === "string" && connectionId.trim().length > 0 - ); - return ids.length > 0 ? ids : null; -} - -function filterAllowedConnectionIds( - connectionIds: string[], - apiKeyAllowedConnectionIds: string[] | null | undefined -): string[] { - const allowedIds = normalizeConnectionIds(apiKeyAllowedConnectionIds); - if (!allowedIds) return connectionIds; - const allowedSet = new Set(allowedIds); - return connectionIds.filter((connectionId) => allowedSet.has(connectionId)); -} - -function getTargetConnectionIds( - target: ResolvedComboTarget, - connections: Array> -): string[] { - let connectionIds: string[]; - if (target.connectionId) { - return [target.connectionId]; - } - - if (Array.isArray(target.allowedConnectionIds) && target.allowedConnectionIds.length > 0) { - return target.allowedConnectionIds.filter( - (connectionId): connectionId is string => - typeof connectionId === "string" && connectionId.trim().length > 0 - ); - } - - connectionIds = connections - .map((connection) => (typeof connection.id === "string" ? connection.id : null)) - .filter((connectionId): connectionId is string => !!connectionId); - return connectionIds; -} - -async function mapWithConcurrency( - items: T[], - concurrency: number, - mapper: (item: T, index: number) => Promise -): Promise { - const results = new Array(items.length); - let nextIndex = 0; - const workerCount = Math.max(1, Math.min(concurrency, items.length)); - - await Promise.all( - Array.from({ length: workerCount }, async () => { - while (nextIndex < items.length) { - const currentIndex = nextIndex++; - results[currentIndex] = await mapper(items[currentIndex], currentIndex); - } - }) - ); - - return results; -} - -async function fetchResetAwareQuotaWithCache({ - provider, - connectionId, - connection, - fetcher, - config, - log, - comboName, -}: { - provider: string; - connectionId: string; - connection?: Record; - fetcher: (connectionId: string, connection?: Record) => Promise; - config: QuotaFetchCacheConfig; - log: { debug?: (...args: unknown[]) => void; warn?: (...args: unknown[]) => void }; - comboName: string; -}): Promise { - const cacheKey = `${provider}:${connectionId}`; - const ttlMs = config.quotaCacheTtlMs; - const maxStaleMs = config.quotaCacheMaxStaleMs; - const now = Date.now(); - const cached = resetAwareQuotaCache.get(cacheKey); - - if (ttlMs <= 0 && maxStaleMs <= 0) { - try { - return await fetcher(connectionId, connection); - } catch (error) { - log.warn?.("COMBO", "Reset-aware quota fetch failed.", { - comboName, - connectionId, - err: error, - operation: "quotaFetch", - provider, - }); - return null; - } - } - - const refresh = () => { - const existing = resetAwareQuotaCache.get(cacheKey); - if (existing?.refreshPromise != null) return existing.refreshPromise; - - const refreshPromise = fetcher(connectionId, connection) - .then((quota) => { - if (quota) { - if ( - !resetAwareQuotaCache.has(cacheKey) && - resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE - ) { - const oldest = resetAwareQuotaCache.keys().next().value; - if (oldest !== undefined) resetAwareQuotaCache.delete(oldest); - } - resetAwareQuotaCache.set(cacheKey, { - quota, - fetchedAt: Date.now(), - refreshPromise: null, - }); - } else { - resetAwareQuotaCache.delete(cacheKey); - } - return quota; - }) - .catch((error) => { - const previous = resetAwareQuotaCache.get(cacheKey); - if (previous) { - if ( - !resetAwareQuotaCache.has(cacheKey) && - resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE - ) { - const oldest = resetAwareQuotaCache.keys().next().value; - if (oldest !== undefined) resetAwareQuotaCache.delete(oldest); - } - resetAwareQuotaCache.set(cacheKey, { ...previous, refreshPromise: null }); - } - log.warn?.("COMBO", "Reset-aware quota fetch failed.", { - comboName, - connectionId, - err: error, - operation: "quotaFetch", - provider, - }); - return null; - }); - - if (!resetAwareQuotaCache.has(cacheKey) && resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE) { - const oldest = resetAwareQuotaCache.keys().next().value; - if (oldest !== undefined) resetAwareQuotaCache.delete(oldest); - } - resetAwareQuotaCache.set(cacheKey, { - quota: existing?.quota ?? cached?.quota ?? null, - fetchedAt: existing?.fetchedAt ?? cached?.fetchedAt ?? 0, - refreshPromise, - }); - return refreshPromise; - }; - - if (ttlMs > 0 && cached) { - const age = now - cached.fetchedAt; - if (age <= ttlMs) return cached.quota; - if (maxStaleMs > 0 && age <= ttlMs + maxStaleMs) { - void refresh(); - return cached.quota; - } - } - - return refresh(); -} - -type PreScreenResult = { profile: ProviderProfile | null; available: boolean }; - -export async function preScreenTargets( - targets: ResolvedComboTarget[], - isModelAvailable?: IsModelAvailable | null -): Promise> { - if (targets.length === 0) { - return new Map(); - } - - const results = await mapWithConcurrency( - targets, - PRE_SCREEN_CONCURRENCY, - async (target): Promise<{ key: string; result: PreScreenResult }> => { - const profile = await getRuntimeProviderProfile(target.provider).catch(() => null); - - const breaker = getCircuitBreaker(target.provider); - if (breaker.getStatus().state === "OPEN") { - return { key: target.executionKey, result: { profile, available: false } }; - } - - let available = true; - if (isModelAvailable) { - // IsModelAvailable may return a sync boolean or a Promise; Promise.resolve - // normalizes both so the .catch() never runs against a bare boolean. - available = await Promise.resolve(isModelAvailable(target.modelStr, target)).catch( - () => true - ); - } - return { key: target.executionKey, result: { profile, available } }; - } - ); - - const map = new Map(); - for (const { key, result } of results) { - map.set(key, result); - } - return map; -} - -async function orderTargetsByResetAwareQuota( - targets: ResolvedComboTarget[], - comboName: string, - configSource: Record | null | undefined, - log: { warn?: (...args: unknown[]) => void }, - apiKeyAllowedConnectionIds?: string[] | null -) { - if (targets.length === 0) return targets; - - const config = resolveResetAwareConfig(configSource); - const connectionCache = new Map>>(); - const connectionLoadPromises = new Map>>>(); - const quotaPromises = new Map>(); - const connectionById = new Map>(); - const expandedTargets: ResolvedComboTarget[] = []; - - const targetsWithConnections = await Promise.all( - targets.map(async (target) => ({ - connections: await getQuotaAwareConnectionsForTarget( - target, - connectionCache, - connectionLoadPromises, - comboName, - log - ), - target, - })) - ); - - for (const { target, connections } of targetsWithConnections) { - for (const connection of connections) { - if (typeof connection.id === "string") connectionById.set(connection.id, connection); - } - - const unrestrictedConnectionIds = getTargetConnectionIds(target, connections); - const connectionIds = filterAllowedConnectionIds( - unrestrictedConnectionIds, - apiKeyAllowedConnectionIds - ); - if (connectionIds.length === 0) { - if ( - unrestrictedConnectionIds.length > 0 && - normalizeConnectionIds(apiKeyAllowedConnectionIds) - ) { - continue; - } - expandedTargets.push(target); - continue; - } - - for (const connectionId of connectionIds) { - expandedTargets.push({ - ...target, - connectionId, - executionKey: - target.connectionId === connectionId - ? target.executionKey - : `${target.executionKey}@${connectionId}`, - }); - } - } - - const scoredTargets = await mapWithConcurrency( - expandedTargets, - RESET_AWARE_QUOTA_FETCH_CONCURRENCY, - async (target, index) => { - let quota: unknown = null; - const provider = getResetAwareProvider(target); - const fetcher = provider ? getQuotaFetcher(provider) : null; - if (fetcher && provider && target.connectionId) { - const quotaKey = `${provider}:${target.connectionId}`; - if (!quotaPromises.has(quotaKey)) { - quotaPromises.set( - quotaKey, - fetchResetAwareQuotaWithCache({ - provider, - connectionId: target.connectionId, - connection: connectionById.get(target.connectionId), - fetcher, - config, - log, - comboName, - }) - ); - } - quota = await quotaPromises.get(quotaKey)!; - } - const { score } = scoreResetAwareQuota(quota, config); - return { target, score, index }; - } - ); - - scoredTargets.sort((a, b) => { - if (b.score !== a.score) return b.score - a.score; - return a.index - b.index; - }); - - const bestScore = scoredTargets[0]?.score ?? 0; - const tiedTargets = scoredTargets.filter((entry) => bestScore - entry.score <= config.tieBand); - let orderedTiedTargets = tiedTargets; - if (tiedTargets.length > 1) { - const key = `reset-aware:${comboName}`; - const counter = rrCounters.get(key) || 0; - if (!rrCounters.has(key) && rrCounters.size >= MAX_RR_COUNTERS) { - const oldest = rrCounters.keys().next().value; - if (oldest !== undefined) rrCounters.delete(oldest); - } - rrCounters.set(key, counter + 1); - const startIndex = counter % tiedTargets.length; - orderedTiedTargets = [...tiedTargets.slice(startIndex), ...tiedTargets.slice(0, startIndex)]; - } - - const tiedExecutionKeys = new Set(orderedTiedTargets.map((entry) => entry.target.executionKey)); - return [ - ...orderedTiedTargets, - ...scoredTargets.filter((entry) => !tiedExecutionKeys.has(entry.target.executionKey)), - ].map((entry) => entry.target); -} - -function getResetWindowTimestampMs(quota: unknown, windows: ResetWindowName[]): number { - if (!quota || !isRecord(quota) || quota.limitReached === true) return Infinity; - - let selectedResetMs = Infinity; - for (const windowName of windows) { - const window = resolveQuotaWindowByName(quota, windowName); - const resetMs = parseResetTimeMs(window?.resetAt ?? null); - if (Number.isFinite(resetMs)) { - selectedResetMs = Math.min(selectedResetMs, resetMs); - } - } - - if (!Number.isFinite(selectedResetMs)) { - selectedResetMs = parseResetTimeMs(normalizeResetAt(quota.resetAt)); - } - - return Number.isFinite(selectedResetMs) ? selectedResetMs : Infinity; -} - -function getResetWindowHorizonMs(windows: ResetWindowName[]): number { - if (windows.includes("monthly")) return 30 * 24 * 60 * 60 * 1000; - if (windows.includes("weekly")) return RESET_AWARE_WEEKLY_WINDOW_MS; - return RESET_AWARE_SESSION_WINDOW_MS; -} - -function calculateResetWindowAffinity(quota: unknown, config: ResetWindowConfig): number { - const resetMs = getResetWindowTimestampMs(quota, config.windows); - if (!Number.isFinite(resetMs)) return 0.5; - - const msUntilReset = resetMs - Date.now(); - if (msUntilReset <= 0) return 1; - return clamp01(1 - msUntilReset / getResetWindowHorizonMs(config.windows)); -} +// Reset-aware / reset-window quota config, scoring, and window-math helpers were +// extracted to combo/quotaScoring.ts (pure) and the stateful cache + strategy +// orderers to combo/quotaStrategies.ts (combo split D7b). The two cache Maps +// (resetAwareConnectionCache, resetAwareQuotaCache) live as single instances in +// quotaStrategies.ts alongside their only readers/writers (state cohesion). +// combo.ts imports back the three reset-window helpers buildAutoCandidates + +// orchestration need, plus the strategy orderers and preScreenTargets (above). // Bootstrap defaults from ClawRouter benchmark (used when no local latency history exists yet) const DEFAULT_MODEL_P95_MS: Record = { @@ -1069,137 +399,6 @@ export async function buildAutoCandidates( return candidates; } -async function orderTargetsByResetWindow( - targets: ResolvedComboTarget[], - comboName: string, - configSource: Record | null | undefined, - log: { warn?: (...args: unknown[]) => void }, - apiKeyAllowedConnectionIds?: string[] | null -) { - if (targets.length === 0) return targets; - - const config = resolveResetWindowConfig(configSource); - const connectionCache = new Map>>(); - const connectionLoadPromises = new Map>>>(); - const quotaPromises = new Map>(); - const connectionById = new Map>(); - const expandedTargets: ResolvedComboTarget[] = []; - - const targetsWithConnections = await Promise.all( - targets.map(async (target) => ({ - connections: await getQuotaAwareConnectionsForTarget( - target, - connectionCache, - connectionLoadPromises, - comboName, - log - ), - target, - })) - ); - - for (const { target, connections } of targetsWithConnections) { - for (const connection of connections) { - if (typeof connection.id === "string") connectionById.set(connection.id, connection); - } - - const unrestrictedConnectionIds = getTargetConnectionIds(target, connections); - const connectionIds = filterAllowedConnectionIds( - unrestrictedConnectionIds, - apiKeyAllowedConnectionIds - ); - if (connectionIds.length === 0) { - if ( - unrestrictedConnectionIds.length > 0 && - normalizeConnectionIds(apiKeyAllowedConnectionIds) - ) { - continue; - } - expandedTargets.push(target); - continue; - } - - for (const connectionId of connectionIds) { - expandedTargets.push({ - ...target, - connectionId, - executionKey: - target.connectionId === connectionId - ? target.executionKey - : `${target.executionKey}@${connectionId}`, - }); - } - } - - const scoredTargets = await mapWithConcurrency( - expandedTargets, - RESET_AWARE_QUOTA_FETCH_CONCURRENCY, - async (target, index) => { - let quota: unknown = null; - const provider = getResetAwareProvider(target); - const fetcher = provider ? getQuotaFetcher(provider) : null; - if (fetcher && provider && target.connectionId) { - const quotaKey = `${provider}:${target.connectionId}`; - if (!quotaPromises.has(quotaKey)) { - quotaPromises.set( - quotaKey, - fetchResetAwareQuotaWithCache({ - provider, - connectionId: target.connectionId, - connection: connectionById.get(target.connectionId), - fetcher, - config, - log, - comboName, - }) - ); - } - quota = await quotaPromises.get(quotaKey)!; - } - - return { - target, - resetMs: getResetWindowTimestampMs(quota, config.windows), - index, - }; - } - ); - - scoredTargets.sort((a, b) => { - if (a.resetMs !== b.resetMs) return a.resetMs - b.resetMs; - return a.index - b.index; - }); - - const bestResetMs = scoredTargets[0]?.resetMs ?? Infinity; - if (!Number.isFinite(bestResetMs) || config.tieBandMs <= 0) { - return scoredTargets.map((entry) => entry.target); - } - - const tiedTargets = scoredTargets.filter( - (entry) => entry.resetMs - bestResetMs <= config.tieBandMs - ); - if (tiedTargets.length <= 1) return scoredTargets.map((entry) => entry.target); - - const key = `reset-window:${comboName}`; - const counter = rrCounters.get(key) || 0; - if (!rrCounters.has(key) && rrCounters.size >= MAX_RR_COUNTERS) { - const oldest = rrCounters.keys().next().value; - if (oldest !== undefined) rrCounters.delete(oldest); - } - rrCounters.set(key, counter + 1); - const startIndex = counter % tiedTargets.length; - const orderedTiedTargets = [ - ...tiedTargets.slice(startIndex), - ...tiedTargets.slice(0, startIndex), - ]; - const tiedExecutionKeys = new Set(orderedTiedTargets.map((entry) => entry.target.executionKey)); - - return [ - ...orderedTiedTargets, - ...scoredTargets.filter((entry) => !tiedExecutionKeys.has(entry.target.executionKey)), - ].map((entry) => entry.target); -} - /** * Handle combo chat with fallback. * @param {Object} options diff --git a/open-sse/services/combo/quotaScoring.ts b/open-sse/services/combo/quotaScoring.ts new file mode 100644 index 0000000000..8f9dbbca92 --- /dev/null +++ b/open-sse/services/combo/quotaScoring.ts @@ -0,0 +1,311 @@ +/** + * Pure scoring + window-math helpers for the reset-aware / reset-window combo + * strategies. No module state, no async, no I/O — extracted byte-identically + * from combo.ts (QG v2 Fase 9 T5 D7b) as the pure half of the reset-aware quota + * block. The stateful/async half (cache Maps + connection/quota fetchers + + * orderTargets*) lives in ./quotaStrategies.ts, which imports the scoring + * helpers from here. + * + * Pure leaf: this module never imports from the combo barrel. + */ + +import { clamp01 } from "../../utils/number.ts"; +import { isRecord } from "./comboData.ts"; +import type { SlaRoutingPolicy } from "../autoCombo/routerStrategy.ts"; +import { RESET_WINDOW_NAMES } from "./types.ts"; +import type { ResolvedComboTarget } from "./types.ts"; + +const RESET_AWARE_SESSION_WINDOW_MS = 5 * 60 * 60 * 1000; +const RESET_AWARE_WEEKLY_WINDOW_MS = 7 * 24 * 60 * 60 * 1000; +const RESET_AWARE_SESSION_REMAINING_WEIGHT = 0.45; +const RESET_AWARE_SESSION_RESET_PRESSURE_WEIGHT = 0.55; +const RESET_AWARE_WEEKLY_REMAINING_WEIGHT = 0.25; +const RESET_AWARE_WEEKLY_RESET_PRESSURE_WEIGHT = 0.75; +const RESET_AWARE_DEFAULTS = { + sessionWeight: 0.35, + weeklyWeight: 0.65, + tieBandPercent: 5, + exhaustionGuardPercent: 10, +}; +const RESET_WINDOW_DEFAULT_TIE_BAND_MS = 60_000; + +type ResetWindowName = (typeof RESET_WINDOW_NAMES)[number]; +export type QuotaFetchCacheConfig = { + quotaCacheTtlMs: number; + quotaCacheMaxStaleMs: number; +}; +export type ResetWindowConfig = ReturnType; + +function finiteNumberOrNull(value: unknown): number | null { + const numericValue = Number(value); + return Number.isFinite(numericValue) ? numericValue : null; +} + +function getPercentConfig(value: unknown, fallback: number): number { + const numericValue = finiteNumberOrNull(value); + if (numericValue === null) return fallback; + return Math.max(0, Math.min(100, numericValue)); +} + +function getWeightConfig(value: unknown, fallback: number): number { + const numericValue = finiteNumberOrNull(value); + if (numericValue === null || numericValue < 0) return fallback; + return numericValue; +} + +function getDurationConfig(value: unknown, fallback: number, max: number): number { + const numericValue = finiteNumberOrNull(value); + if (numericValue === null || numericValue < 0) return fallback; + return Math.min(max, Math.floor(numericValue)); +} + +export function resolveResetAwareConfig(config: Record | null | undefined) { + const sessionWeight = getWeightConfig( + config?.resetAwareSessionWeight, + RESET_AWARE_DEFAULTS.sessionWeight + ); + const weeklyWeight = getWeightConfig( + config?.resetAwareWeeklyWeight, + RESET_AWARE_DEFAULTS.weeklyWeight + ); + const totalWeight = sessionWeight + weeklyWeight; + const normalizedSessionWeight = + totalWeight > 0 ? sessionWeight / totalWeight : RESET_AWARE_DEFAULTS.sessionWeight; + + return { + sessionWeight: normalizedSessionWeight, + weeklyWeight: 1 - normalizedSessionWeight, + tieBand: + getPercentConfig(config?.resetAwareTieBandPercent, RESET_AWARE_DEFAULTS.tieBandPercent) / 100, + exhaustionGuard: + getPercentConfig( + config?.resetAwareExhaustionGuardPercent, + RESET_AWARE_DEFAULTS.exhaustionGuardPercent + ) / 100, + quotaCacheTtlMs: getDurationConfig(config?.resetAwareQuotaCacheTtlMs, 0, 300_000), + quotaCacheMaxStaleMs: getDurationConfig(config?.resetAwareQuotaCacheMaxStaleMs, 0, 3_600_000), + }; +} + +export function resolveResetWindowConfig(config: Record | null | undefined) { + const rawWindows = Array.isArray(config?.resetWindowWindows) ? config.resetWindowWindows : null; + const windows = rawWindows + ?.filter((windowName): windowName is ResetWindowName => + (RESET_WINDOW_NAMES as readonly string[]).includes(String(windowName)) + ) + .filter((windowName, index, array) => array.indexOf(windowName) === index); + + const effectiveWindows = + windows && windows.length > 0 + ? windows + : config?.resetWindowIncludeSession === true + ? (["weekly", "session"] as ResetWindowName[]) + : (["weekly"] as ResetWindowName[]); + + return { + windows: effectiveWindows, + tieBandMs: Math.max( + 0, + finiteNumberOrNull(config?.resetWindowTieBandMs) ?? RESET_WINDOW_DEFAULT_TIE_BAND_MS + ), + quotaCacheTtlMs: getDurationConfig(config?.resetWindowQuotaCacheTtlMs, 0, 300_000), + quotaCacheMaxStaleMs: getDurationConfig(config?.resetWindowQuotaCacheMaxStaleMs, 0, 3_600_000), + }; +} + +export function resolveSlaRoutingPolicy( + config: Record | null | undefined +): SlaRoutingPolicy | undefined { + if (!config) return undefined; + const nestedSla = isRecord(config.sla) ? config.sla : {}; + const targetP95Ms = finiteNumberOrNull(config.slaTargetP95Ms ?? nestedSla.targetP95Ms); + const maxErrorRate = finiteNumberOrNull(config.slaMaxErrorRate ?? nestedSla.maxErrorRate); + const maxCostPer1MTokens = finiteNumberOrNull( + config.slaMaxCostPer1MTokens ?? nestedSla.maxCostPer1MTokens + ); + const hardConstraints = config.slaHardConstraints ?? nestedSla.hardConstraints; + + const policy: SlaRoutingPolicy = {}; + if (targetP95Ms !== null && targetP95Ms > 0) policy.targetP95Ms = targetP95Ms; + if (maxErrorRate !== null && maxErrorRate >= 0) policy.maxErrorRate = clamp01(maxErrorRate); + if (maxCostPer1MTokens !== null && maxCostPer1MTokens > 0) { + policy.maxCostPer1MTokens = maxCostPer1MTokens; + } + if (typeof hardConstraints === "boolean") policy.hardConstraints = hardConstraints; + + return Object.keys(policy).length > 0 ? policy : undefined; +} + +export function getResetAwareProvider(target: ResolvedComboTarget): string | null { + const provider = (target.providerId || target.provider || "").toLowerCase(); + return provider || null; +} + +function normalizeResetAt(value: unknown): string | null { + if (typeof value === "string" && value.trim().length > 0) return value.trim(); + if (typeof value === "number" && Number.isFinite(value)) return String(value); + return null; +} + +function parseResetTimeMs(resetAt: string | null | undefined): number { + if (!resetAt) return NaN; + const resetTime = Date.parse(resetAt); + if (Number.isFinite(resetTime)) return resetTime; + + if (!/^\d+(?:\.\d+)?$/.test(resetAt)) return NaN; + const numericResetAt = Number(resetAt); + if (!Number.isFinite(numericResetAt)) return NaN; + return numericResetAt < 10_000_000_000 ? numericResetAt * 1000 : numericResetAt; +} + +function getQuotaWindow( + quota: unknown, + key: "window5h" | "window7d" | "windowWeekly" | "windowMonthly" +): { percentUsed: number | null; resetAt: string | null } | null { + if (!isRecord(quota)) return null; + const window = quota[key]; + if (!isRecord(window)) return null; + const percentUsed = finiteNumberOrNull(window.percentUsed); + const resetAt = normalizeResetAt(window.resetAt); + return { percentUsed, resetAt }; +} + +function normalizeWindowPercentUsed(value: unknown): number | null { + const numericValue = finiteNumberOrNull(value); + if (numericValue === null) return null; + if (numericValue > 1) return clamp01(numericValue / 100); + return clamp01(numericValue); +} + +function getNamedQuotaWindow( + quota: unknown, + windowName: ResetWindowName +): { percentUsed: number | null; resetAt: string | null } | null { + if (!quota || !isRecord(quota)) return null; + + if (windowName === "session") return getQuotaWindow(quota, "window5h"); + if (windowName === "weekly") { + return getQuotaWindow(quota, "window7d") || getQuotaWindow(quota, "windowWeekly"); + } + if (windowName === "monthly") return getQuotaWindow(quota, "windowMonthly"); + + return null; +} + +function getWindowsMapQuotaWindow( + quota: unknown, + windowName: ResetWindowName +): { percentUsed: number | null; resetAt: string | null } | null { + if (!quota || !isRecord(quota) || !isRecord(quota.windows)) return null; + const candidates = Object.entries(quota.windows) + .map(([key, value]) => ({ key: key.toLowerCase(), value })) + .filter(({ key }) => key === windowName || key.startsWith(`${windowName} `)); + + if (candidates.length === 0) return null; + candidates.sort((a, b) => a.key.localeCompare(b.key)); + const window = candidates[0].value; + if (!isRecord(window)) return null; + + return { + percentUsed: normalizeWindowPercentUsed(window.percentUsed), + resetAt: normalizeResetAt(window.resetAt), + }; +} + +function resolveQuotaWindowByName( + quota: unknown, + windowName: ResetWindowName +): { percentUsed: number | null; resetAt: string | null } | null { + return getNamedQuotaWindow(quota, windowName) || getWindowsMapQuotaWindow(quota, windowName); +} + +function getResetUrgency(resetAt: string | null | undefined, windowMs: number): number { + if (!resetAt) return 0.5; + const resetTime = parseResetTimeMs(resetAt); + if (!Number.isFinite(resetTime)) return 0.5; + const msUntilReset = resetTime - Date.now(); + if (msUntilReset <= 0) return 1; + return clamp01(1 - msUntilReset / windowMs); +} + +function scoreQuotaWindow( + remaining: number, + resetAt: string | null | undefined, + windowMs: number, + remainingWeight: number, + resetPressureWeight: number +): number { + const normalizedRemaining = clamp01(remaining); + const resetUrgency = getResetUrgency(resetAt, windowMs); + const resetPressure = resetUrgency * (1 - normalizedRemaining); + return remainingWeight * normalizedRemaining + resetPressureWeight * resetPressure; +} + +export function scoreResetAwareQuota( + quota: unknown, + config: ReturnType +) { + if (!quota || !isRecord(quota)) return { score: 0.5 }; + if (quota.limitReached === true) return { score: -Infinity }; + + const overallPercentUsed = clamp01(finiteNumberOrNull(quota.percentUsed) ?? 0.5); + const sessionWindow = getQuotaWindow(quota, "window5h"); + const weeklyWindow = getQuotaWindow(quota, "window7d") || getQuotaWindow(quota, "windowWeekly"); + const sessionRemaining = clamp01(1 - (sessionWindow?.percentUsed ?? overallPercentUsed)); + const weeklyRemaining = clamp01(1 - (weeklyWindow?.percentUsed ?? overallPercentUsed)); + const sessionScore = scoreQuotaWindow( + sessionRemaining, + sessionWindow?.resetAt, + RESET_AWARE_SESSION_WINDOW_MS, + RESET_AWARE_SESSION_REMAINING_WEIGHT, + RESET_AWARE_SESSION_RESET_PRESSURE_WEIGHT + ); + const weeklyScore = scoreQuotaWindow( + weeklyRemaining, + weeklyWindow?.resetAt ?? normalizeResetAt(quota.resetAt), + RESET_AWARE_WEEKLY_WINDOW_MS, + RESET_AWARE_WEEKLY_REMAINING_WEIGHT, + RESET_AWARE_WEEKLY_RESET_PRESSURE_WEIGHT + ); + let score = config.sessionWeight * sessionScore + config.weeklyWeight * weeklyScore; + + if (config.exhaustionGuard > 0 && sessionRemaining < config.exhaustionGuard) { + score *= Math.max(0.05, sessionRemaining / config.exhaustionGuard); + } + + return { score }; +} + +export function getResetWindowTimestampMs(quota: unknown, windows: ResetWindowName[]): number { + if (!quota || !isRecord(quota) || quota.limitReached === true) return Infinity; + + let selectedResetMs = Infinity; + for (const windowName of windows) { + const window = resolveQuotaWindowByName(quota, windowName); + const resetMs = parseResetTimeMs(window?.resetAt ?? null); + if (Number.isFinite(resetMs)) { + selectedResetMs = Math.min(selectedResetMs, resetMs); + } + } + + if (!Number.isFinite(selectedResetMs)) { + selectedResetMs = parseResetTimeMs(normalizeResetAt(quota.resetAt)); + } + + return Number.isFinite(selectedResetMs) ? selectedResetMs : Infinity; +} + +function getResetWindowHorizonMs(windows: ResetWindowName[]): number { + if (windows.includes("monthly")) return 30 * 24 * 60 * 60 * 1000; + if (windows.includes("weekly")) return RESET_AWARE_WEEKLY_WINDOW_MS; + return RESET_AWARE_SESSION_WINDOW_MS; +} + +export function calculateResetWindowAffinity(quota: unknown, config: ResetWindowConfig): number { + const resetMs = getResetWindowTimestampMs(quota, config.windows); + if (!Number.isFinite(resetMs)) return 0.5; + + const msUntilReset = resetMs - Date.now(); + if (msUntilReset <= 0) return 1; + return clamp01(1 - msUntilReset / getResetWindowHorizonMs(config.windows)); +} diff --git a/open-sse/services/combo/quotaStrategies.ts b/open-sse/services/combo/quotaStrategies.ts new file mode 100644 index 0000000000..48d466a162 --- /dev/null +++ b/open-sse/services/combo/quotaStrategies.ts @@ -0,0 +1,568 @@ +/** + * Stateful + async reset-aware / reset-window quota strategies for combo routing. + * + * Holds the two mutable module-level caches that back reset-aware routing + * (`resetAwareConnectionCache` for per-provider active connections and + * `resetAwareQuotaCache` for per-connection quota snapshots), plus the helpers + * that read/write them and the strategy orderers. Extracted byte-identically + * from combo.ts (QG v2 Fase 9 T5 D7b) — the larger, stateful half of the + * reset-aware quota block. The pure scoring/window-math half lives in + * ./quotaScoring.ts and is imported here. + * + * State cohesion: `resetAwareConnectionCache`, `resetAwareQuotaCache`, and + * `MAX_RESET_AWARE_CACHE` MUST remain single instances defined once here, + * alongside their only readers/writers (getQuotaAwareConnectionsForTarget, + * fetchResetAwareQuotaWithCache) — never duplicate a Map. + * + * Cross-module state: the tie-band round-robin in orderTargetsByResetAwareQuota + * and orderTargetsByResetWindow shares the same rrCounters Map from ./rrState.ts + * (D7a) so reset-aware tie rotation stays consistent with round-robin routing. + * + * Pure leaf: this module never imports from the combo barrel. + */ + +import { getRuntimeProviderProfile, type ProviderProfile } from "../accountFallback.ts"; +import { PRE_SCREEN_CONCURRENCY } from "../comboConfig.ts"; +import { getQuotaFetcher } from "../quotaPreflight.ts"; +import { getCircuitBreaker } from "../../../src/shared/utils/circuitBreaker"; +import { getProviderConnections } from "../../../src/lib/db/providers"; +import { MAX_RR_COUNTERS, rrCounters } from "./rrState.ts"; +import type { ResolvedComboTarget, IsModelAvailable } from "./types.ts"; +import { + resolveResetAwareConfig, + resolveResetWindowConfig, + getResetAwareProvider, + scoreResetAwareQuota, + getResetWindowTimestampMs, + type QuotaFetchCacheConfig, +} from "./quotaScoring.ts"; + +const RESET_AWARE_CONNECTION_CACHE_TTL_MS = 30_000; +const RESET_AWARE_QUOTA_FETCH_CONCURRENCY = 5; + +const MAX_RESET_AWARE_CACHE = 200; + +const resetAwareConnectionCache = new Map< + string, + { fetchedAt: number; connections: Array> } +>(); +const resetAwareQuotaCache = new Map< + string, + { fetchedAt: number; quota: unknown; refreshPromise: Promise | null } +>(); + +async function getQuotaAwareConnectionsForTarget( + target: ResolvedComboTarget, + connectionCache: Map>>, + connectionLoadPromises: Map>>>, + comboName: string, + log: { warn?: (...args: unknown[]) => void } +) { + const provider = getResetAwareProvider(target); + if (!provider || !getQuotaFetcher(provider)) return []; + if (!connectionCache.has(provider)) { + const cached = resetAwareConnectionCache.get(provider); + if (cached && Date.now() - cached.fetchedAt < RESET_AWARE_CONNECTION_CACHE_TTL_MS) { + connectionCache.set(provider, cached.connections); + return cached.connections; + } + + if (!connectionLoadPromises.has(provider)) { + connectionLoadPromises.set( + provider, + (async () => { + try { + const connections = await getProviderConnections({ provider, isActive: true }); + const activeConnections = Array.isArray(connections) + ? (connections as Array>) + : []; + if ( + !resetAwareConnectionCache.has(provider) && + resetAwareConnectionCache.size >= MAX_RESET_AWARE_CACHE + ) { + const oldest = resetAwareConnectionCache.keys().next().value; + if (oldest !== undefined) resetAwareConnectionCache.delete(oldest); + } + resetAwareConnectionCache.set(provider, { + connections: activeConnections, + fetchedAt: Date.now(), + }); + return activeConnections; + } catch (error) { + log.warn?.("COMBO", "Reset-aware failed to load quota-aware connections.", { + comboName, + err: error, + operation: "getProviderConnections", + provider, + }); + return []; + } + })() + ); + } + + const connections = await connectionLoadPromises.get(provider)!; + connectionCache.set(provider, connections); + } + return connectionCache.get(provider) || []; +} + +function normalizeConnectionIds(value: unknown): string[] | null { + if (!Array.isArray(value)) return null; + const ids = value.filter( + (connectionId): connectionId is string => + typeof connectionId === "string" && connectionId.trim().length > 0 + ); + return ids.length > 0 ? ids : null; +} + +function filterAllowedConnectionIds( + connectionIds: string[], + apiKeyAllowedConnectionIds: string[] | null | undefined +): string[] { + const allowedIds = normalizeConnectionIds(apiKeyAllowedConnectionIds); + if (!allowedIds) return connectionIds; + const allowedSet = new Set(allowedIds); + return connectionIds.filter((connectionId) => allowedSet.has(connectionId)); +} + +function getTargetConnectionIds( + target: ResolvedComboTarget, + connections: Array> +): string[] { + let connectionIds: string[]; + if (target.connectionId) { + return [target.connectionId]; + } + + if (Array.isArray(target.allowedConnectionIds) && target.allowedConnectionIds.length > 0) { + return target.allowedConnectionIds.filter( + (connectionId): connectionId is string => + typeof connectionId === "string" && connectionId.trim().length > 0 + ); + } + + connectionIds = connections + .map((connection) => (typeof connection.id === "string" ? connection.id : null)) + .filter((connectionId): connectionId is string => !!connectionId); + return connectionIds; +} + +async function mapWithConcurrency( + items: T[], + concurrency: number, + mapper: (item: T, index: number) => Promise +): Promise { + const results = new Array(items.length); + let nextIndex = 0; + const workerCount = Math.max(1, Math.min(concurrency, items.length)); + + await Promise.all( + Array.from({ length: workerCount }, async () => { + while (nextIndex < items.length) { + const currentIndex = nextIndex++; + results[currentIndex] = await mapper(items[currentIndex], currentIndex); + } + }) + ); + + return results; +} + +export async function fetchResetAwareQuotaWithCache({ + provider, + connectionId, + connection, + fetcher, + config, + log, + comboName, +}: { + provider: string; + connectionId: string; + connection?: Record; + fetcher: (connectionId: string, connection?: Record) => Promise; + config: QuotaFetchCacheConfig; + log: { debug?: (...args: unknown[]) => void; warn?: (...args: unknown[]) => void }; + comboName: string; +}): Promise { + const cacheKey = `${provider}:${connectionId}`; + const ttlMs = config.quotaCacheTtlMs; + const maxStaleMs = config.quotaCacheMaxStaleMs; + const now = Date.now(); + const cached = resetAwareQuotaCache.get(cacheKey); + + if (ttlMs <= 0 && maxStaleMs <= 0) { + try { + return await fetcher(connectionId, connection); + } catch (error) { + log.warn?.("COMBO", "Reset-aware quota fetch failed.", { + comboName, + connectionId, + err: error, + operation: "quotaFetch", + provider, + }); + return null; + } + } + + const refresh = () => { + const existing = resetAwareQuotaCache.get(cacheKey); + if (existing?.refreshPromise != null) return existing.refreshPromise; + + const refreshPromise = fetcher(connectionId, connection) + .then((quota) => { + if (quota) { + if ( + !resetAwareQuotaCache.has(cacheKey) && + resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE + ) { + const oldest = resetAwareQuotaCache.keys().next().value; + if (oldest !== undefined) resetAwareQuotaCache.delete(oldest); + } + resetAwareQuotaCache.set(cacheKey, { + quota, + fetchedAt: Date.now(), + refreshPromise: null, + }); + } else { + resetAwareQuotaCache.delete(cacheKey); + } + return quota; + }) + .catch((error) => { + const previous = resetAwareQuotaCache.get(cacheKey); + if (previous) { + if ( + !resetAwareQuotaCache.has(cacheKey) && + resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE + ) { + const oldest = resetAwareQuotaCache.keys().next().value; + if (oldest !== undefined) resetAwareQuotaCache.delete(oldest); + } + resetAwareQuotaCache.set(cacheKey, { ...previous, refreshPromise: null }); + } + log.warn?.("COMBO", "Reset-aware quota fetch failed.", { + comboName, + connectionId, + err: error, + operation: "quotaFetch", + provider, + }); + return null; + }); + + if (!resetAwareQuotaCache.has(cacheKey) && resetAwareQuotaCache.size >= MAX_RESET_AWARE_CACHE) { + const oldest = resetAwareQuotaCache.keys().next().value; + if (oldest !== undefined) resetAwareQuotaCache.delete(oldest); + } + resetAwareQuotaCache.set(cacheKey, { + quota: existing?.quota ?? cached?.quota ?? null, + fetchedAt: existing?.fetchedAt ?? cached?.fetchedAt ?? 0, + refreshPromise, + }); + return refreshPromise; + }; + + if (ttlMs > 0 && cached) { + const age = now - cached.fetchedAt; + if (age <= ttlMs) return cached.quota; + if (maxStaleMs > 0 && age <= ttlMs + maxStaleMs) { + void refresh(); + return cached.quota; + } + } + + return refresh(); +} + +export type PreScreenResult = { profile: ProviderProfile | null; available: boolean }; + +export async function preScreenTargets( + targets: ResolvedComboTarget[], + isModelAvailable?: IsModelAvailable | null +): Promise> { + if (targets.length === 0) { + return new Map(); + } + + const results = await mapWithConcurrency( + targets, + PRE_SCREEN_CONCURRENCY, + async (target): Promise<{ key: string; result: PreScreenResult }> => { + const profile = await getRuntimeProviderProfile(target.provider).catch(() => null); + + const breaker = getCircuitBreaker(target.provider); + if (breaker.getStatus().state === "OPEN") { + return { key: target.executionKey, result: { profile, available: false } }; + } + + let available = true; + if (isModelAvailable) { + // IsModelAvailable may return a sync boolean or a Promise; Promise.resolve + // normalizes both so the .catch() never runs against a bare boolean. + available = await Promise.resolve(isModelAvailable(target.modelStr, target)).catch( + () => true + ); + } + return { key: target.executionKey, result: { profile, available } }; + } + ); + + const map = new Map(); + for (const { key, result } of results) { + map.set(key, result); + } + return map; +} + +export async function orderTargetsByResetAwareQuota( + targets: ResolvedComboTarget[], + comboName: string, + configSource: Record | null | undefined, + log: { warn?: (...args: unknown[]) => void }, + apiKeyAllowedConnectionIds?: string[] | null +) { + if (targets.length === 0) return targets; + + const config = resolveResetAwareConfig(configSource); + const connectionCache = new Map>>(); + const connectionLoadPromises = new Map>>>(); + const quotaPromises = new Map>(); + const connectionById = new Map>(); + const expandedTargets: ResolvedComboTarget[] = []; + + const targetsWithConnections = await Promise.all( + targets.map(async (target) => ({ + connections: await getQuotaAwareConnectionsForTarget( + target, + connectionCache, + connectionLoadPromises, + comboName, + log + ), + target, + })) + ); + + for (const { target, connections } of targetsWithConnections) { + for (const connection of connections) { + if (typeof connection.id === "string") connectionById.set(connection.id, connection); + } + + const unrestrictedConnectionIds = getTargetConnectionIds(target, connections); + const connectionIds = filterAllowedConnectionIds( + unrestrictedConnectionIds, + apiKeyAllowedConnectionIds + ); + if (connectionIds.length === 0) { + if ( + unrestrictedConnectionIds.length > 0 && + normalizeConnectionIds(apiKeyAllowedConnectionIds) + ) { + continue; + } + expandedTargets.push(target); + continue; + } + + for (const connectionId of connectionIds) { + expandedTargets.push({ + ...target, + connectionId, + executionKey: + target.connectionId === connectionId + ? target.executionKey + : `${target.executionKey}@${connectionId}`, + }); + } + } + + const scoredTargets = await mapWithConcurrency( + expandedTargets, + RESET_AWARE_QUOTA_FETCH_CONCURRENCY, + async (target, index) => { + let quota: unknown = null; + const provider = getResetAwareProvider(target); + const fetcher = provider ? getQuotaFetcher(provider) : null; + if (fetcher && provider && target.connectionId) { + const quotaKey = `${provider}:${target.connectionId}`; + if (!quotaPromises.has(quotaKey)) { + quotaPromises.set( + quotaKey, + fetchResetAwareQuotaWithCache({ + provider, + connectionId: target.connectionId, + connection: connectionById.get(target.connectionId), + fetcher, + config, + log, + comboName, + }) + ); + } + quota = await quotaPromises.get(quotaKey)!; + } + const { score } = scoreResetAwareQuota(quota, config); + return { target, score, index }; + } + ); + + scoredTargets.sort((a, b) => { + if (b.score !== a.score) return b.score - a.score; + return a.index - b.index; + }); + + const bestScore = scoredTargets[0]?.score ?? 0; + const tiedTargets = scoredTargets.filter((entry) => bestScore - entry.score <= config.tieBand); + let orderedTiedTargets = tiedTargets; + if (tiedTargets.length > 1) { + const key = `reset-aware:${comboName}`; + const counter = rrCounters.get(key) || 0; + if (!rrCounters.has(key) && rrCounters.size >= MAX_RR_COUNTERS) { + const oldest = rrCounters.keys().next().value; + if (oldest !== undefined) rrCounters.delete(oldest); + } + rrCounters.set(key, counter + 1); + const startIndex = counter % tiedTargets.length; + orderedTiedTargets = [...tiedTargets.slice(startIndex), ...tiedTargets.slice(0, startIndex)]; + } + + const tiedExecutionKeys = new Set(orderedTiedTargets.map((entry) => entry.target.executionKey)); + return [ + ...orderedTiedTargets, + ...scoredTargets.filter((entry) => !tiedExecutionKeys.has(entry.target.executionKey)), + ].map((entry) => entry.target); +} + +export async function orderTargetsByResetWindow( + targets: ResolvedComboTarget[], + comboName: string, + configSource: Record | null | undefined, + log: { warn?: (...args: unknown[]) => void }, + apiKeyAllowedConnectionIds?: string[] | null +) { + if (targets.length === 0) return targets; + + const config = resolveResetWindowConfig(configSource); + const connectionCache = new Map>>(); + const connectionLoadPromises = new Map>>>(); + const quotaPromises = new Map>(); + const connectionById = new Map>(); + const expandedTargets: ResolvedComboTarget[] = []; + + const targetsWithConnections = await Promise.all( + targets.map(async (target) => ({ + connections: await getQuotaAwareConnectionsForTarget( + target, + connectionCache, + connectionLoadPromises, + comboName, + log + ), + target, + })) + ); + + for (const { target, connections } of targetsWithConnections) { + for (const connection of connections) { + if (typeof connection.id === "string") connectionById.set(connection.id, connection); + } + + const unrestrictedConnectionIds = getTargetConnectionIds(target, connections); + const connectionIds = filterAllowedConnectionIds( + unrestrictedConnectionIds, + apiKeyAllowedConnectionIds + ); + if (connectionIds.length === 0) { + if ( + unrestrictedConnectionIds.length > 0 && + normalizeConnectionIds(apiKeyAllowedConnectionIds) + ) { + continue; + } + expandedTargets.push(target); + continue; + } + + for (const connectionId of connectionIds) { + expandedTargets.push({ + ...target, + connectionId, + executionKey: + target.connectionId === connectionId + ? target.executionKey + : `${target.executionKey}@${connectionId}`, + }); + } + } + + const scoredTargets = await mapWithConcurrency( + expandedTargets, + RESET_AWARE_QUOTA_FETCH_CONCURRENCY, + async (target, index) => { + let quota: unknown = null; + const provider = getResetAwareProvider(target); + const fetcher = provider ? getQuotaFetcher(provider) : null; + if (fetcher && provider && target.connectionId) { + const quotaKey = `${provider}:${target.connectionId}`; + if (!quotaPromises.has(quotaKey)) { + quotaPromises.set( + quotaKey, + fetchResetAwareQuotaWithCache({ + provider, + connectionId: target.connectionId, + connection: connectionById.get(target.connectionId), + fetcher, + config, + log, + comboName, + }) + ); + } + quota = await quotaPromises.get(quotaKey)!; + } + + return { + target, + resetMs: getResetWindowTimestampMs(quota, config.windows), + index, + }; + } + ); + + scoredTargets.sort((a, b) => { + if (a.resetMs !== b.resetMs) return a.resetMs - b.resetMs; + return a.index - b.index; + }); + + const bestResetMs = scoredTargets[0]?.resetMs ?? Infinity; + if (!Number.isFinite(bestResetMs) || config.tieBandMs <= 0) { + return scoredTargets.map((entry) => entry.target); + } + + const tiedTargets = scoredTargets.filter( + (entry) => entry.resetMs - bestResetMs <= config.tieBandMs + ); + if (tiedTargets.length <= 1) return scoredTargets.map((entry) => entry.target); + + const key = `reset-window:${comboName}`; + const counter = rrCounters.get(key) || 0; + if (!rrCounters.has(key) && rrCounters.size >= MAX_RR_COUNTERS) { + const oldest = rrCounters.keys().next().value; + if (oldest !== undefined) rrCounters.delete(oldest); + } + rrCounters.set(key, counter + 1); + const startIndex = counter % tiedTargets.length; + const orderedTiedTargets = [ + ...tiedTargets.slice(startIndex), + ...tiedTargets.slice(0, startIndex), + ]; + const tiedExecutionKeys = new Set(orderedTiedTargets.map((entry) => entry.target.executionKey)); + + return [ + ...orderedTiedTargets, + ...scoredTargets.filter((entry) => !tiedExecutionKeys.has(entry.target.executionKey)), + ].map((entry) => entry.target); +} From b05bc7469e39a604cdbb29b16f65c8bde004c05e Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Thu, 18 Jun 2026 19:12:17 -0300 Subject: [PATCH 2/4] fix(compression): harden RTK raw-output redaction + ReDoS guard for custom filters (F5.3) (#4203) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(compression): broaden RTK raw-output secret redaction (Basic/Proxy auth + keys) F5.3 follow-up. SECRET_PATTERNS missed Authorization / Proxy-Authorization with Basic (curl -v emits "Basic ") and credential field names that don't contain token/secret/password as a substring (private_key, access_key, credential). Add a (Proxy-)Authorization Bearer|Basic pattern and broaden the key-name alternation. All patterns are flat (no nested quantifiers → no ReDoS). * fix(compression): drop ReDoS-prone patterns from custom RTK filters F5.3 follow-up. Custom filters (DATA_DIR/rtk/filters.json) carry user-supplied regex strings compiled and run against untrusted tool output; a nested unbounded quantifier ((a+)+, (a*)*, ([a-z]+)+ …) can cause catastrophic backtracking. validateRtkFilter now drops such patterns via a conservative, dependency-free heuristic (isReDoSProne) applied to every regex-bearing array in the canonical and legacy filter shapes. safe-regex would be ideal but is not installable in this symlinked worktree; the heuristic catches the common single-group nested-quantifier shapes and is itself linear. Surfaced (not fixed here): a canonical pack filter omitting `preserve` crashes validateRtkFilter because rtkFilterPreserveSchema.default({}) leaves the inner arrays undefined — tracked as a pre-existing fragility. --- .../compression/engines/rtk/filterSchema.ts | 41 ++++++++++---- .../compression/engines/rtk/rawOutput.ts | 10 +++- .../rtk-filter-redos-guard.test.ts | 54 +++++++++++++++++++ tests/unit/compression/rtk-raw-output.test.ts | 18 +++++++ 4 files changed, 112 insertions(+), 11 deletions(-) create mode 100644 tests/unit/compression/rtk-filter-redos-guard.test.ts diff --git a/open-sse/services/compression/engines/rtk/filterSchema.ts b/open-sse/services/compression/engines/rtk/filterSchema.ts index b3d2842b45..b4bfa70774 100644 --- a/open-sse/services/compression/engines/rtk/filterSchema.ts +++ b/open-sse/services/compression/engines/rtk/filterSchema.ts @@ -142,14 +142,37 @@ function isCanonicalFilter(value: z.infer): value is Rtk return "label" in value && "match" in value && "rules" in value; } +/** + * Conservative, dependency-free ReDoS guard. Custom RTK filters (DATA_DIR/rtk/filters.json) + * carry user-supplied regex strings that are compiled and run against untrusted tool output; + * a nested unbounded quantifier ((a+)+, (a*)*, ([a-z]+)+, (a+|b)+ …) causes catastrophic + * backtracking. This flags the common single-group nested-quantifier shapes so the loader + * never compiles them. Heuristic by design — a full analysis would use `safe-regex` (not + * installable in this symlinked worktree); it is itself linear (no nested quantifier). + */ +export function isReDoSProne(pattern: string): boolean { + return /\([^()]*(?:[+*]|\{\d+,\})[^()]*\)\s*(?:[+*]|\{\d+,\})/.test(pattern); +} + +function dropReDoSProne(patterns: string[]): string[] { + return patterns.filter((p) => !isReDoSProne(p)); +} + export function validateRtkFilter(value: unknown): RtkFilterDefinition { const parsed = rtkFilterSchema.parse(value); if (!isCanonicalFilter(parsed)) { + const collapse = dropReDoSProne(parsed.collapsePatterns); return { ...parsed, + stripPatterns: dropReDoSProne(parsed.stripPatterns), + keepPatterns: dropReDoSProne(parsed.keepPatterns), + priorityPatterns: dropReDoSProne(parsed.priorityPatterns), + collapsePatterns: collapse, + replace: parsed.replace.filter((r) => !isReDoSProne(r.pattern)), + matchOutput: parsed.matchOutput.filter((r) => !isReDoSProne(r.pattern)), commandPatterns: [], matchPatterns: [], - deduplicate: parsed.collapsePatterns.length > 0, + deduplicate: collapse.length > 0, }; } @@ -159,17 +182,17 @@ export function validateRtkFilter(value: unknown): RtkFilterDefinition { name: parsed.label, description: parsed.description, commandTypes: parsed.match.outputTypes, - commandPatterns: parsed.match.commands, - matchPatterns: parsed.match.patterns, + commandPatterns: dropReDoSProne(parsed.match.commands), + matchPatterns: dropReDoSProne(parsed.match.patterns), category: parsed.category, priority: parsed.priority, - stripPatterns: parsed.rules.dropPatterns, - keepPatterns: parsed.rules.includePatterns, - priorityPatterns: preservePatterns, - collapsePatterns: parsed.rules.collapsePatterns, + stripPatterns: dropReDoSProne(parsed.rules.dropPatterns), + keepPatterns: dropReDoSProne(parsed.rules.includePatterns), + priorityPatterns: dropReDoSProne(preservePatterns), + collapsePatterns: dropReDoSProne(parsed.rules.collapsePatterns), stripAnsi: parsed.rules.stripAnsi, - replace: parsed.rules.replace, - matchOutput: parsed.rules.matchOutput, + replace: parsed.rules.replace.filter((r) => !isReDoSProne(r.pattern)), + matchOutput: parsed.rules.matchOutput.filter((r) => !isReDoSProne(r.pattern)), truncateLineAt: parsed.rules.truncateLineAt, onEmpty: parsed.rules.onEmpty, filterStderr: parsed.rules.filterStderr, diff --git a/open-sse/services/compression/engines/rtk/rawOutput.ts b/open-sse/services/compression/engines/rtk/rawOutput.ts index b536e08713..57c2e4da34 100644 --- a/open-sse/services/compression/engines/rtk/rawOutput.ts +++ b/open-sse/services/compression/engines/rtk/rawOutput.ts @@ -19,8 +19,14 @@ const SECRET_PATTERNS: Array<[RegExp, string]> = [ [/\b(sk-[A-Za-z0-9_-]{16,})\b/g, "[REDACTED_OPENAI_KEY]"], [/\b(xox[baprs]-[A-Za-z0-9-]{16,})\b/g, "[REDACTED_SLACK_TOKEN]"], [/\b(AKIA[0-9A-Z]{16})\b/g, "[REDACTED_AWS_KEY]"], - [/((?:api[_-]?key|token|secret|password)\s*[:=]\s*)("[^"]+"|'[^']+'|[^\s]+)/gi, "$1[REDACTED]"], - [/(Authorization:\s*Bearer\s+)[A-Za-z0-9._~+/-]+=*/gi, "$1[REDACTED]"], + // key=value / key: value for common credential field names (flat alternation — no nesting, + // so no ReDoS). Covers names the bare token/secret/password set misses (private_key, etc). + [ + /((?:api[_-]?key|api[_-]?token|access[_-]?key|access[_-]?token|client[_-]?secret|auth[_-]?token|private[_-]?key|secret[_-]?key|credentials?|token|secret|password)\s*[:=]\s*)("[^"]+"|'[^']+'|[^\s]+)/gi, + "$1[REDACTED]", + ], + // Authorization / Proxy-Authorization with Bearer OR Basic (curl -v emits Basic ). + [/((?:Proxy-)?Authorization:\s*(?:Bearer|Basic)\s+)[A-Za-z0-9._~+/-]+=*/gi, "$1[REDACTED]"], ]; function dataDir(): string { diff --git a/tests/unit/compression/rtk-filter-redos-guard.test.ts b/tests/unit/compression/rtk-filter-redos-guard.test.ts new file mode 100644 index 0000000000..fa5463aedb --- /dev/null +++ b/tests/unit/compression/rtk-filter-redos-guard.test.ts @@ -0,0 +1,54 @@ +import { describe, it } from "node:test"; +import assert from "node:assert/strict"; + +import { + validateRtkFilter, + isReDoSProne, +} from "../../../open-sse/services/compression/engines/rtk/filterSchema.ts"; + +// Custom RTK filters (DATA_DIR/rtk/filters.json) carry user-supplied regex strings that are +// compiled and run against untrusted tool output. A nested unbounded quantifier ((a+)+, (a*)*) +// can trigger catastrophic backtracking (ReDoS). validateRtkFilter must drop such patterns so +// they are never compiled. Dependency-free heuristic (safe-regex not installable here). +describe("RTK filter ReDoS guard", () => { + it("flags nested unbounded quantifiers and accepts safe patterns", () => { + assert.equal(isReDoSProne("(a+)+"), true); + assert.equal(isReDoSProne("(a*)*"), true); + assert.equal(isReDoSProne("([a-z]+)+"), true); + assert.equal(isReDoSProne("(a+|b)+"), true); + assert.equal(isReDoSProne("(?:\\d+)+"), true); + + assert.equal(isReDoSProne("(ab)+"), false); + assert.equal(isReDoSProne("\\d{1,8}"), false); + assert.equal(isReDoSProne("error|fail|FAIL"), false); + assert.equal(isReDoSProne("^\\s*$"), false); + }); + + it("strips ReDoS-prone patterns from a canonical filter at validation", () => { + const def = validateRtkFilter({ + id: "x", + label: "X", + category: "generic", + match: { commands: [], patterns: ["(a+)+", "ERROR\\b"], outputTypes: [] }, + rules: { dropPatterns: ["(.*)*", "^\\s*$"] }, + // preserve provided explicitly: a pack filter omitting it crashes validateRtkFilter + // (pre-existing: rtkFilterPreserveSchema.default({}) leaves errorPatterns undefined). + preserve: { errorPatterns: [], summaryPatterns: [] }, + }); + + assert.deepEqual(def.matchPatterns, ["ERROR\\b"], "catastrophic matchPattern dropped"); + assert.deepEqual(def.stripPatterns, ["^\\s*$"], "catastrophic dropPattern removed"); + }); + + it("strips ReDoS-prone patterns from a legacy filter at validation", () => { + const def = validateRtkFilter({ + id: "y", + name: "Y", + category: "generic", + commandTypes: ["shell"], + stripPatterns: ["([a-z]+)*", "keepme"], + }); + + assert.deepEqual(def.stripPatterns, ["keepme"], "catastrophic legacy stripPattern removed"); + }); +}); diff --git a/tests/unit/compression/rtk-raw-output.test.ts b/tests/unit/compression/rtk-raw-output.test.ts index d4e739d224..7344eb6b30 100644 --- a/tests/unit/compression/rtk-raw-output.test.ts +++ b/tests/unit/compression/rtk-raw-output.test.ts @@ -146,4 +146,22 @@ describe("RTK raw output retention", () => { assert.equal(pointer, null); fs.rmSync(blocker, { force: true }); }); + + it("redacts Basic/Proxy auth headers and key fields the base patterns miss", () => { + const basic = redactRtkRawOutput("> Authorization: Basic dXNlcjpwYXNzd29yZA=="); + assert.equal(basic.redacted, true); + assert.ok(!basic.text.includes("dXNlcjpwYXNzd29yZA=="), "Authorization: Basic base64 redacted"); + + const proxy = redactRtkRawOutput("Proxy-Authorization: Basic c2VjcmV0OnBhc3M="); + assert.equal(proxy.redacted, true); + assert.ok(!proxy.text.includes("c2VjcmV0OnBhc3M="), "Proxy-Authorization redacted"); + + const pkey = redactRtkRawOutput("private_key=abc123def456ghi"); + assert.equal(pkey.redacted, true); + assert.ok(!pkey.text.includes("abc123def456ghi"), "private_key value redacted"); + + const cred = redactRtkRawOutput("credential: my-credential-value-x"); + assert.equal(cred.redacted, true); + assert.ok(!cred.text.includes("my-credential-value-x"), "credential value redacted"); + }); }); From 8e921cab4b08d771be6b923e4303d7ebd32a27e6 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Thu, 18 Jun 2026 19:12:22 -0300 Subject: [PATCH 3/4] chore(ci): re-add 8 combo split leaves to stryker mutate + expand nightly batch-matrix 3->5 (QG v2 Fase 9 T5 Fase 3) (#4205) --- .github/workflows/nightly-mutation.yml | 34 ++++++++++++-------- stryker.conf.json | 43 ++++++++++++++++++++------ 2 files changed, 54 insertions(+), 23 deletions(-) diff --git a/.github/workflows/nightly-mutation.yml b/.github/workflows/nightly-mutation.yml index 722129f0ab..19d2d3b7d0 100644 --- a/.github/workflows/nightly-mutation.yml +++ b/.github/workflows/nightly-mutation.yml @@ -13,17 +13,21 @@ jobs: runs-on: ubuntu-latest # Mutation testing is expensive. History of the budget: # - Full 8-module set TIMED OUT at the 180min cap (run 27705123780 = exactly 180min). - # The two god-files chatCore.ts/combo.ts dominate ~2/3 of the mutants and were - # removed from stryker.conf.json `mutate` (the Onda 3 split re-adds them). - # - The remaining 6 modules STILL timed out at 180min (run 27732877861 = exactly - # 180min): auth.ts and accountFallback.ts are large and the perTest dry-run over - # ~130 covering test files is itself costly. - # So the 6 modules are split into 3 PARALLEL batches of 2 (the two big modules in - # separate batches), each overriding the mutate set via `--mutate` (Stryker 9 CLI: + # The two god-files chatCore.ts/combo.ts dominated ~2/3 of the mutants and were + # removed from stryker.conf.json `mutate`. + # - The remaining 6 modules in 3 batches: auth.ts and accountFallback.ts are large; the + # a (auth+publicCreds) and b (accountFallback+error) batches still ran near the 180min + # cap (the perTest dry-run over ~130 covering test files is itself costly), so the two + # big modules are now ISOLATED into their own batches (a=auth, b=accountFallback). + # - Onda 3 / Fase 9 T5 re-add: combo.ts was split into 11 leaves; the 8 well-covered + # combo/* leaves are back in `mutate` (covered by the 24 combo-*.test.ts), grouped into + # 2 batches (d=heavy, e=light). chatCore/* leaves + quotaScoring/quotaStrategies (#4204) + # are follow-ups — see _mutate_godfiles_excluded_comment in stryker.conf.json. + # 5 PARALLEL batches, each overriding the mutate set via `--mutate` (Stryker 9 CLI: # `-m, --mutate `; the conf's `mutate[]` remains the local-run default/union). - # Each batch is ~1/3 of the mutants and fits 180min; full coverage every night, in - # parallel (≈180min wall-clock). Re-merge into fewer batches once the Onda 3 god-file - # split shrinks per-module mutant counts. + # Each batch targets <180min; full coverage every night in parallel (~180min wall-clock). + # The `incremental` cache (per-batch key) makes runs after the first cold one much cheaper, + # so the isolated big-module batches recover even if a first cold run nears the cap. # Runs at stryker concurrency=4 with per-process DATA_DIR isolation # (tests/_setup/isolateDataDir.ts) — see _concurrency_comment in stryker.conf.json. strategy: @@ -31,11 +35,15 @@ jobs: matrix: batch: - name: a - mutate: "src/sse/services/auth.ts,open-sse/utils/publicCreds.ts" + mutate: "src/sse/services/auth.ts" - name: b - mutate: "open-sse/services/accountFallback.ts,open-sse/utils/error.ts" + mutate: "open-sse/services/accountFallback.ts" - name: c - mutate: "src/server/authz/routeGuard.ts,src/shared/utils/circuitBreaker.ts" + mutate: "src/server/authz/routeGuard.ts,src/shared/utils/circuitBreaker.ts,open-sse/utils/error.ts,open-sse/utils/publicCreds.ts" + - name: d + mutate: "open-sse/services/combo/comboStructure.ts,open-sse/services/combo/autoStrategy.ts,open-sse/services/combo/validateQuality.ts" + - name: e + mutate: "open-sse/services/combo/shadowRouting.ts,open-sse/services/combo/targetSorters.ts,open-sse/services/combo/comboPredicates.ts,open-sse/services/combo/rrState.ts,open-sse/services/combo/comboData.ts" timeout-minutes: 180 steps: - uses: actions/checkout@v6 diff --git a/stryker.conf.json b/stryker.conf.json index eb588a90d8..427e818a8b 100644 --- a/stryker.conf.json +++ b/stryker.conf.json @@ -186,15 +186,30 @@ ] }, "_mutate_godfiles_excluded_comment": [ - "2026-06-18 (Onda 2 budget): chatCore.ts (5874 LOC) + combo.ts (5282 LOC) — the two", - "god-files — were REMOVED from `mutate`. They dominate ~2/3 of the ~15k mutants, and the", - "full 8-module run TIMED OUT at the 180min nightly cap (run 27705123780, concurrency=4 +", - "DATA_DIR isolation: started 16:47:33 -> killed 19:47:48 = exactly 180min; the prior 120min", - "scheduled run also timed out). #4078 made concurrency safe but the tap-runner re-spawns a", - "node process per test file PER MUTANT, so spawn cost dominates and 15k mutants does not fit.", - "The 6 smaller modules below fit the budget and produce REAL mutation scores now; the two", - "god-files regain mutation coverage after the Onda 3 split into smaller units (re-add them", - "here — ideally as the split sub-modules). See project memory: Quality Gate v2 / Fase 9." + "2026-06-18 (Onda 2 budget): chatCore.ts + combo.ts — the two god-files — were REMOVED", + "from `mutate`. They dominated ~2/3 of the ~15k mutants; the full 8-module run TIMED OUT", + "at the 180min nightly cap (run 27705123780: 16:47:33 -> killed 19:47:48 = exactly 180min;", + "the prior 120min scheduled run also timed out). #4078 made concurrency safe but the", + "tap-runner re-spawns a node process per test file PER MUTANT, so spawn cost dominates and", + "15k mutants does not fit.", + "", + "2026-06-18 (Onda 3 / Fase 9 T5 re-add): the combo.ts god-file was split into 11 small", + "leaf modules under open-sse/services/combo/ (PRs #4162/#4175/#4186/#4196/#4204). The", + "routing LOGIC that justified combo.ts being in `mutate` now lives in those leaves, so the", + "8 well-covered combo/* leaves are ADDED back to `mutate` here (comboStructure/autoStrategy/", + "validateQuality/shadowRouting/targetSorters/comboPredicates/rrState/comboData). They are", + "covered by the 24 combo-*.test.ts files already in tap.testFiles. types.ts is omitted", + "(pure type declarations produce 0 mutants).", + "", + "STILL EXCLUDED (follow-ups, NOT in `mutate` yet):", + " - combo.ts + chatCore.ts barrels: their handleComboChat/handleChatCore CORES were not", + " split (out of scope — Fase 3 ChatCoreContext refactor). The barrels are now thin-ish", + " but still large; keep excluded until the cores are split.", + " - combo/quotaScoring.ts + combo/quotaStrategies.ts: in PR #4204 (D7b), not yet merged.", + " Add them here once #4204 lands (they are covered by combo-prescreen/combo-config tests).", + " - chatCore/* leaves (15 modules): thinner test coverage (only ~7 chatcore-*.test.ts in", + " tap.testFiles); add after a covering-test audit so they don't all 'survive (no coverage)'.", + "See project memory: Quality Gate v2 / Fase 9 (project-combo-split)." ], "mutate": [ "open-sse/services/accountFallback.ts", @@ -202,7 +217,15 @@ "src/server/authz/routeGuard.ts", "open-sse/utils/error.ts", "open-sse/utils/publicCreds.ts", - "src/shared/utils/circuitBreaker.ts" + "src/shared/utils/circuitBreaker.ts", + "open-sse/services/combo/comboStructure.ts", + "open-sse/services/combo/autoStrategy.ts", + "open-sse/services/combo/validateQuality.ts", + "open-sse/services/combo/shadowRouting.ts", + "open-sse/services/combo/targetSorters.ts", + "open-sse/services/combo/comboPredicates.ts", + "open-sse/services/combo/rrState.ts", + "open-sse/services/combo/comboData.ts" ], "_ignorePatterns_comment": [ "ignorePatterns = files NOT copied into the Stryker sandbox. It does NOT scope", From aed2a5024c6fb9ccda81b0a87ee3a3f6c5a8cddd Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Thu, 18 Jun 2026 19:12:27 -0300 Subject: [PATCH 4/4] feat(compression): persist RTK grouping config (unlock R5 enableGrouping) (#4207) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The RTK R5 grouping strategy (collapse near-equivalent consecutive lines) is read by the engine (config.enableGrouping / groupingThreshold) but was unreachable in production: the rtkConfigSchema (.strict()) rejected the two fields on write and normalizeRtkConfig dropped them on read, so they could never be anything but undefined. Add enableGrouping (bool) + groupingThreshold (int, 2..100, default 3) to DEFAULT_RTK_CONFIG, normalizeRtkConfig, and rtkConfigSchema so the feature is actually settable and survives the DB round-trip. Default stays OFF — no behavior change for existing configs. Part of the compression "100% functional" program (audit follow-up). --- open-sse/services/compression/types.ts | 2 + src/lib/db/compression.ts | 10 +++ .../validation/compressionConfigSchemas.ts | 2 + .../compression/rtk-grouping-config.test.ts | 63 +++++++++++++++++++ 4 files changed, 77 insertions(+) create mode 100644 tests/unit/compression/rtk-grouping-config.test.ts diff --git a/open-sse/services/compression/types.ts b/open-sse/services/compression/types.ts index e3eebad8bc..0a84874976 100644 --- a/open-sse/services/compression/types.ts +++ b/open-sse/services/compression/types.ts @@ -230,6 +230,8 @@ export const DEFAULT_RTK_CONFIG: RtkConfig = { trustProjectFilters: false, rawOutputRetention: "never", rawOutputMaxBytes: 1_048_576, + enableGrouping: false, + groupingThreshold: 3, }; export const DEFAULT_COMPRESSION_LANGUAGE_CONFIG: CompressionLanguageConfig = { diff --git a/src/lib/db/compression.ts b/src/lib/db/compression.ts index 8e19fb9bc6..5b8ed98af4 100644 --- a/src/lib/db/compression.ts +++ b/src/lib/db/compression.ts @@ -171,6 +171,16 @@ function normalizeRtkConfig(value: unknown): RtkConfig { 1024, 10_000_000 ), + enableGrouping: + typeof record.enableGrouping === "boolean" + ? record.enableGrouping + : (DEFAULT_RTK_CONFIG.enableGrouping ?? false), + groupingThreshold: boundedInt( + record.groupingThreshold, + DEFAULT_RTK_CONFIG.groupingThreshold ?? 3, + 2, + 100 + ), }; } diff --git a/src/shared/validation/compressionConfigSchemas.ts b/src/shared/validation/compressionConfigSchemas.ts index d2fb09d8b4..5d7bbe6bdb 100644 --- a/src/shared/validation/compressionConfigSchemas.ts +++ b/src/shared/validation/compressionConfigSchemas.ts @@ -49,6 +49,8 @@ export const rtkConfigSchema = z trustProjectFilters: z.boolean().optional(), rawOutputRetention: rtkRawOutputRetentionSchema.optional(), rawOutputMaxBytes: z.number().int().min(1024).max(10_000_000).optional(), + enableGrouping: z.boolean().optional(), + groupingThreshold: z.number().int().min(2).max(100).optional(), }) .strict(); diff --git a/tests/unit/compression/rtk-grouping-config.test.ts b/tests/unit/compression/rtk-grouping-config.test.ts new file mode 100644 index 0000000000..6a65b1b6e3 --- /dev/null +++ b/tests/unit/compression/rtk-grouping-config.test.ts @@ -0,0 +1,63 @@ +import { describe, it, beforeEach, afterEach, after } from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { rtkConfigSchema } from "../../../src/shared/validation/compressionConfigSchemas.ts"; +import { DEFAULT_RTK_CONFIG } from "../../../open-sse/services/compression/types.ts"; + +// The RTK R5 grouping feature is read by the engine (config.enableGrouping / groupingThreshold) +// but was unreachable in production: the Zod schema (.strict()) rejected the two fields on write +// and normalizeRtkConfig dropped them on read. This proves both gates now let them through. + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-rtk-grouping-")); +const ORIGINAL_DATA_DIR = process.env.DATA_DIR; +process.env.DATA_DIR = TEST_DATA_DIR; + +const core = await import("../../../src/lib/db/core.ts"); +const { getCompressionSettings, updateCompressionSettings } = await import( + "../../../src/lib/db/compression.ts" +); + +describe("RTK grouping config persistence (R5)", () => { + beforeEach(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); + }); + + afterEach(() => { + core.resetDbInstance(); + }); + + after(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + if (ORIGINAL_DATA_DIR === undefined) delete process.env.DATA_DIR; + else process.env.DATA_DIR = ORIGINAL_DATA_DIR; + }); + + it("accepts enableGrouping / groupingThreshold on the write schema", () => { + assert.equal( + rtkConfigSchema.safeParse({ enableGrouping: true, groupingThreshold: 5 }).success, + true + ); + // groupingThreshold below the minimum run length (2) is rejected. + assert.equal(rtkConfigSchema.safeParse({ groupingThreshold: 1 }).success, false); + }); + + it("preserves enableGrouping / groupingThreshold through a DB round-trip", async () => { + const settings = await updateCompressionSettings({ + rtkConfig: { ...DEFAULT_RTK_CONFIG, enableGrouping: true, groupingThreshold: 7 }, + }); + assert.equal(settings.rtkConfig.enableGrouping, true); + assert.equal(settings.rtkConfig.groupingThreshold, 7); + + // Survives a fresh read (not just the write-path return value). + core.resetDbInstance(); + const reread = await getCompressionSettings(); + assert.equal(reread.rtkConfig.enableGrouping, true); + assert.equal(reread.rtkConfig.groupingThreshold, 7); + }); +});