mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-04 14:22:09 +03:00
refactor(combo): extract reset-aware quota block to combo/quotaStrategies.ts (QG v2 Fase 9 T5 D7b) (#4204)
This commit is contained in:
Diego Rodrigues de Sa e Souza
committed by
GitHub
GitHub
parent
3135be8e7b
commit
565eb281a7
@@ -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, <cap): runtime-step normalization, nested-combo/DAG expansion (resolveNestedComboTargets/expandRuntimeStep/getComboFromData/getComboModelsFromData/validateComboDAG/resolveNestedComboModels), weighted/direct resolution (resolveComboTargets/resolveWeightedTargets/getDirectComboTargets + composite-tier ordering), request-compatibility filtering (filterTargetsByRequestCompatibility + deriveRequestCompatibilityRequirements/getTargetCompatibilityFailures and their private helpers) and context-size sorting. Shared dedupeTargetsByExecutionKey + the toTrimmedString/toComboLike/getCombosArray normalizers also moved (dedupe to comboData.ts since both combo.ts and comboStructure use it). The 7 previously-public symbols are re-exported from combo.ts; the ~20 external consumers (chatCore.ts, /api/combos routes, embeddings, usage, catalog) are unchanged. shadowRouting.ts's resolveNestedComboTargets import redirected from the barrel to ./comboStructure.ts (removing the D4 temporary edge). Orphaned combo.ts imports dropped (getComboModelString/normalizeComboStep, getModelContextLimit, getResolvedModelCapabilities, getTargetProvider, MAX_COMBO_DEPTH, and 6 now-unused type imports). Pure move, no logic change.",
|
||||
"_rebaseline_2026_06_18_qg9_combo_split_d8": "QG v2 Fase 9 T5 D8 (reduced): combo.ts 3819->3432 — auto-strategy scoring/intent/tag-routing extracted byte-identically to the new open-sse/services/combo/autoStrategy.ts (434, <cap): the quota-soft execution-candidate registry (the single _activeExecutionCandidates Map + setCandidateQuotaSoftPenalty/_registerExecutionCandidates/_unregisterExecutionCandidates, kept together for state cohesion), QUOTA_SOFT_DEPRIORITIZE_FACTOR, scoreAutoTargets, expandAutoComboCandidatePool, intent extraction (toTextContent/extractPromptForIntent/mapIntentToTaskType/toStringArray/getIntentConfig), applyRequestTagRouting and deriveComboSessionKey. buildAutoCandidates + its two private-only helpers calculateTargetContextAffinity/getBootstrapLatencyMs (and the DEFAULT_MODEL_P95_MS/MIN_HISTORY_SAMPLES/OUTPUT_TOKEN_RATIO consts they own) were DELIBERATELY KEPT in combo.ts: buildAutoCandidates is the sole user of the internal reset-window helpers (resolveResetWindowConfig/fetchResetAwareQuotaWithCache/calculateResetWindowAffinity + the ResetWindowConfig type), so keeping it there leaves those helpers private (no export) and avoids a combo <-> 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, <cap). Moved exactly the 6 RR-state symbols: MAX_RR_COUNTERS, the two mutable module Maps rrCounters (new Map<string,number>) + 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 <cap). Split was size-driven (~818 source LOC > 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 <cap), NOT inlined. Thin wiring only; not further shrinkable.",
|
||||
"_rebaseline_2026_06_18_8_1_no_thinking_alias": "Fase 8.1 own growth: catalog.ts 1435->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 (<cap); both edits are thin wiring of tested helpers at the single correct integration point in each file. Not extractable.",
|
||||
"_rebaseline_2026_06_18_4_4_midstream": "Fase 4.4 own growth: chatCore.ts 5980->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 (<cap). Closes over the same per-attempt executor locals as the existing reopen; not extractable without hiding the dispatch boundary. Opt-in (default OFF).",
|
||||
@@ -77,7 +78,7 @@
|
||||
"open-sse/services/batchProcessor.ts": 828,
|
||||
"open-sse/services/browserBackedChat.ts": 850,
|
||||
"open-sse/services/claudeCodeCompatible.ts": 1202,
|
||||
"open-sse/services/combo.ts": 3398,
|
||||
"open-sse/services/combo.ts": 2597,
|
||||
"open-sse/services/rateLimitManager.ts": 1035,
|
||||
"open-sse/services/tokenRefresh.ts": 1997,
|
||||
"open-sse/services/usage.ts": 3408,
|
||||
|
||||
@@ -17,11 +17,9 @@ import {
|
||||
recordProviderFailure,
|
||||
isProviderExhaustedReason,
|
||||
hasPerModelQuota,
|
||||
type ProviderProfile,
|
||||
} from "./accountFallback.ts";
|
||||
import { FETCH_TIMEOUT_MS, RateLimitReason } from "../config/constants.ts";
|
||||
import { errorResponse, unavailableResponse } from "../utils/error.ts";
|
||||
import { clamp01 } from "../utils/number.ts";
|
||||
import {
|
||||
recordComboIntent,
|
||||
recordComboRequest,
|
||||
@@ -32,7 +30,6 @@ import {
|
||||
resolveComboConfig,
|
||||
getDefaultComboConfig,
|
||||
resolveComboTargetTimeoutMs,
|
||||
PRE_SCREEN_CONCURRENCY,
|
||||
} from "./comboConfig.ts";
|
||||
import {
|
||||
maybeGenerateHandoff,
|
||||
@@ -62,7 +59,7 @@ import { emit } from "../../src/lib/events/eventBus";
|
||||
import { notifyWebhookEvent } from "../../src/lib/webhookDispatcher";
|
||||
import { classifyWithConfig } from "./intentClassifier.ts";
|
||||
import { selectProvider as selectAutoProvider } from "./autoCombo/engine.ts";
|
||||
import { selectWithStrategy, type SlaRoutingPolicy } from "./autoCombo/routerStrategy.ts";
|
||||
import { selectWithStrategy } from "./autoCombo/routerStrategy.ts";
|
||||
import { parseAutoPrefix } from "./autoCombo/autoPrefix.ts";
|
||||
import { handlePipelineCombo, buildPipelineResponse } from "./autoCombo/pipelineRouter.ts";
|
||||
import { DEFAULT_WEIGHTS, type ProviderCandidate, type ScoringWeights } from "./autoCombo/scoring.ts";
|
||||
@@ -91,7 +88,6 @@ import type {
|
||||
ComboRetryAfter,
|
||||
ComboErrorBody,
|
||||
SingleModelTarget,
|
||||
IsModelAvailable,
|
||||
HandleComboChatOptions,
|
||||
HandleRoundRobinOptions,
|
||||
ResolvedComboTarget,
|
||||
@@ -150,6 +146,19 @@ import {
|
||||
expandAutoComboCandidatePool,
|
||||
deriveComboSessionKey,
|
||||
} from "./combo/autoStrategy.ts";
|
||||
import {
|
||||
resolveResetWindowConfig,
|
||||
resolveSlaRoutingPolicy,
|
||||
calculateResetWindowAffinity,
|
||||
type ResetWindowConfig,
|
||||
} from "./combo/quotaScoring.ts";
|
||||
import {
|
||||
fetchResetAwareQuotaWithCache,
|
||||
preScreenTargets,
|
||||
orderTargetsByResetAwareQuota,
|
||||
orderTargetsByResetWindow,
|
||||
type PreScreenResult,
|
||||
} from "./combo/quotaStrategies.ts";
|
||||
|
||||
// Backward-compatible re-exports — these were public from combo.ts before the
|
||||
// types extraction (Quality Gate v2 / Fase 9). Keep the external surface stable.
|
||||
@@ -162,6 +171,9 @@ export type { SingleModelTarget, ResolvedComboTarget };
|
||||
export { validateResponseQuality };
|
||||
export { clampComboDepth, shouldSkipForPredictedTtft, shouldRecordProviderBreakerFailure };
|
||||
export { resolveShadowTargets, scheduleShadowRouting };
|
||||
// preScreenTargets was public from combo.ts before the reset-aware quota
|
||||
// extraction (combo split D7b). Keep the external surface stable.
|
||||
export { preScreenTargets };
|
||||
export { resolveComboTargets, filterTargetsByRequestCompatibility };
|
||||
export {
|
||||
getComboFromData,
|
||||
@@ -171,695 +183,13 @@ export {
|
||||
validateComboDAG,
|
||||
} from "./combo/comboStructure.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_CONNECTION_CACHE_TTL_MS = 30_000;
|
||||
const RESET_AWARE_QUOTA_FETCH_CONCURRENCY = 5;
|
||||
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];
|
||||
type QuotaFetchCacheConfig = {
|
||||
quotaCacheTtlMs: number;
|
||||
quotaCacheMaxStaleMs: number;
|
||||
};
|
||||
type ResetWindowConfig = ReturnType<typeof resolveResetWindowConfig>;
|
||||
|
||||
const MAX_RESET_AWARE_CACHE = 200;
|
||||
|
||||
const resetAwareConnectionCache = new Map<
|
||||
string,
|
||||
{ fetchedAt: number; connections: Array<Record<string, unknown>> }
|
||||
>();
|
||||
const resetAwareQuotaCache = new Map<
|
||||
string,
|
||||
{ fetchedAt: number; quota: unknown; refreshPromise: Promise<unknown> | 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<string, unknown> | 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<string, unknown> | 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<string, unknown> | 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<typeof resolveResetAwareConfig>) {
|
||||
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<string, Array<Record<string, unknown>>>,
|
||||
connectionLoadPromises: Map<string, Promise<Array<Record<string, unknown>>>>,
|
||||
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<Record<string, unknown>>)
|
||||
: [];
|
||||
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<Record<string, unknown>>
|
||||
): 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<T, R>(
|
||||
items: T[],
|
||||
concurrency: number,
|
||||
mapper: (item: T, index: number) => Promise<R>
|
||||
): Promise<R[]> {
|
||||
const results = new Array<R>(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<string, unknown>;
|
||||
fetcher: (connectionId: string, connection?: Record<string, unknown>) => Promise<unknown>;
|
||||
config: QuotaFetchCacheConfig;
|
||||
log: { debug?: (...args: unknown[]) => void; warn?: (...args: unknown[]) => void };
|
||||
comboName: string;
|
||||
}): Promise<unknown> {
|
||||
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<Map<string, PreScreenResult>> {
|
||||
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<string, PreScreenResult>();
|
||||
for (const { key, result } of results) {
|
||||
map.set(key, result);
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
async function orderTargetsByResetAwareQuota(
|
||||
targets: ResolvedComboTarget[],
|
||||
comboName: string,
|
||||
configSource: Record<string, unknown> | null | undefined,
|
||||
log: { warn?: (...args: unknown[]) => void },
|
||||
apiKeyAllowedConnectionIds?: string[] | null
|
||||
) {
|
||||
if (targets.length === 0) return targets;
|
||||
|
||||
const config = resolveResetAwareConfig(configSource);
|
||||
const connectionCache = new Map<string, Array<Record<string, unknown>>>();
|
||||
const connectionLoadPromises = new Map<string, Promise<Array<Record<string, unknown>>>>();
|
||||
const quotaPromises = new Map<string, Promise<unknown>>();
|
||||
const connectionById = new Map<string, Record<string, unknown>>();
|
||||
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<string, number> = {
|
||||
@@ -1069,137 +399,6 @@ export async function buildAutoCandidates(
|
||||
return candidates;
|
||||
}
|
||||
|
||||
async function orderTargetsByResetWindow(
|
||||
targets: ResolvedComboTarget[],
|
||||
comboName: string,
|
||||
configSource: Record<string, unknown> | null | undefined,
|
||||
log: { warn?: (...args: unknown[]) => void },
|
||||
apiKeyAllowedConnectionIds?: string[] | null
|
||||
) {
|
||||
if (targets.length === 0) return targets;
|
||||
|
||||
const config = resolveResetWindowConfig(configSource);
|
||||
const connectionCache = new Map<string, Array<Record<string, unknown>>>();
|
||||
const connectionLoadPromises = new Map<string, Promise<Array<Record<string, unknown>>>>();
|
||||
const quotaPromises = new Map<string, Promise<unknown>>();
|
||||
const connectionById = new Map<string, Record<string, unknown>>();
|
||||
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
|
||||
|
||||
311
open-sse/services/combo/quotaScoring.ts
Normal file
311
open-sse/services/combo/quotaScoring.ts
Normal file
@@ -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<typeof resolveResetWindowConfig>;
|
||||
|
||||
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<string, unknown> | 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<string, unknown> | 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<string, unknown> | 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<typeof resolveResetAwareConfig>
|
||||
) {
|
||||
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));
|
||||
}
|
||||
568
open-sse/services/combo/quotaStrategies.ts
Normal file
568
open-sse/services/combo/quotaStrategies.ts
Normal file
@@ -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<Record<string, unknown>> }
|
||||
>();
|
||||
const resetAwareQuotaCache = new Map<
|
||||
string,
|
||||
{ fetchedAt: number; quota: unknown; refreshPromise: Promise<unknown> | null }
|
||||
>();
|
||||
|
||||
async function getQuotaAwareConnectionsForTarget(
|
||||
target: ResolvedComboTarget,
|
||||
connectionCache: Map<string, Array<Record<string, unknown>>>,
|
||||
connectionLoadPromises: Map<string, Promise<Array<Record<string, unknown>>>>,
|
||||
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<Record<string, unknown>>)
|
||||
: [];
|
||||
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<Record<string, unknown>>
|
||||
): 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<T, R>(
|
||||
items: T[],
|
||||
concurrency: number,
|
||||
mapper: (item: T, index: number) => Promise<R>
|
||||
): Promise<R[]> {
|
||||
const results = new Array<R>(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<string, unknown>;
|
||||
fetcher: (connectionId: string, connection?: Record<string, unknown>) => Promise<unknown>;
|
||||
config: QuotaFetchCacheConfig;
|
||||
log: { debug?: (...args: unknown[]) => void; warn?: (...args: unknown[]) => void };
|
||||
comboName: string;
|
||||
}): Promise<unknown> {
|
||||
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<Map<string, PreScreenResult>> {
|
||||
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<string, PreScreenResult>();
|
||||
for (const { key, result } of results) {
|
||||
map.set(key, result);
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
export async function orderTargetsByResetAwareQuota(
|
||||
targets: ResolvedComboTarget[],
|
||||
comboName: string,
|
||||
configSource: Record<string, unknown> | null | undefined,
|
||||
log: { warn?: (...args: unknown[]) => void },
|
||||
apiKeyAllowedConnectionIds?: string[] | null
|
||||
) {
|
||||
if (targets.length === 0) return targets;
|
||||
|
||||
const config = resolveResetAwareConfig(configSource);
|
||||
const connectionCache = new Map<string, Array<Record<string, unknown>>>();
|
||||
const connectionLoadPromises = new Map<string, Promise<Array<Record<string, unknown>>>>();
|
||||
const quotaPromises = new Map<string, Promise<unknown>>();
|
||||
const connectionById = new Map<string, Record<string, unknown>>();
|
||||
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<string, unknown> | null | undefined,
|
||||
log: { warn?: (...args: unknown[]) => void },
|
||||
apiKeyAllowedConnectionIds?: string[] | null
|
||||
) {
|
||||
if (targets.length === 0) return targets;
|
||||
|
||||
const config = resolveResetWindowConfig(configSource);
|
||||
const connectionCache = new Map<string, Array<Record<string, unknown>>>();
|
||||
const connectionLoadPromises = new Map<string, Promise<Array<Record<string, unknown>>>>();
|
||||
const quotaPromises = new Map<string, Promise<unknown>>();
|
||||
const connectionById = new Map<string, Record<string, unknown>>();
|
||||
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);
|
||||
}
|
||||
Reference in New Issue
Block a user