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

feat: pre-request token validation, mid-stream error handling, and usage stats improvements

Browse Source 
- Qoder: handle mid-stream errors by returning proper error Response instead of embedding in stream
- Qoder: add refreshCredentials() to validate token via quota endpoint before requests
- chatCore: validate and refresh provider tokens before sending chat requests
- chatCore: bail early with 401 on unrecoverable token refresh errors
- Usage stats: track apiKey, comboName, fallbackHistory in request details
- Dashboard: improve Combos, Endpoint, Provider, Usage, and RequestDetails pages
- API keys route: upsert logic with provider_type support
- DB repos: usageRepo query improvements, requestDetailsRepo pagination, apiKeysRepo updates
This commit is contained in:
luulam
2026-06-29 10:08:07 +07:00
parent 1fe8115dad
commit 2305e26e25
27 changed files with 1433 additions and 459 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -13,7 +13,7 @@ import { HTTP_STATUS } from "../config/runtimeConfig.js";
import { handleBypassRequest } from "../utils/bypassHandler.js";
import { trackPendingRequest, appendRequestLog, saveRequestDetail } from "@/lib/usageDb.js";
import { getExecutor } from "../executors/index.js";
import { buildRequestDetail, extractRequestConfig } from "./chatCore/requestDetail.js";
import { buildRequestDetail, extractRequestConfig, saveUsageStats } from "./chatCore/requestDetail.js";
import { handleForcedSSEToJson } from "./chatCore/sseToJsonHandler.js";
import { handleNonStreamingResponse } from "./chatCore/nonStreamingHandler.js";
import { handleStreamingResponse, buildOnStreamComplete } from "./chatCore/streamingHandler.js";
@@ -34,7 +34,7 @@ import { prefetchRemoteImages } from "../translator/concerns/prefetch.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, onMidStreamError, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, headroomEnabled, headroomUrl, headroomCompressUserMessages, cavemanEnabled, cavemanLevel, ponytailEnabled, ponytailLevel, sourceFormatOverride, providerThinking }) {
export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, onMidStreamError, clientRawRequest, connectionId, userAgent, apiKey, comboName, fallbackHistory, ccFilterNaming, rtkEnabled, headroomEnabled, headroomUrl, headroomCompressUserMessages, cavemanEnabled, cavemanLevel, ponytailEnabled, ponytailLevel, sourceFormatOverride, providerThinking }) {
const { provider, model } = modelInfo;
const requestStartTime = Date.now();
@@ -185,6 +185,36 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
const msgCount = translatedBody.messages?.length || translatedBody.input?.length || translatedBody.contents?.length || translatedBody.request?.contents?.length || 0;
log?.debug?.("REQUEST", `${provider.toUpperCase()} | ${model} | ${msgCount} msgs`);
// Validate credentials before sending request - try refresh if needed
if (!executor.noAuth && credentials?.accessToken) {
const needsRefresh = executor.needsRefresh?.(credentials) ?? shouldRefreshCredentials(provider, credentials);
if (needsRefresh) {
try {
const newCredentials = await refreshWithRetry(() => executor.refreshCredentials(credentials, log), 3, log);
if (newCredentials?.accessToken || newCredentials?.copilotToken) {
log?.info?.("TOKEN", `${provider.toUpperCase()} | refreshed before request`);
Object.assign(credentials, newCredentials);
if (onCredentialsRefreshed) {
try { await onCredentialsRefreshed(newCredentials); } catch (e) { log?.warn?.("TOKEN", `onCredentialsRefreshed failed: ${e.message}`); }
}
} else if (newCredentials?.error === "unrecoverable_refresh_error") {
const msg = newCredentials.message || `${provider} token invalid; reconnect the account`;
log?.warn?.("TOKEN", `${provider.toUpperCase()} | unrecoverable refresh error: ${msg}`);
trackPendingRequest(model, provider, connectionId, false, true);
return createErrorResult(HTTP_STATUS.UNAUTHORIZED, msg);
} else {
log?.warn?.("TOKEN", `${provider.toUpperCase()} | refresh failed before request`);
trackPendingRequest(model, provider, connectionId, false, true);
return createErrorResult(HTTP_STATUS.UNAUTHORIZED, `${provider} token refresh failed. Please reconnect the account.`);
}
} catch (e) {
log?.warn?.("TOKEN", `${provider.toUpperCase()} | refresh threw before request: ${e.message}`);
trackPendingRequest(model, provider, connectionId, false, true);
return createErrorResult(HTTP_STATUS.UNAUTHORIZED, `${provider} token refresh failed: ${e.message}`);
}
}
}
const streamController = createStreamController({
onDisconnect: (reason) => {
trackPendingRequest(model, provider, connectionId, false);
@@ -241,7 +271,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
trackPendingRequest(model, provider, connectionId, false, true);
appendRequestLog({ model, provider, connectionId, status: `FAILED ${error.name === "AbortError" ? 499 : HTTP_STATUS.BAD_GATEWAY}` }).catch(() => { });
saveRequestDetail(buildRequestDetail({
provider, model, connectionId,
provider, model, connectionId, apiKey, comboName, fallbackHistory,
latency: { ttft: 0, total: Date.now() - requestStartTime },
tokens: { prompt_tokens: 0, completion_tokens: 0 },
request: extractRequestConfig(body, stream),
@@ -249,6 +279,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
response: { error: error.message || String(error), status: error.name === "AbortError" ? 499 : 502, thinking: null },
status: "error"
})).catch(() => { });
saveUsageStats({ provider, model, tokens: { prompt_tokens: 0, completion_tokens: 0 }, connectionId, apiKey, endpoint: clientRawRequest?.endpoint, comboName, fallbackHistory, status: "error", label: "ERROR" });
if (error.name === "AbortError") {
streamController.handleError(error);
@@ -287,7 +318,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
const { statusCode, message, resetsAtMs } = await parseUpstreamError(providerResponse, executor);
appendRequestLog({ model, provider, connectionId, status: `FAILED ${statusCode}` }).catch(() => { });
saveRequestDetail(buildRequestDetail({
provider, model, connectionId,
provider, model, connectionId, apiKey, comboName, fallbackHistory,
latency: { ttft: 0, total: Date.now() - requestStartTime },
tokens: { prompt_tokens: 0, completion_tokens: 0 },
request: extractRequestConfig(body, stream),
@@ -295,6 +326,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
response: { error: message, status: statusCode, thinking: null },
status: "error"
})).catch(() => { });
saveUsageStats({ provider, model, tokens: { prompt_tokens: 0, completion_tokens: 0 }, connectionId, apiKey, endpoint: clientRawRequest?.endpoint, comboName, fallbackHistory, status: "error", label: "ERROR" });
const errMsg = formatProviderError(new Error(message), provider, model, statusCode);
console.log(`${COLORS.red}[ERROR] ${errMsg}${COLORS.reset}`);
@@ -302,7 +334,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, midStreamError, onMidStreamError };
const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, comboName, fallbackHistory, clientRawRequest, onRequestSuccess, midStreamError, onMidStreamError };
const appendLog = (extra) => appendRequestLog({ model, provider, connectionId, ...extra }).catch(() => { });
const trackDone = () => trackPendingRequest(model, provider, connectionId, false);

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

@@ -143,7 +143,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, midStreamError, onMidStreamError }) {
export async function handleNonStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, comboName, fallbackHistory, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, trackDone, appendLog, midStreamError, onMidStreamError }) {
trackDone();
const contentType = providerResponse.headers.get("content-type") || "";
let responseBody;
@@ -189,7 +189,7 @@ export async function handleNonStreamingResponse({ providerResponse, provider, m
const usage = extractUsageFromResponse(responseBody);
appendLog({ tokens: usage, status: "200 OK" });
saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint });
saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint, comboName });
const translatedResponse = needsTranslation(targetFormat, sourceFormat)
? translateNonStreamingResponse(responseBody, targetFormat, sourceFormat)
@@ -234,7 +234,7 @@ export async function handleNonStreamingResponse({ providerResponse, provider, m
const totalLatency = Date.now() - requestStartTime;
saveRequestDetail(buildRequestDetail({
provider, model, connectionId,
provider, model, connectionId, apiKey, comboName, fallbackHistory,
latency: { ttft: totalLatency, total: totalLatency },
tokens: usage || { prompt_tokens: 0, completion_tokens: 0 },
request: extractRequestConfig(body, stream),

12
open-sse/handlers/chatCore/requestDetail.js
View File

@@ -60,6 +60,9 @@ export function buildRequestDetail(base, overrides = {}) {
provider: base.provider || "unknown",
model: base.model || "unknown",
connectionId: base.connectionId || undefined,
apiKey: base.apiKey || undefined,
comboName: base.comboName || null,
fallbackHistory: base.fallbackHistory || null,
timestamp: new Date().toISOString(),
latency: base.latency || { ttft: 0, total: 0 },
tokens: base.tokens || { prompt_tokens: 0, completion_tokens: 0 },
@@ -72,13 +75,13 @@ export function buildRequestDetail(base, overrides = {}) {
};
}
export function saveUsageStats({ provider, model, tokens, connectionId, apiKey, endpoint, label = "USAGE" }) {
export function saveUsageStats({ provider, model, tokens, connectionId, apiKey, endpoint, comboName, fallbackHistory, status = "ok", label = "USAGE" }) {
if (!tokens || typeof tokens !== "object") return;
const inTokens = tokens.input_tokens ?? tokens.prompt_tokens ?? 0;
const outTokens = tokens.output_tokens ?? tokens.completion_tokens ?? 0;
if (inTokens === 0 && outTokens === 0) return;
if (inTokens === 0 && outTokens === 0 && status !== "error") return;
// Extract cache/reasoning tokens (unified from different formats)
const cacheRead = tokens.cache_read_input_tokens || tokens.cached_tokens || tokens.prompt_tokens_details?.cached_tokens || 0;
@@ -112,6 +115,9 @@ export function saveUsageStats({ provider, model, tokens, connectionId, apiKey,
timestamp: new Date().toISOString(),
connectionId: connectionId || undefined,
apiKey: apiKey || undefined,
endpoint: endpoint || null
endpoint: endpoint || null,
comboName: comboName || undefined,
fallbackHistory: fallbackHistory || undefined,
status: status || "ok",
}).catch(() => {});
}

8
open-sse/handlers/chatCore/sseToJsonHandler.js
View File

@@ -102,7 +102,7 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) {
* Handle case: provider forced streaming but client wants JSON.
* Supports both Codex/Responses API SSE and standard Chat Completions SSE.
*/
export async function handleForcedSSEToJson({ providerResponse, sourceFormat, provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, trackDone, appendLog }) {
export async function handleForcedSSEToJson({ providerResponse, sourceFormat, provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, comboName, fallbackHistory, clientRawRequest, onRequestSuccess, trackDone, appendLog }) {
const contentType = providerResponse.headers.get("content-type") || "";
const isSSE = contentType.includes("text/event-stream") || (contentType === "" && isResponsesProvider(provider));
if (!isSSE) return null; // not handled here
@@ -110,7 +110,7 @@ export async function handleForcedSSEToJson({ providerResponse, sourceFormat, pr
trackDone();
const ctx = {
provider, model, connectionId,
provider, model, connectionId, apiKey, comboName, fallbackHistory,
request: extractRequestConfig(body, stream),
providerRequest: finalBody || translatedBody || null
};
@@ -124,7 +124,7 @@ export async function handleForcedSSEToJson({ providerResponse, sourceFormat, pr
const usage = jsonResponse.usage || {};
appendLog({ tokens: usage, status: "200 OK" });
saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint });
saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint, comboName });
const { msgItem, textContent } = pickAssistantMessageForChatCompletion(jsonResponse.output);
const totalLatency = Date.now() - requestStartTime;
@@ -200,7 +200,7 @@ export async function handleForcedSSEToJson({ providerResponse, sourceFormat, pr
const usage = parsed.usage || {};
appendLog({ tokens: usage, status: "200 OK" });
saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint });
saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint, comboName });
const totalLatency = Date.now() - requestStartTime;
saveRequestDetail(buildRequestDetail({

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

@@ -22,7 +22,7 @@ const CODEX_SOURCE_TO_TARGET = {
/**
* Determine which SSE transform stream to use based on provider/format.
*/
function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey }) {
function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey, endpoint, comboName }) {
const isDroidCLI = userAgent?.toLowerCase().includes("droid") || userAgent?.toLowerCase().includes("codex-cli");
// Responses-API providers (e.g. codex) emit Responses SSE → translate into client format
const isResponsesProvider = PROVIDERS[provider]?.format === FORMATS.OPENAI_RESPONSES;
@@ -30,23 +30,23 @@ function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent,
if (needsCodexTranslation) {
const codexTarget = CODEX_SOURCE_TO_TARGET[sourceFormat] || FORMATS.OPENAI;
return createSSETransformStreamWithLogger(FORMATS.OPENAI_RESPONSES, codexTarget, provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey);
return createSSETransformStreamWithLogger(FORMATS.OPENAI_RESPONSES, codexTarget, provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey, endpoint, comboName);
}
if (needsTranslation(targetFormat, sourceFormat)) {
return createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey);
return createSSETransformStreamWithLogger(targetFormat, sourceFormat, provider, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey, endpoint, comboName);
}
return createPassthroughStreamWithLogger(provider, reqLogger, model, connectionId, body, onStreamComplete, apiKey);
return createPassthroughStreamWithLogger(provider, reqLogger, model, connectionId, body, onStreamComplete, apiKey, endpoint, comboName);
}
/**
* Handle streaming response — pipe provider SSE through transform stream to client.
*/
export function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete }) {
export function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, comboName, fallbackHistory, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete }) {
if (onRequestSuccess) onRequestSuccess();
const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey });
const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, model, connectionId, body, onStreamComplete, apiKey, endpoint: clientRawRequest?.endpoint, comboName });
// 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;
@@ -56,7 +56,7 @@ export function handleStreamingResponse({ providerResponse, provider, model, sou
const streamDetailId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`;
saveRequestDetail(buildRequestDetail({
provider, model, connectionId,
provider, model, connectionId, apiKey, comboName, fallbackHistory,
latency: { ttft: 0, total: Date.now() - requestStartTime },
tokens: { prompt_tokens: 0, completion_tokens: 0 },
request: extractRequestConfig(body, stream),
@@ -79,7 +79,7 @@ export function handleStreamingResponse({ providerResponse, provider, model, sou
* @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, midStreamError, onMidStreamError }) {
export function buildOnStreamComplete({ provider, model, connectionId, apiKey, comboName, fallbackHistory, requestStartTime, body, stream, finalBody, translatedBody, clientRawRequest, midStreamError, onMidStreamError }) {
const streamDetailId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`;
const onStreamComplete = (contentObj, usage, ttftAt) => {
@@ -100,7 +100,7 @@ export function buildOnStreamComplete({ provider, model, connectionId, apiKey, r
}
saveRequestDetail(buildRequestDetail({
provider, model, connectionId,
provider, model, connectionId, apiKey, comboName, fallbackHistory,
latency,
tokens: usage || { prompt_tokens: 0, completion_tokens: 0 },
request: extractRequestConfig(body, stream),