From 84a2bc5fbcdd24c8e437fba042c40dc820f6c54b Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Wed, 27 May 2026 10:55:56 +0200 Subject: [PATCH] feat: fix so restart of server restarts batch jobs instead of failing them (#2755) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merge PR #2755 — feat: fix so restart of server restarts batch jobs instead of failing them --- .dockerignore | 9 ++-- src/sse/handlers/chat.ts | 110 ++++++++++++--------------------------- 2 files changed, 36 insertions(+), 83 deletions(-) diff --git a/.dockerignore b/.dockerignore index 0a050c9dc1..171c9d4631 100644 --- a/.dockerignore +++ b/.dockerignore @@ -38,14 +38,11 @@ playwright-report blob-report # Documentation -# Issue #2348: The Dashboard Docs viewer reads markdown from `/app/docs` at -# runtime. The previous `docs/*` block hid every file except openapi.yaml, -# so the in-product help screen failed with ENOENT for every page. # Translations (~51 MB) are excluded — the Docs viewer reads English sources. -# v3.8.3+ (fumadocs-mdx): `npm run build` webpack-bundles docs — keep assets referenced -# from English markdown (docs/diagrams/exported/*.svg, docs/screenshots/*.png). +# Screenshots (~1.7 MB) and SVGs (~250 KB) are needed at build time for MDX +# image resolution (fumadocs-mdx bundles them). +# Raster sources under docs/diagrams/ only (exported SVGs are required). docs/i18n/** -# Raster sources under docs/diagrams/ only (exported SVGs are required at build time). docs/diagrams/**/*.png docs/diagrams/**/*.jpg docs/diagrams/**/*.jpeg diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index 8517453ab8..ca5da75c62 100644 --- a/src/sse/handlers/chat.ts +++ b/src/sse/handlers/chat.ts @@ -123,7 +123,7 @@ const COMBOS_CACHE_TTL_MS = 10_000; async function getCombosCachedForChat(): Promise { const now = Date.now(); - if (combosCachePromise !== null && now - combosCacheTs < COMBOS_CACHE_TTL_MS) { + if (combosCachePromise && now - combosCacheTs < COMBOS_CACHE_TTL_MS) { return combosCachePromise; } @@ -153,27 +153,6 @@ function intersectAllowedConnectionIds(primary: unknown, secondary: unknown): st const PROVIDER_BREAKER_FAILURE_STATUSES = new Set([408, 500, 502, 503, 504]); -/** - * Wrap a Request-like object so callers reading `.signal` see a merged signal - * combining the original request signal with an extra per-call signal (e.g. a - * combo per-model timeout). Returns the original request unchanged when no - * extra signal is provided, so the hot path stays a no-op. - */ -function wrapRequestWithExtraSignal(request: any, extraSignal: AbortSignal | null) { - if (!extraSignal || !request) return request; - const baseSignal: AbortSignal | null | undefined = request?.signal ?? null; - const mergedSignal: AbortSignal = baseSignal - ? AbortSignal.any([baseSignal, extraSignal]) - : extraSignal; - return new Proxy(request, { - get(target, prop, receiver) { - if (prop === "signal") return mergedSignal; - const value = target[prop]; - return typeof value === "function" ? value.bind(target) : value; - }, - }); -} - /** * Handle chat completion request * Supports: OpenAI, Claude, Gemini, OpenAI Responses API formats @@ -221,9 +200,6 @@ export async function handleChat(request: any, clientRawRequest: any = null) { // Log request endpoint and model const url = new URL(request.url); - // `let` because the middleware-hook pipeline (line ~319) may reassign this - // when a hook rewrites the target model. Previously declared `const`, which - // broke turbopack/strict-mode builds (PR [PR #2670](file:///home/diegosouzapw/dev/proxys/OmniRoute/package.json#L2670) regression). let modelStr = body.model; // Count messages (support both messages[] and input[] formats) @@ -332,7 +308,7 @@ export async function handleChat(request: any, clientRawRequest: any = null) { >, model: modelStr, combo: undefined, - apiKeyInfo: apiKeyInfo as unknown as Record | undefined, + apiKeyInfo: apiKeyInfo as Record | undefined, log, }); @@ -493,11 +469,6 @@ export async function handleChat(request: any, clientRawRequest: any = null) { resolvedModel, { sessionKey: sessionAffinityKey, - sessionAffinityTtlMs: Number.isFinite( - Number((settings as any)?.codexSessionAffinityTtlMs) - ) - ? Number((settings as any).codexSessionAffinityTtlMs) - : null, ...(target?.connectionId ? { forcedConnectionId: target.connectionId } : {}), } ); @@ -530,15 +501,13 @@ export async function handleChat(request: any, clientRawRequest: any = null) { stepId?: string | null; allowedConnectionIds?: string[] | null; failoverBeforeRetry?: boolean; - trafficType?: "production" | "shadow"; - modelAbortSignal?: AbortSignal | null; } ) => handleSingleModelChat( b, m, clientRawRequest, - wrapRequestWithExtraSignal(request, target?.modelAbortSignal ?? null), + request, combo.name, apiKeyInfo, telemetry, @@ -551,7 +520,6 @@ export async function handleChat(request: any, clientRawRequest: any = null) { comboStepId: target?.stepId || null, comboExecutionKey: target?.executionKey || target?.stepId || null, skipUpstreamRetry: target?.failoverBeforeRetry ?? false, - trafficType: target?.trafficType === "shadow" ? "shadow" : "production", preselectedCredentials: comboPreselectedCredentials.get( getComboCredentialCacheKey(m, target) ), @@ -566,7 +534,7 @@ export async function handleChat(request: any, clientRawRequest: any = null) { allCombos, apiKeyAllowedConnections: apiKeyInfo?.allowedConnections ?? null, relayOptions: - combo.strategy === "context-relay" || combo.strategy === "auto" + combo.strategy === "context-relay" ? { sessionId, config: relayConfig, @@ -685,7 +653,6 @@ async function handleSingleModelChat( skipUpstreamRetry?: boolean; preselectedCredentials?: any; cachedSettings?: any; - trafficType?: "production" | "shadow"; } = {}, comboStrategy: string | null = null, isCombo: boolean = false @@ -709,12 +676,21 @@ async function handleSingleModelChat( return handleComboChat({ body, combo: redirectCombo, - handleSingleModel: (b: any, m: string, target?: any) => + handleSingleModel: ( + b: any, + m: string, + target?: { + connectionId?: string | null; + executionKey?: string | null; + stepId?: string | null; + failoverBeforeRetry?: boolean; + } + ) => handleSingleModelChat( b, m, clientRawRequest, - wrapRequestWithExtraSignal(request, target?.modelAbortSignal ?? null), + request, redirectCombo.name ?? modelStr, apiKeyInfo, telemetry, @@ -726,7 +702,6 @@ async function handleSingleModelChat( comboStepId: null, comboExecutionKey: null, skipUpstreamRetry: target?.failoverBeforeRetry ?? false, - trafficType: target?.trafficType === "shadow" ? "shadow" : "production", }, redirectCombo.strategy ?? "priority", false @@ -742,7 +717,6 @@ async function handleSingleModelChat( const { provider, model, sourceFormat, targetFormat, extendedContext, apiFormat } = resolved; const forceLiveComboTest = runtimeOptions.forceLiveComboTest === true; - const isShadowTraffic = runtimeOptions.trafficType === "shadow"; const hasForcedConnection = typeof runtimeOptions.forcedConnectionId === "string" && runtimeOptions.forcedConnectionId.trim().length > 0; @@ -837,11 +811,6 @@ async function handleSingleModelChat( { sessionKey: runtimeOptions.sessionAffinityKey ?? runtimeOptions.sessionId ?? null, excludeConnectionIds: Array.from(excludedConnectionIds), - sessionAffinityTtlMs: Number.isFinite( - Number((runtimeOptions.cachedSettings as any)?.codexSessionAffinityTtlMs) - ) - ? Number((runtimeOptions.cachedSettings as any).codexSessionAffinityTtlMs) - : null, ...(forceLiveComboTest ? { allowSuppressedConnections: true, @@ -953,11 +922,7 @@ async function handleSingleModelChat( ...(workspaceId ? { workspaceId } : {}), }); } - if ( - !isShadowTraffic && - runtimeOptions.sessionId && - body?._omnirouteInternalRequest !== "context-handoff" - ) { + if (runtimeOptions.sessionId && body?._omnirouteInternalRequest !== "context-handoff") { touchSession(runtimeOptions.sessionId, credentials.connectionId); startQuotaMonitor( runtimeOptions.sessionId, @@ -972,7 +937,7 @@ async function handleSingleModelChat( // 4. Execute chat via core after breaker gate checks (with optional TLS tracking) if (telemetry) telemetry.startPhase("connect"); const { result, tlsFingerprintUsed } = await executeChatWithBreaker({ - bypassCircuitBreaker: forceLiveComboTest || hasForcedConnection || isShadowTraffic, + bypassCircuitBreaker: forceLiveComboTest || hasForcedConnection, breaker, body: requestBody, provider, @@ -994,7 +959,6 @@ async function handleSingleModelChat( providerProfile, cachedSettings: runtimeOptions.cachedSettings, skipUpstreamRetry: runtimeOptions.skipUpstreamRetry ?? false, - trafficType: isShadowTraffic ? "shadow" : "production", }); if (telemetry) telemetry.endPhase(); @@ -1021,13 +985,11 @@ async function handleSingleModelChat( }); if (result.success) { - if (!isShadowTraffic) { - clearModelLock(provider, credentials.connectionId, model); - } - if (!forceLiveComboTest && !isShadowTraffic) { + clearModelLock(provider, credentials.connectionId, model); + if (!forceLiveComboTest) { breaker._onSuccess(); } - if (!isShadowTraffic && injectedHandoff && runtimeOptions.sessionId && comboName) { + if (injectedHandoff && runtimeOptions.sessionId && comboName) { deleteHandoff(runtimeOptions.sessionId, comboName); } if (telemetry) telemetry.startPhase("finalize"); @@ -1111,7 +1073,7 @@ async function handleSingleModelChat( // Emergency fallback for budget exhaustion (402 / billing / quota keywords): // reroute to a free model (default provider/model: nvidia + openai/gpt-oss-120b) exactly once. - if (!isShadowTraffic && !isCombo && !runtimeOptions.emergencyFallbackTried) { + if (!runtimeOptions.emergencyFallbackTried) { const fallbackDecision = shouldUseFallback( Number(result.status || 0), String(result.error || ""), @@ -1178,11 +1140,7 @@ async function handleSingleModelChat( // Daily quota lockout overrides subsequent rate_limited lockout, ensuring lockout until tomorrow 0:00 let dailyQuotaExhausted = false; const errorStr = String(result.error || ""); - if (isCombo && comboStrategy !== "context-relay" && result.status === 429) { - return result.response; - } - - if (!isShadowTraffic && result.status === 429 && isDailyQuotaExhausted(errorStr)) { + if (result.status === 429 && isDailyQuotaExhausted(errorStr)) { // Parse which model is quota-limited const match = errorStr.match(/today's quota for model ([^,]+)/); const limitedModel = match ? match[1].trim() : model; @@ -1214,7 +1172,7 @@ async function handleSingleModelChat( // 7. Mark account as quota-exhausted only for explicit long-window quota signals. // A plain 429/high-traffic response should trigger fallback/cooldown, not poison // quotaCache as exhausted for 5 minutes while usage quota may still be available. - if (!isShadowTraffic && !dailyQuotaExhausted) { + if (!dailyQuotaExhausted) { const passthroughModels = credentials.providerSpecificData?.passthroughModels; const failureKind = result.status === 429 @@ -1238,17 +1196,16 @@ async function handleSingleModelChat( const is401 = result.status === 401; const skipConnectionDisable = is401 && hasExtraKeys; - const { shouldFallback, cooldownMs } = - skipConnectionDisable || isShadowTraffic - ? { shouldFallback: false, cooldownMs: 0 } - : await markAccountUnavailable( - credentials.connectionId, - result.status, - result.error, - provider, - model, - providerProfile - ); + const { shouldFallback, cooldownMs } = skipConnectionDisable + ? { shouldFallback: false, cooldownMs: 0 } + : await markAccountUnavailable( + credentials.connectionId, + result.status, + result.error, + provider, + model, + providerProfile + ); if (shouldFallback) { if (Number.isFinite(cooldownMs) && cooldownMs > 0) { @@ -1266,7 +1223,6 @@ async function handleSingleModelChat( if ( !forceLiveComboTest && - !isShadowTraffic && !isCombo && PROVIDER_BREAKER_FAILURE_STATUSES.has(Number(result.status)) ) {