feat(providers): restore per-connection upstream timeout tier (#10885)

Reconciliado com a release (conflito aditivo em targetTimeoutRunner.ts — combina o warning G3 já mergeado com a resolução de effectiveTimeoutMs deste PR) e revalidado: 41/41 testes focados passando (upstream-timeout-connection-tier, combo-target-timeout-runner, provider-specific-data-schema). CI vermelho é o base-red já rastreado em #9985. Obrigado!

changelog.d/features/per-connection-upstream-timeout.md

@@ -0,0 +1 @@
+- **feat(providers):** restore the operator-owned upstream timeout tier per connection via `providerSpecificData.timeoutMs` (preempts the maintainer-only model/provider registry tiers and the global `FETCH_TIMEOUT_MS`), and make the combo per-target timeout ceiling follow the selected connection

open-sse/handlers/chatCore.ts

@@ -347,6 +347,7 @@ import {
   computeBillableTokens,
   normalizeExecutorResult,
   executeWithUpstreamStartTimeout,
+  resolveConnectionTimeoutMs,
 } from "./chatCore/upstreamTimeouts.ts";
 import { getModelNormalizeToolCallId, getModelPreserveOpenAIDeveloperRole } from "@/lib/db/models";
 import { getProviderCredentials, extractSessionAffinityKey } from "@/sse/services/auth";
@@ -3065,6 +3066,7 @@ export async function handleChatCore({
                     executor,
                     provider,
                     model: modelToCall,
+                    connectionTimeoutMs: resolveConnectionTimeoutMs(execCreds?.providerSpecificData),
                     signal: streamController.signal,
                     log,
                     execute: (signal) =>
@@ -3368,6 +3370,7 @@ export async function handleChatCore({
                         executor,
                         provider,
                         model: modelToCall,
+                        connectionTimeoutMs: resolveConnectionTimeoutMs(execCreds?.providerSpecificData),
                         signal: streamController.signal,
                         log,
                         execute: (signal) =>

open-sse/handlers/chatCore/upstreamTimeouts.ts

@@ -9,6 +9,7 @@ import {
   getLoggedOutputTokens,
   getReasoningTokens,
 } from "@/lib/usage/tokenAccounting";
+import { MAX_PROVIDER_SPECIFIC_TIMEOUT_MS } from "@/shared/validation/providerSpecificData";
 
 export function createBodyTimeoutError(timeoutMs: number): Error {
   const err = new Error(`Response body read timeout after ${timeoutMs}ms`);
@@ -89,14 +90,44 @@ function resolveProviderTimeoutMs(executor: unknown): number {
   }
 }
 
+/** Per-connection operator timeout tier: reads
+ *  `providerSpecificData.timeoutMs`, bounded to 1..86_400_000 ms.
+ *  Returns undefined when absent or invalid so the chain falls through. */
+export function resolveConnectionTimeoutMs(psd: unknown): number | undefined {
+  const timeoutMs = (psd as Record<string, unknown> | null | undefined)?.timeoutMs;
+  if (typeof timeoutMs !== "number" || !Number.isFinite(timeoutMs)) return undefined;
+  const floored = Math.floor(timeoutMs);
+  if (floored < 1 || floored > MAX_PROVIDER_SPECIFIC_TIMEOUT_MS) return undefined;
+  return floored;
+}
+
 /**
  * Resolves the upstream header-response timeout in precedence order:
+ * connection-level override (`providerSpecificData.timeoutMs`) →
  * model-level override (registry `RegistryModel.timeoutMs`) → provider-level
  * override (`executor.getTimeoutMs()`) → global `FETCH_TIMEOUT_MS` default.
  * `provider`/`model` are optional so existing single-argument call sites
  * keep resolving to the provider/global chain unchanged (#6354).
  */
-export function getExecutorTimeoutMs(executor: unknown, provider?: string, model?: string): number {
+export function getExecutorTimeoutMs(
+  executor: unknown,
+  provider?: string,
+  model?: string,
+  connectionTimeoutMs?: number
+): number {
+  if (
+    typeof connectionTimeoutMs === "number" &&
+    Number.isFinite(connectionTimeoutMs) &&
+    connectionTimeoutMs > 0
+  ) {
+    // Defensive backstop for direct callers: resolveConnectionTimeoutMs is the
+    // gate (it rejects out-of-range values so the chain falls through); this
+    // clamp only caps values a future caller could pass unvetted.
+    return Math.min(
+      Math.max(0, Math.floor(connectionTimeoutMs)),
+      MAX_PROVIDER_SPECIFIC_TIMEOUT_MS
+    );
+  }
   const modelOverride = resolveModelTimeoutOverride(provider, model);
   if (modelOverride !== undefined) return modelOverride;
   return resolveProviderTimeoutMs(executor);
@@ -196,6 +227,7 @@ export async function executeWithUpstreamStartTimeout<T>({
   executor,
   provider,
   model,
+  connectionTimeoutMs,
   signal,
   log,
   execute,
@@ -203,11 +235,12 @@ export async function executeWithUpstreamStartTimeout<T>({
   executor: unknown;
   provider: string;
   model: string;
+  connectionTimeoutMs?: number;
   signal: AbortSignal;
   log?: { warn?: (tag: string, message: string) => void } | null;
   execute: (signal: AbortSignal) => Promise<T>;
 }): Promise<T> {
-  const timeoutMs = getExecutorTimeoutMs(executor, provider, model);
+  const timeoutMs = getExecutorTimeoutMs(executor, provider, model, connectionTimeoutMs);
   if (timeoutMs <= 0) return execute(signal);
   if (signal.aborted) throw createAbortError(signal);
 

open-sse/services/combo.ts

@@ -46,6 +46,7 @@ import {
   getDefaultComboConfig,
   resolveComboQueueDepth,
   isComboCooldownWaitEligible,
+  resolveComboTargetTimeoutMsForCombo,
 } from "./comboConfig.ts";
 import {
   maybeGenerateHandoff,
@@ -89,6 +90,7 @@ import { selectQuotaShareTarget } from "./combo/quotaShareStrategy.ts";
 import { makeConnectionConcurrencyResolver, lookupPositiveCap } from "./combo/concurrencyCaps.ts";
 import { acquireQuotaShareConcurrencySlot } from "./combo/quotaShareConcurrency.ts";
 import { canAffordRequest } from "../../src/lib/quota/quotaScheduler.ts";
+import { resolveConnectionTimeoutMs } from "../handlers/chatCore/upstreamTimeouts.ts";
 import { getCachedProviderConnectionById } from "../../src/lib/db/readCache.ts";
 import { orderTargetsByEvalScores } from "./evalRouting.ts";
 
@@ -133,6 +135,7 @@ import { isProviderInCooldown, recordProviderCooldown } from "./providerCooldown
 import {
   resolveResilienceSettings,
   type ResilienceSettings,
+  type ComboCooldownWaitSettings,
 } from "../../src/lib/resilience/settings";
 import { resolveReasoningBufferedMaxTokens, toPositiveInteger } from "./reasoningTokenBuffer.ts";
 import { RESET_WINDOW_NAMES } from "./combo/types.ts";
@@ -141,6 +144,7 @@ import type {
   ComboRetryAfter,
   ComboErrorBody,
   SingleModelTarget,
+  ComboLogger,
   HandleComboChatOptions,
   HandleRoundRobinOptions,
   ResolvedComboTarget,
@@ -621,6 +625,40 @@ export { pinIsDurablyUnhealthy };
 /** @param {string} errorText */
 
 /** @param {object} options */
+/**
+ * Resolves the per-target timeout ceiling for a combo target: when the target's
+ * connection carries `providerSpecificData.timeoutMs`, re-runs
+ * resolveComboTargetTimeoutMsForCombo with that timeout as the ceiling so the
+ * combo's per-target timer follows the selected connection.
+ * Returns undefined when the connection or its timeout is absent — the runner
+ * then falls back to the setup-time comboTargetTimeoutMs.
+ */
+export async function resolveTargetTimeoutMsForTarget(
+  config: Record<string, unknown> | null | undefined,
+  strategy: string,
+  comboCooldownWait: Pick<ComboCooldownWaitSettings, "enabled" | "budgetMs">,
+  target?: SingleModelTarget,
+  log?: Pick<ComboLogger, "debug"> | null
+): Promise<number | undefined> {
+  const connectionId = target && "connectionId" in target ? target.connectionId : null;
+  if (!connectionId) return undefined;
+  try {
+    const connection = await getCachedProviderConnectionById(connectionId);
+    if (!connection) return undefined;
+    const timeoutMs = resolveConnectionTimeoutMs(connection.providerSpecificData);
+    if (timeoutMs === undefined) return undefined;
+    return resolveComboTargetTimeoutMsForCombo(config, timeoutMs, strategy, comboCooldownWait);
+  } catch (err) {
+    log?.debug?.(
+      "COMBO",
+      `resolveTargetTimeoutMsForTarget connection lookup failed: ${
+        err instanceof Error ? err.message : String(err)
+      }`
+    );
+    return undefined;
+  }
+}
+
 /**
  * #10681 egress: every combo response carries the opaque trace id in an
  * `X-OmniRoute-Combo-Trace` header so a post-incident lookup of the ordered
@@ -683,6 +721,14 @@ async function handleComboChatInner({
   const handleSingleModelWithTimeout = buildTargetTimeoutRunner({
     handleSingleModel,
     comboTargetTimeoutMs,
+    resolveTargetTimeoutMs: (target) =>
+      resolveTargetTimeoutMsForTarget(
+        config,
+        strategy,
+        resilienceSettings.comboCooldownWait,
+        target,
+        log
+      ),
     log,
   });
 

open-sse/services/combo/targetTimeoutRunner.ts

@@ -93,25 +93,33 @@ export function buildTargetTimeoutRunner(deps: {
   handleSingleModel: HandleSingleModel;
   comboTargetTimeoutMs: number;
   log: ComboLogger;
+  resolveTargetTimeoutMs?: (
+    target?: SingleModelTarget
+  ) => Promise<number | undefined> | number | undefined;
 }): (
   b: Record<string, unknown>,
   modelStr: string,
   target?: SingleModelTarget
 ) => Promise<Response> {
-  const { handleSingleModel, comboTargetTimeoutMs, log } = deps;
+  const { handleSingleModel, comboTargetTimeoutMs, log, resolveTargetTimeoutMs } = deps;
   ensureDiagnosticListener();
   return async (
     b: Record<string, unknown>,
     modelStr: string,
     target?: SingleModelTarget
   ): Promise<Response> => {
-    if (comboTargetTimeoutMs <= 0) {
+    const resolvedTimeoutMs = await resolveTargetTimeoutMs?.(target);
+    const effectiveTimeoutMs =
+      typeof resolvedTimeoutMs === "number" && Number.isFinite(resolvedTimeoutMs)
+        ? resolvedTimeoutMs
+        : comboTargetTimeoutMs;
+    if (effectiveTimeoutMs <= 0) {
       // G3 (silent-stop fix): a disabled per-model timeout means a hung upstream
       // stalls the target until the combo loop safety timer (COMBO_LOOP_SAFETY_TIMEOUT_MS)
       // force-terminates — surface that dependency instead of silently running bare.
       log.warn(
         "COMBO",
-        `Per-model combo timeout is DISABLED (comboTargetTimeoutMs=${comboTargetTimeoutMs}) for ${modelStr} — a hung upstream will hang this target until the combo loop safety timeout`
+        `Per-model combo timeout is DISABLED (effectiveTimeoutMs=${effectiveTimeoutMs}) for ${modelStr} — a hung upstream will hang this target until the combo loop safety timeout`
       );
       return handleSingleModel(b, modelStr, target).catch((err) =>
         errorResponse(502, err?.message ?? "Upstream model error")
@@ -127,13 +135,13 @@ export function buildTargetTimeoutRunner(deps: {
         const abortErr = new Error(COMBO_PER_MODEL_TIMEOUT_REASON);
         recordTimeoutContext({
           modelStr,
-          timeoutMs: comboTargetTimeoutMs,
+          timeoutMs: effectiveTimeoutMs,
           abortError: abortErr,
           timestamp: Date.now(),
         });
         log.warn(
           "COMBO",
-          `Model ${modelStr} exceeded ${comboTargetTimeoutMs}ms timeout — falling back`
+          `Model ${modelStr} exceeded ${effectiveTimeoutMs}ms timeout — falling back`
         );
         timeoutController.abort(abortErr);
         // HTTP 504 (not proprietary 524): this is OmniRoute's own per-target timer.
@@ -154,7 +162,7 @@ export function buildTargetTimeoutRunner(deps: {
             }
           )
         );
-      }, comboTargetTimeoutMs);
+      }, effectiveTimeoutMs);
     });
     const targetWithSignal = {
       ...(target ?? {}),

src/lib/providers/requestDefaults.ts

@@ -218,6 +218,14 @@ export function normalizeProviderSpecificData(
     delete normalized.autoFetchModels;
   }
 
+  // Per-connection operator timeout — only persist a real integer.
+  if (
+    "timeoutMs" in normalized &&
+    (typeof normalized.timeoutMs !== "number" || !Number.isInteger(normalized.timeoutMs))
+  ) {
+    delete normalized.timeoutMs;
+  }
+
   if ("preset" in normalized) {
     const preset = provider === "openrouter" ? normalizeOpenRouterPreset(normalized.preset) : null;
     if (preset) {

src/shared/validation/providerSpecificData.ts

@@ -17,6 +17,7 @@ const CODEX_REASONING_EFFORT_VALUES = new Set(["none", "low", "medium", "high",
 const REQUEST_DEFAULT_SERVICE_TIER_VALUES = new Set(["default", "priority", "fast", "flex"]);
 const CODEX_FINGERPRINT_MODE_VALUES = new Set(["off", "device", "session", "full"]);
 const CACHE_PASSTHROUGH_VALUES = new Set(["strip", "openai-format", "claude-format"]);
+export const MAX_PROVIDER_SPECIFIC_TIMEOUT_MS = 86_400_000; // 24h — operator cap, anti-DoS
 
 // #6880 — per-connection prompt-cache capability override, extracted so
 // validateProviderSpecificData() stays under the complexity gate.
@@ -496,4 +497,23 @@ export function validateProviderSpecificData(
       });
     }
   }
+
+  // Per-connection operator timeout tier: a slow model must not monopolize an
+  // executor slot indefinitely. Bounded to 24h (anti-DoS); below 1ms is
+  // meaningless.
+  const timeoutMs = data.timeoutMs;
+  if (timeoutMs !== undefined && timeoutMs !== null) {
+    if (
+      typeof timeoutMs !== "number" ||
+      !Number.isInteger(timeoutMs) ||
+      timeoutMs < 1 ||
+      timeoutMs > MAX_PROVIDER_SPECIFIC_TIMEOUT_MS
+    ) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message: `providerSpecificData.timeoutMs must be an integer between 1 and ${MAX_PROVIDER_SPECIFIC_TIMEOUT_MS}`,
+        path: ["timeoutMs"],
+      });
+    }
+  }
 }

tests/unit/combo-target-timeout-runner.test.ts

@@ -204,3 +204,75 @@ test("drainLastTimeoutContexts returns and clears recorded contexts", async () =
   const second = drainLastTimeoutContexts();
   assert.equal(second.length, 0, "second drain must return empty");
 });
+
+test("resolveTargetTimeoutMs provided: uses per-target timeout when present", async () => {
+  let aborted = false;
+  const runner = buildTargetTimeoutRunner({
+    handleSingleModel: (_b, _m, target) =>
+      new Promise<Response>((resolve) => {
+        const sig = target?.modelAbortSignal ?? undefined;
+        sig?.addEventListener("abort", () => {
+          aborted = true;
+          resolve(new Response(null, { status: 599 }));
+        });
+      }),
+    comboTargetTimeoutMs: 20,
+    resolveTargetTimeoutMs: async (target) =>
+      target?.connectionId === "conn-1" ? 50 : undefined,
+    log: noopLog,
+  });
+  const res = await runner({}, "slow-model", {
+    connectionId: "conn-1",
+    modelAbortSignal: undefined as unknown as AbortSignal,
+  } as SingleModelTarget);
+  assert.equal(res.status, 504);
+  assert.equal(aborted, true);
+  const body = await res.json();
+  assert.equal(body?.error?.code, "combo_target_timeout");
+});
+
+test("resolveTargetTimeoutMs without connection: falls back to comboTargetTimeoutMs", async () => {
+  let aborted = false;
+  const runner = buildTargetTimeoutRunner({
+    handleSingleModel: (_b, _m, target) =>
+      new Promise<Response>((resolve) => {
+        target?.modelAbortSignal?.addEventListener("abort", () => {
+          aborted = true;
+          resolve(new Response(null, { status: 599 }));
+        });
+      }),
+    comboTargetTimeoutMs: 20,
+    resolveTargetTimeoutMs: async () => undefined,
+    log: noopLog,
+  });
+  const res = await runner({}, "slow-model", {
+    connectionId: "conn-unknown",
+    modelAbortSignal: undefined as unknown as AbortSignal,
+  } as SingleModelTarget);
+  assert.equal(res.status, 504);
+  assert.equal(aborted, true);
+});
+
+test("resolveTargetTimeoutMs extended: 50ms outlives the 20ms base (does not abort at 20ms)", async () => {
+  let resolvedWith = "";
+  const runner = buildTargetTimeoutRunner({
+    handleSingleModel: async (_b, _m, target) => {
+      await new Promise((resolve) => {
+        target?.modelAbortSignal?.addEventListener("abort", resolve);
+        setTimeout(resolve, 45);
+      });
+      resolvedWith = target?.modelAbortSignal?.aborted ? "aborted" : "completed";
+      return new Response(resolvedWith, { status: resolvedWith === "aborted" ? 599 : 200 });
+    },
+    comboTargetTimeoutMs: 20,
+    resolveTargetTimeoutMs: async (target) =>
+      target?.connectionId === "conn-1" ? 50 : undefined,
+    log: noopLog,
+  });
+  const res = await runner({}, "slow-model", {
+    connectionId: "conn-1",
+    modelAbortSignal: undefined as unknown as AbortSignal,
+  } as SingleModelTarget);
+  assert.equal(res.status, 200);
+  assert.equal(resolvedWith, "completed");
+});

tests/unit/provider-specific-data-schema.test.ts

@@ -390,3 +390,26 @@ test("provider schemas reject invalid quotaPerUnit values", () => {
   assert.equal(negative.success, false);
   assert.equal(string.success, false);
 });
+
+test("provider schemas accept integer timeoutMs in providerSpecificData", () => {
+  const created = createProviderSchema.safeParse({
+    provider: "openai",
+    apiKey: "token",
+    name: "OpenAI",
+    providerSpecificData: { timeoutMs: 1_800_000 },
+  });
+  const updated = updateProviderConnectionSchema.safeParse({
+    providerSpecificData: { timeoutMs: 900_000 },
+  });
+  assert.equal(created.success, true);
+  assert.equal(updated.success, true);
+});
+
+test("provider schemas reject invalid timeoutMs values", () => {
+  for (const bad of [-1, 0, 1.5, "60000", 86_400_001]) {
+    const updated = updateProviderConnectionSchema.safeParse({
+      providerSpecificData: { timeoutMs: bad },
+    });
+    assert.equal(updated.success, false, `timeoutMs=${String(bad)} must be rejected`);
+  }
+});

tests/unit/upstream-timeout-connection-tier.test.ts

@@ -0,0 +1,133 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import {
+  getExecutorTimeoutMs,
+  resolveConnectionTimeoutMs,
+  executeWithUpstreamStartTimeout,
+} from "../../open-sse/handlers/chatCore/upstreamTimeouts.ts";
+import { resolveTargetTimeoutMsForTarget } from "../../open-sse/services/combo.ts";
+
+const GLOBAL = 600_000;
+
+function fakeExecutor(timeoutMs?: number) {
+  return { getTimeoutMs: () => timeoutMs ?? GLOBAL };
+}
+
+test("resolveConnectionTimeoutMs: returns floored integer for valid values", () => {
+  assert.equal(resolveConnectionTimeoutMs({ timeoutMs: 1_800_000.9 }), 1_800_000);
+  assert.equal(resolveConnectionTimeoutMs({ timeoutMs: 1 }), 1);
+});
+
+test("resolveConnectionTimeoutMs: undefined for absent/invalid values", () => {
+  assert.equal(resolveConnectionTimeoutMs(undefined), undefined);
+  assert.equal(resolveConnectionTimeoutMs(null), undefined);
+  assert.equal(resolveConnectionTimeoutMs({}), undefined);
+  assert.equal(resolveConnectionTimeoutMs({ timeoutMs: 0 }), undefined);
+  assert.equal(resolveConnectionTimeoutMs({ timeoutMs: -5 }), undefined);
+  assert.equal(resolveConnectionTimeoutMs({ timeoutMs: "60000" }), undefined);
+  assert.equal(resolveConnectionTimeoutMs({ timeoutMs: 86_400_001 }), undefined);
+});
+
+test("getExecutorTimeoutMs: connection tier preempts model, provider and global", () => {
+  const executor = fakeExecutor(30_000);
+  // model override present (registry) + provider override + connection:
+  // connection wins
+  assert.equal(getExecutorTimeoutMs(executor, "openai", "gpt-5", 1_800_000), 1_800_000);
+});
+
+test("getExecutorTimeoutMs: invalid connection timeout falls through to model/provider/global", () => {
+  const executor = fakeExecutor(30_000);
+  assert.equal(getExecutorTimeoutMs(executor, "openai", "gpt-5", 0), 30_000);
+  assert.equal(getExecutorTimeoutMs(executor, "openai", "gpt-5", undefined), 30_000);
+  assert.equal(getExecutorTimeoutMs(fakeExecutor(), undefined, undefined, undefined), GLOBAL);
+});
+
+test("executeWithUpstreamStartTimeout: connection timeout aborts before the global", async () => {
+  let signalAbortedAt = Number.POSITIVE_INFINITY;
+  const started = Date.now();
+  const result = await executeWithUpstreamStartTimeout({
+    executor: fakeExecutor(600_000),
+    provider: "openai",
+    model: "gpt-5",
+    connectionTimeoutMs: 50,
+    signal: new AbortController().signal,
+    log: null,
+    execute: (signal) =>
+      new Promise((resolve) => {
+        signal.addEventListener("abort", () => {
+          signalAbortedAt = Date.now() - started;
+          resolve("aborted");
+        });
+      }),
+  });
+  assert.equal(result, "aborted");
+  assert.ok(signalAbortedAt < 1_000, `aborted at ${signalAbortedAt}ms, expected < 1000ms`);
+});
+test("connection timeout extraction matches execCreds.providerSpecificData shape", () => {
+  const execCreds = { providerSpecificData: { timeoutMs: 1_800_000, someOtherKey: 1 } };
+  assert.equal(resolveConnectionTimeoutMs(execCreds?.providerSpecificData), 1_800_000);
+  assert.equal(
+    resolveConnectionTimeoutMs({ providerSpecificData: undefined }?.providerSpecificData),
+    undefined
+  );
+});
+
+test("resolveTargetTimeoutMsForTarget: undefined without connectionId", async () => {
+  const result = await resolveTargetTimeoutMsForTarget(
+    null,
+    "fallback",
+    { enabled: false, budgetMs: 0 },
+    {
+      kind: "model",
+      stepId: "s",
+      executionKey: "k",
+      modelStr: "m",
+      provider: "p",
+      providerId: null,
+      connectionId: null,
+      weight: 1,
+      label: null,
+    }
+  );
+  assert.equal(result, undefined);
+});
+
+test("resolveTargetTimeoutMsForTarget: unknown connection -> undefined", async () => {
+  const result = await resolveTargetTimeoutMsForTarget(
+    null,
+    "fallback",
+    { enabled: false, budgetMs: 0 },
+    {
+      kind: "model",
+      stepId: "s",
+      executionKey: "k",
+      modelStr: "m",
+      provider: "p",
+      providerId: null,
+      connectionId: "conn-does-not-exist",
+      weight: 1,
+      label: null,
+    }
+  );
+  assert.equal(result, undefined);
+});
+
+test("resolveTargetTimeoutMsForTarget: connection without timeoutMs -> undefined", async () => {
+  const result = await resolveTargetTimeoutMsForTarget(
+    null,
+    "fallback",
+    { enabled: false, budgetMs: 0 },
+    {
+      kind: "model",
+      stepId: "s",
+      executionKey: "k",
+      modelStr: "m",
+      provider: "p",
+      providerId: null,
+      connectionId: "conn-no-timeout",
+      weight: 1,
+      label: null,
+    }
+  );
+  assert.equal(result, undefined);
+});