luulam/9router
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update: sync local changes with latest features

Browse Source
This commit is contained in:
luulam
2026-06-22 11:26:25 +07:00
parent 1cf55126f5
commit 98412aa0bb
16 changed files with 491 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
open-sse/executors/base.js
View File

@@ -116,8 +116,20 @@ export class BaseExecutor {
for (let urlIndex = 0; urlIndex < fallbackCount; urlIndex++) {
const url = this.buildUrl(model, stream, urlIndex, credentials);
const transformedBody = this.transformRequest(model, body, stream, credentials);
const headers = this.buildHeaders(credentials, stream);
// Merge extra body params from provider-specific data (request body takes priority)
let mergedBody = this.transformRequest(model, body, stream, credentials);
const extraBodyParams = credentials?.providerSpecificData?.bodyParams;
if (extraBodyParams && typeof extraBodyParams === "object" && !Array.isArray(extraBodyParams)) {
mergedBody = { ...extraBodyParams, ...mergedBody };
}
// Merge extra header params from provider-specific data (request headers take priority)
let headers = this.buildHeaders(credentials, stream);
const extraHeaderParams = credentials?.providerSpecificData?.headerParams;
if (extraHeaderParams && typeof extraHeaderParams === "object" && !Array.isArray(extraHeaderParams)) {
headers = { ...extraHeaderParams, ...headers };
}
if (!retryAttemptsByUrl[urlIndex]) retryAttemptsByUrl[urlIndex] = 0;
@@ -128,7 +140,7 @@ export class BaseExecutor {
const mergedSignal = signal ? AbortSignal.any([signal, connectCtrl.signal]) : connectCtrl.signal;
try {
const bodyStr = JSON.stringify(transformedBody);
const bodyStr = JSON.stringify(mergedBody);
const fetchT0 = Date.now();
dbg("FETCH", `${this.provider.toUpperCase()} → ${url} | body=${bodyStr.length}B | connectTimeout=${timeoutMs}ms`);
const response = await proxyAwareFetch(url, {
@@ -150,7 +162,7 @@ export class BaseExecutor {
continue;
}
return { response, url, headers, transformedBody };
return { response, url, headers, transformedBody: mergedBody };
} catch (error) {
clearTimeout(connectTimer);
lastError = error;

76
open-sse/executors/qoder.js
View File

@@ -121,6 +121,68 @@ function truncate(s, n) {
return s && s.length > n ? `${s.slice(0, n)}...` : s || "";
}
/**
* Parse Qoder-specific mid-stream error codes into user-friendly messages.
* Qoder embeds errors inside SSE envelope body as JSON: {"code":"112","message":"{...}"}
*
* Known codes:
* 112 → quota/billing exceeded (message contains pricingUrl)
* 113 → model not available for current plan
*
* @param {number} statusVal - Upstream status code from envelope
* @param {string} bodyStr - Raw body string from envelope
* @returns {{ statusCode: number, message: string, errorCode: string|null }}
*/
function parseQoderStreamError(statusVal, bodyStr) {
let code = null;
let innerMessage = "";
try {
const inner = JSON.parse(bodyStr);
code = String(inner.code || "");
innerMessage = inner.message || bodyStr;
// Code 112: quota exceeded — inner message is JSON with pricingUrl
if (code === "112") {
let pricingUrl = "";
try {
const msgObj = JSON.parse(innerMessage);
pricingUrl = msgObj.pricingUrl || "";
} catch { /* innerMessage is not JSON */ }
return {
statusCode: statusVal,
message: pricingUrl
? `Qoder quota exceeded. Upgrade your plan at: ${pricingUrl}`
: "Qoder quota exceeded. Please check your plan limits.",
errorCode: code,
};
}
// Code 113: model not available
if (code === "113") {
return {
statusCode: statusVal,
message: `Qoder model not available for your current plan. ${innerMessage}`,
errorCode: code,
};
}
} catch {
// bodyStr is not valid JSON — fall through to generic message
}
// Generic fallback
const statusLabel = statusVal >= 500 ? "server error"
: statusVal === 429 ? "rate limit exceeded"
: statusVal === 403 ? "permission error"
: statusVal === 401 ? "authentication error"
: "request error";
return {
statusCode: statusVal,
message: `Qoder ${statusLabel}: ${truncate(innerMessage || bodyStr, 200)}`,
errorCode: code,
};
}
/**
* Map the OpenAI-style request body into the exact shape Qoder expects.
*/
@@ -222,7 +284,7 @@ async function buildQoderRequestBody({ model, body, credentials, log, proxyOptio
* and re-emit as `data: <inner>\n\n`. Errors become `data: [DONE]\n\n` plus
* a synthetic OpenAI error chunk.
*/
async function wrapQoderSSE(response, model) {
async function wrapQoderSSE(response, model, midStreamError = {}) {
if (!response.ok || !response.body) return response;
const decoder = new TextDecoder();
@@ -304,13 +366,15 @@ async function wrapQoderSSE(response, model) {
const statusVal = typeof envelope.statusCodeValue === "number" ? envelope.statusCodeValue : 200;
const inner = typeof envelope.body === "string" ? envelope.body : "";
if (statusVal !== 200) {
const msg = inner || `upstream status ${statusVal}`;
const parsed = parseQoderStreamError(statusVal, inner);
// Store error in shared state so onStreamComplete can trigger cooldown
midStreamError.error = { status: parsed.statusCode, message: parsed.message, errorCode: parsed.errorCode };
const errChunk = JSON.stringify({
id: `qoder-error-${Date.now()}`,
object: "chat.completion.chunk",
created: Math.floor(Date.now() / 1000),
model,
choices: [{ index: 0, delta: { content: `\n[qoder error ${statusVal}: ${truncate(msg, 200)}]` }, finish_reason: "stop" }],
choices: [{ index: 0, delta: { content: `\n\n${parsed.message}` }, finish_reason: "stop" }],
});
controller.enqueue(encoder.encode(`data: ${errChunk}\n\n`));
controller.enqueue(encoder.encode("data: [DONE]\n\n"));
@@ -489,8 +553,9 @@ export class QoderExecutor extends BaseExecutor {
return { response, url, headers, transformedBody: payload };
}
const wrapped = await wrapQoderSSE(response, `qoder/${qoderKey}`);
return { response: wrapped, url, headers, transformedBody: payload };
const midStreamError = {};
const wrapped = await wrapQoderSSE(response, `qoder/${qoderKey}`, midStreamError);
return { response: wrapped, url, headers, transformedBody: payload, midStreamError };
}
// Qoder device tokens don't refresh through OAuth — the upstream returns
@@ -511,6 +576,7 @@ export default QoderExecutor;
// should import QoderExecutor and use its public methods.
export const __test__ = {
normalizeMessages,
parseQoderStreamError,
wrapQoderSSE,
buildQoderRequestBody,
};

11
open-sse/handlers/chatCore.js
View File

@@ -28,7 +28,7 @@ import { compressMessages, formatRtkLog } from "../rtk/index.js";
* @param {object} options.credentials - Provider credentials
* @param {string} options.sourceFormatOverride - Override detected source format (e.g. "openai-responses")
*/
export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, cavemanEnabled, cavemanLevel, sourceFormatOverride, providerThinking }) {
export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, onMidStreamError, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, cavemanEnabled, cavemanLevel, sourceFormatOverride, providerThinking }) {
const { provider, model } = modelInfo;
const requestStartTime = Date.now();
@@ -185,13 +185,14 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
}
// Execute request
let providerResponse, providerUrl, providerHeaders, finalBody;
let providerResponse, providerUrl, providerHeaders, finalBody, midStreamError;
try {
const result = await executor.execute({ model, body: translatedBody, stream, credentials, signal: streamController.signal, log, proxyOptions });
providerResponse = result.response;
providerUrl = result.url;
providerHeaders = result.headers;
finalBody = result.transformedBody;
midStreamError = result.midStreamError;
reqLogger.logTargetRequest(providerUrl, providerHeaders, finalBody);
} catch (error) {
trackPendingRequest(model, provider, connectionId, false, true);
@@ -258,7 +259,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
return createErrorResult(statusCode, errMsg, resetsAtMs);
}
const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess };
const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, midStreamError, onMidStreamError };
const appendLog = (extra) => appendRequestLog({ model, provider, connectionId, ...extra }).catch(() => { });
const trackDone = () => trackPendingRequest(model, provider, connectionId, false);
@@ -275,8 +276,8 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
return result;
}
// Streaming response
const { onStreamComplete } = buildOnStreamComplete({ ...sharedCtx });
// Streaming response (midStreamError + onMidStreamError already in sharedCtx)
const { onStreamComplete } = buildOnStreamComplete(sharedCtx);
return handleStreamingResponse({ ...sharedCtx, providerResponse, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, streamController, onStreamComplete });
}

17
open-sse/handlers/chatCore/nonStreamingHandler.js
View File

@@ -132,7 +132,7 @@ export function translateNonStreamingResponse(responseBody, targetFormat, source
/**
* Handle non-streaming response from provider.
*/
export async function handleNonStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, trackDone, appendLog }) {
export async function handleNonStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, trackDone, appendLog, midStreamError, onMidStreamError }) {
trackDone();
const contentType = providerResponse.headers.get("content-type") || "";
let responseBody;
@@ -144,6 +144,21 @@ export async function handleNonStreamingResponse({ providerResponse, provider, m
appendLog({ status: `FAILED ${HTTP_STATUS.BAD_GATEWAY}` });
return createErrorResult(HTTP_STATUS.BAD_GATEWAY, "Invalid SSE response for non-streaming request");
}
// Check for mid-stream errors detected during SSE parsing (e.g. Qoder quota exceeded).
// The upstream returned HTTP 200 but the SSE envelope contained a non-200 statusCodeValue.
// Without this check, the error message would be silently served as normal content with HTTP 200.
if (midStreamError?.error) {
const err = midStreamError.error;
if (typeof onMidStreamError === "function") {
onMidStreamError(err).catch(e => {
console.error("[MidStreamError] Failed to apply cooldown:", e.message);
});
}
appendLog({ status: `FAILED ${err.status || 429}` });
return createErrorResult(err.status || 429, err.message || "Upstream error detected in SSE stream");
}
responseBody = parsed;
} else {
try {

15
open-sse/handlers/chatCore/streamingHandler.js
View File

@@ -75,8 +75,10 @@ export function handleStreamingResponse({ providerResponse, provider, model, sou
/**
* Build onStreamComplete callback for streaming usage tracking.
* @param {object} options.midStreamError - Shared state object from executor (filled during streaming)
* @param {function} options.onMidStreamError - Callback to invoke when mid-stream error is detected
*/
export function buildOnStreamComplete({ provider, model, connectionId, apiKey, requestStartTime, body, stream, finalBody, translatedBody, clientRawRequest }) {
export function buildOnStreamComplete({ provider, model, connectionId, apiKey, requestStartTime, body, stream, finalBody, translatedBody, clientRawRequest, midStreamError, onMidStreamError }) {
const streamDetailId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`;
const onStreamComplete = (contentObj, usage, ttftAt) => {
@@ -87,6 +89,15 @@ export function buildOnStreamComplete({ provider, model, connectionId, apiKey, r
const safeContent = contentObj?.content || "[Empty streaming response]";
const safeThinking = contentObj?.thinking || null;
// Check if a mid-stream error was detected during streaming (e.g. Qoder quota exceeded)
// The stream already returned HTTP 200, so the pre-stream error path in chatCore didn't fire.
// Apply cooldown here so the account gets locked for subsequent requests.
if (midStreamError?.error && typeof onMidStreamError === "function") {
onMidStreamError(midStreamError.error).catch(err => {
console.error("[MidStreamError] Failed to apply cooldown:", err.message);
});
}
saveRequestDetail(buildRequestDetail({
provider, model, connectionId,
latency,
@@ -95,7 +106,7 @@ export function buildOnStreamComplete({ provider, model, connectionId, apiKey, r
providerRequest: finalBody || translatedBody || null,
providerResponse: safeContent,
response: { content: safeContent, thinking: safeThinking, type: "streaming" },
status: "success"
status: midStreamError?.error ? "error" : "success"
}, { id: streamDetailId })).catch(err => {
console.error("[RequestDetail] Failed to update streaming content:", err.message);
});

12
open-sse/services/accountFallback.js
View File

@@ -18,9 +18,10 @@ export function getQuotaCooldown(backoffLevel = 0) {
* @param {number} status - HTTP status code
* @param {string} errorText - Error message text
* @param {number} backoffLevel - Current backoff level for exponential backoff
* @param {number} [fixedCooldownMs=0] - When >0, override all cooldowns with this fixed value (ms)
* @returns {{ shouldFallback: boolean, cooldownMs: number, newBackoffLevel?: number }}
*/
export function checkFallbackError(status, errorText, backoffLevel = 0) {
export function checkFallbackError(status, errorText, backoffLevel = 0, fixedCooldownMs = 0) {
const lowerError = errorText
? (typeof errorText === "string" ? errorText : JSON.stringify(errorText)).toLowerCase()
: "";
@@ -28,6 +29,9 @@ export function checkFallbackError(status, errorText, backoffLevel = 0) {
for (const rule of ERROR_RULES) {
// Text-based rule: match substring in error message
if (rule.text && lowerError && lowerError.includes(rule.text)) {
if (fixedCooldownMs > 0) {
return { shouldFallback: true, cooldownMs: fixedCooldownMs, newBackoffLevel: 0 };
}
if (rule.backoff) {
const newLevel = Math.min(backoffLevel + 1, BACKOFF_CONFIG.maxLevel);
return { shouldFallback: true, cooldownMs: getQuotaCooldown(newLevel), newBackoffLevel: newLevel };
@@ -37,6 +41,9 @@ export function checkFallbackError(status, errorText, backoffLevel = 0) {
// Status-based rule: match HTTP status code
if (rule.status && rule.status === status) {
if (fixedCooldownMs > 0) {
return { shouldFallback: true, cooldownMs: fixedCooldownMs, newBackoffLevel: 0 };
}
if (rule.backoff) {
const newLevel = Math.min(backoffLevel + 1, BACKOFF_CONFIG.maxLevel);
return { shouldFallback: true, cooldownMs: getQuotaCooldown(newLevel), newBackoffLevel: newLevel };
@@ -46,7 +53,8 @@ export function checkFallbackError(status, errorText, backoffLevel = 0) {
}
// Default: transient cooldown for any unmatched error
return { shouldFallback: true, cooldownMs: TRANSIENT_COOLDOWN_MS };
const defaultCooldown = fixedCooldownMs > 0 ? fixedCooldownMs : TRANSIENT_COOLDOWN_MS;
return { shouldFallback: true, cooldownMs: defaultCooldown, newBackoffLevel: 0 };
}
/**