diff --git a/open-sse/config/providers.js b/open-sse/config/providers.js index 14e73f29..1dbc3df7 100644 --- a/open-sse/config/providers.js +++ b/open-sse/config/providers.js @@ -101,6 +101,9 @@ export const PROVIDERS = { baseUrl: "https://api3.qoder.sh/algo/api/v2/service/pro/sse/agent_chat_generation", format: "openai", headers: {}, + // Reasoning models think long before first byte; raise both timeouts. + timeoutMs: 120000, + stallTimeoutMs: 120000, }, antigravity: { baseUrls: [ diff --git a/open-sse/executors/qoder.js b/open-sse/executors/qoder.js index 3be1f00f..fa9bcbad 100644 --- a/open-sse/executors/qoder.js +++ b/open-sse/executors/qoder.js @@ -28,6 +28,7 @@ import { createHash } from "crypto"; import { BaseExecutor } from "./base.js"; import { PROVIDERS } from "../config/providers.js"; import { proxyAwareFetch } from "../utils/proxyFetch.js"; +import { FETCH_CONNECT_TIMEOUT_MS } from "../config/runtimeConfig.js"; import { QODER_CHAT_URL_ENCODED, QODER_MODEL_MAP, @@ -407,15 +408,21 @@ export class QoderExecutor extends BaseExecutor { ...cosyHeaders, }; + // Abort if upstream doesn't return response headers within connect timeout. + const timeoutMs = this.config?.timeoutMs || FETCH_CONNECT_TIMEOUT_MS; + const connectCtrl = new AbortController(); + const connectTimer = setTimeout(() => connectCtrl.abort(new Error("fetch connect timeout")), timeoutMs); + const mergedSignal = signal ? AbortSignal.any([signal, connectCtrl.signal]) : connectCtrl.signal; + let response; try { response = await proxyAwareFetch( url, - { method: "POST", headers, body: encodedBodyBuf, signal }, + { method: "POST", headers, body: encodedBodyBuf, signal: mergedSignal }, proxyOptions, ); - } catch (err) { - throw err; + } finally { + clearTimeout(connectTimer); } if (!response.ok) { diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index 89339587..f96bb9bd 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/open-sse/handlers/chatCore/streamingHandler.js @@ -2,6 +2,8 @@ import { FORMATS } from "../../translator/formats.js"; import { needsTranslation } from "../../translator/index.js"; import { createSSETransformStreamWithLogger, createPassthroughStreamWithLogger } from "../../utils/stream.js"; import { pipeWithDisconnect } from "../../utils/streamHandler.js"; +import { PROVIDERS } from "../../config/providers.js"; +import { STREAM_STALL_TIMEOUT_MS } from "../../config/runtimeConfig.js"; import { buildAbortedResponsesTerminalBytes } from "../../utils/responsesStreamHelpers.js"; import { buildRequestDetail, extractRequestConfig, saveUsageStats } from "./requestDetail.js"; import { saveRequestDetail } from "@/lib/usageDb.js"; @@ -48,7 +50,8 @@ export function handleStreamingResponse({ providerResponse, provider, model, sou // Responses passthrough: synthesize response.failed + [DONE] if the stream aborts/stalls before a terminal event const isResponsesPassthrough = sourceFormat === FORMATS.OPENAI_RESPONSES && targetFormat === FORMATS.OPENAI_RESPONSES; const onAbortTerminal = isResponsesPassthrough ? buildAbortedResponsesTerminalBytes : null; - const transformedBody = pipeWithDisconnect(providerResponse, transformStream, streamController, onAbortTerminal); + const stallTimeoutMs = PROVIDERS[provider]?.stallTimeoutMs || STREAM_STALL_TIMEOUT_MS; + const transformedBody = pipeWithDisconnect(providerResponse, transformStream, streamController, onAbortTerminal, stallTimeoutMs); const streamDetailId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`; saveRequestDetail(buildRequestDetail({ diff --git a/open-sse/utils/streamHandler.js b/open-sse/utils/streamHandler.js index 35ef5e4a..6c8b031a 100644 --- a/open-sse/utils/streamHandler.js +++ b/open-sse/utils/streamHandler.js @@ -184,7 +184,7 @@ export function createDisconnectAwareStream(transformStream, streamController, o * @param {TransformStream} transformStream - Transform stream for SSE * @param {object} streamController - Stream controller from createStreamController */ -export function pipeWithDisconnect(providerResponse, transformStream, streamController, onAbortTerminal = null) { +export function pipeWithDisconnect(providerResponse, transformStream, streamController, onAbortTerminal = null, stallTimeoutMs = STREAM_STALL_TIMEOUT_MS) { let stallTimer = null; let chunkCount = 0; let totalBytes = 0; @@ -198,10 +198,10 @@ export function pipeWithDisconnect(providerResponse, transformStream, streamCont clearStall(); stallTimer = setTimeout(() => { stallTimer = null; - dbg(tag, `STALL TIMEOUT ${STREAM_STALL_TIMEOUT_MS}ms | chunks=${chunkCount} | bytes=${totalBytes} | sinceLast=${Date.now() - lastChunkAt}ms`); + dbg(tag, `STALL TIMEOUT ${stallTimeoutMs}ms | chunks=${chunkCount} | bytes=${totalBytes} | sinceLast=${Date.now() - lastChunkAt}ms`); streamController.handleError?.(new Error("stream stall timeout")); streamController.abort?.(); - }, STREAM_STALL_TIMEOUT_MS); + }, stallTimeoutMs); }; // Wrap controller so every termination path clears the stall timer. @@ -218,7 +218,7 @@ export function pipeWithDisconnect(providerResponse, transformStream, streamCont }; armStall(); - dbg(tag, `pipe start | stallTimeout=${STREAM_STALL_TIMEOUT_MS}ms`); + dbg(tag, `pipe start | stallTimeout=${stallTimeoutMs}ms`); const upstreamTap = new TransformStream({ transform(chunk, controller) {