From 81a47f36c6dc271b13824ba0db367c588a14d63e Mon Sep 17 00:00:00 2001 From: luulam Date: Thu, 17 Sep 2026 14:03:59 +0700 Subject: [PATCH] fix(combo): keep nested combos as single units and stop 0-token detail rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nested combos (comboA lists comboB, comboC, …) now stay one slot each: the inner combo always runs as fallback to produce a single answer. Failed hops are no longer written to Details/usage, and streaming no longer inserts a 0-token placeholder row. - chat.js: comboStack cycle detection; nested combos forced to fallback; persistUsage="success-only" for combo hops - combo.js: discardResponse() cancels unused bodies (fusion timeout / fallback) so dropped streams fire onStreamComplete; getComboModelsFromData keeps nested names and honors enabled=false - requestDetail.js: tokensForDetail() canonicalizes Claude/Gemini usage; shouldPersistRequestDetail() skips streaming-start and non-success hops - streamingHandler.js: drop the 0-token streaming placeholder write - RequestDetailsTab.js: read Gemini/Claude token names; show "streaming" status in amber - tests: add combo-nested.test.js (13 cases) - gitignore: ignore local .vitest/ artifacts Co-authored-by: CommandCodeBot --- .gitignore | 4 +- CHANGELOG.md | 1 + open-sse/handlers/chatCore.js | 50 +++---- .../handlers/chatCore/nonStreamingHandler.js | 40 +++--- open-sse/handlers/chatCore/requestDetail.js | 23 ++++ .../handlers/chatCore/streamingHandler.js | 96 +++++-------- open-sse/services/combo.js | 57 ++++++-- .../usage/components/RequestDetailsTab.js | 22 ++- src/sse/handlers/chat.js | 36 +++-- src/sse/services/model.js | 19 +-- tests/unit/combo-nested.test.js | 128 ++++++++++++++++++ 11 files changed, 332 insertions(+), 144 deletions(-) create mode 100644 tests/unit/combo-nested.test.js diff --git a/.gitignore b/.gitignore index fbab4ac0..0a48e000 100644 --- a/.gitignore +++ b/.gitignore @@ -94,4 +94,6 @@ graphify-out/* .commandcode/ # Pi subagent run artifacts -.pi-subagents/ \ No newline at end of file +.pi-subagents/ +# Vitest local artifacts +.vitest/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 24fa7652..a79c3171 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ - **Providers**: wire the dead "Fetch Models" button on compatible nodes to the live upstream catalog, de-duplicating against already-added models ## Fixes +- **Usage**: nested combos (comboA lists comboB, comboC, …) stay one slot each — the inner combo always runs as fallback to produce a single answer, failed hops are not written to Details/usage, and streaming no longer inserts a 0-token placeholder. One user message against a nested fallback combo is one request row with real tokens; fusion of nested combos is N panel slots + judge, not every nested leaf - **Models**: persist per-model capability assertions for custom and compatible providers and honor them everywhere — unsupported media is stripped on the chat path, `/v1/models` and `/api/models` report what the user asserted, and thinking translation follows it (asserting `reasoning:false` now actually strips thinking fields, `reasoning:true` emits them) - **Models**: partial capability edits merge instead of overwriting, so toggling vision off no longer erases a stored reasoning assertion - **Capabilities**: keep server-injected readers (synced catalog, user-asserted capabilities) in process-wide state — Next.js compiles startup and each API route into separate bundles with their own module instances, so a boot-time install was invisible to every request handler and the models.dev catalog contributed nothing to upstream requests since 0532f00d diff --git a/open-sse/handlers/chatCore.js b/open-sse/handlers/chatCore.js index 75c13f4e..82a2b9f7 100644 --- a/open-sse/handlers/chatCore.js +++ b/open-sse/handlers/chatCore.js @@ -14,7 +14,7 @@ import { handleBypassRequest } from "../utils/bypassHandler.js"; import { trackPendingRequest, saveRequestDetail } from "@/lib/usageDb.js"; import { getExecutor } from "../executors/index.js"; import { supportsGrokCliReasoningEffort } from "../config/grokCli.js"; -import { buildRequestDetail, extractRequestConfig } from "./chatCore/requestDetail.js"; +import { buildRequestDetail, extractRequestConfig, shouldPersistRequestDetail } from "./chatCore/requestDetail.js"; import { handleForcedSSEToJson } from "./chatCore/sseToJsonHandler.js"; import { handleNonStreamingResponse } from "./chatCore/nonStreamingHandler.js"; import { handleStreamingResponse, buildOnStreamComplete } from "./chatCore/streamingHandler.js"; @@ -59,7 +59,7 @@ export function stripContinuityFields(body) { return body; } -export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, headroomEnabled, headroomUrl, headroomCompressUserMessages, headroomTimeoutMs, cavemanEnabled, cavemanLevel, ponytailEnabled, ponytailLevel, pxpipeEnabled, pxpipeMinChars, pxpipeTimeoutMs, pxpipeTransform, onPxpipeEvent, sourceFormatOverride, providerThinking, capsOverride = null, streamErrorPatterns = null }) { +export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, headroomEnabled, headroomUrl, headroomCompressUserMessages, headroomTimeoutMs, cavemanEnabled, cavemanLevel, ponytailEnabled, ponytailLevel, pxpipeEnabled, pxpipeMinChars, pxpipeTimeoutMs, pxpipeTransform, onPxpipeEvent, sourceFormatOverride, providerThinking, capsOverride = null, streamErrorPatterns = null, persistUsage = "all" }) { const { provider, model } = modelInfo; const requestStartTime = Date.now(); // Stable per-session color so all lines of one CLI conversation share a tag @@ -377,16 +377,18 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred reqLogger.logTargetRequest(providerUrl, providerHeaders, finalBody); } catch (error) { trackPendingRequest(model, provider, connectionId, false, true); - saveRequestDetail(buildRequestDetail({ - provider, model, connectionId, - latency: { ttft: 0, total: Date.now() - requestStartTime }, - tokens: { prompt_tokens: 0, completion_tokens: 0 }, - request: extractRequestConfig(body, stream), - providerRequest: translatedBody || null, - response: { error: error.message || String(error), status: error.name === "AbortError" ? 499 : 502, thinking: null }, - pxpipe: pxpipeSummary, - status: "error" - })).catch(() => { }); + if (shouldPersistRequestDetail(persistUsage, "error")) { + saveRequestDetail(buildRequestDetail({ + provider, model, connectionId, + latency: { ttft: 0, total: Date.now() - requestStartTime }, + tokens: { prompt_tokens: 0, completion_tokens: 0 }, + request: extractRequestConfig(body, stream), + providerRequest: translatedBody || null, + response: { error: error.message || String(error), status: error.name === "AbortError" ? 499 : 502, thinking: null }, + pxpipe: pxpipeSummary, + status: "error" + })).catch(() => { }); + } if (error.name === "AbortError") { streamController.handleError(error); @@ -451,16 +453,18 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred if (!providerResponse.ok) { trackPendingRequest(model, provider, connectionId, false, true); const { statusCode, message, resetsAtMs } = await parseUpstreamError(providerResponse, executor); - saveRequestDetail(buildRequestDetail({ - provider, model, connectionId, - latency: { ttft: 0, total: Date.now() - requestStartTime }, - tokens: { prompt_tokens: 0, completion_tokens: 0 }, - request: extractRequestConfig(body, stream), - providerRequest: finalBody || translatedBody || null, - response: { error: message, status: statusCode, thinking: null }, - pxpipe: pxpipeSummary, - status: "error" - })).catch(() => { }); + if (shouldPersistRequestDetail(persistUsage, "error")) { + saveRequestDetail(buildRequestDetail({ + provider, model, connectionId, + latency: { ttft: 0, total: Date.now() - requestStartTime }, + tokens: { prompt_tokens: 0, completion_tokens: 0 }, + request: extractRequestConfig(body, stream), + providerRequest: finalBody || translatedBody || null, + response: { error: message, status: statusCode, thinking: null }, + pxpipe: pxpipeSummary, + status: "error" + })).catch(() => { }); + } const errMsg = formatProviderError(new Error(message), provider, model, statusCode); if (log?.errorLine) { @@ -472,7 +476,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred } const appendLog = () => {}; // request log derived from usageHistory; kept as no-op seam for handlers - const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, pxpipe: pxpipeSummary, reqTag, log, streamErrorPatterns }; + const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, pxpipe: pxpipeSummary, reqTag, log, streamErrorPatterns, persistUsage }; // Early-peek streaming responses for configured in-stream error patterns. // Some upstreams fail INSIDE a 200 SSE stream; without this the failure is diff --git a/open-sse/handlers/chatCore/nonStreamingHandler.js b/open-sse/handlers/chatCore/nonStreamingHandler.js index ce713f6a..bc1fc858 100644 --- a/open-sse/handlers/chatCore/nonStreamingHandler.js +++ b/open-sse/handlers/chatCore/nonStreamingHandler.js @@ -6,7 +6,7 @@ import { addBufferToUsage, filterUsageForFormat } from "../../utils/usageTrackin import { createErrorResult } from "../../utils/error.js"; import { HTTP_STATUS } from "../../config/runtimeConfig.js"; import { parseSSEToOpenAIResponse } from "./sseToJsonHandler.js"; -import { buildRequestDetail, extractRequestConfig, extractUsageFromResponse, saveUsageStats, formatDoneLine } from "./requestDetail.js"; +import { buildRequestDetail, extractRequestConfig, extractUsageFromResponse, saveUsageStats, formatDoneLine, tokensForDetail, shouldPersistRequestDetail } from "./requestDetail.js"; import { saveRequestDetail } from "@/lib/usageDb.js"; import { matchStreamErrorPatterns } from "../../utils/streamErrorPatterns.js"; import { decloakToolNames } from "../../utils/claudeCloaking.js"; @@ -282,7 +282,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, customToolNames, trackDone, appendLog, pxpipe, reqTag, log, streamErrorPatterns }) { +export async function handleNonStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, customToolNames, trackDone, appendLog, pxpipe, reqTag, log, streamErrorPatterns, persistUsage = "all" }) { trackDone(); const contentType = providerResponse.headers.get("content-type") || ""; let responseBody; @@ -387,23 +387,25 @@ export async function handleNonStreamingResponse({ providerResponse, provider, m reqLogger.logConvertedResponse(translatedResponse); const totalLatency = Date.now() - requestStartTime; - saveRequestDetail(buildRequestDetail({ - provider, model, connectionId, apiKey, - latency: { ttft: totalLatency, total: totalLatency }, - tokens: usage || { prompt_tokens: 0, completion_tokens: 0 }, - request: extractRequestConfig(body, stream), - providerRequest: finalBody || translatedBody || null, - providerResponse: responseBody || null, - response: { - content: translatedResponse?.choices?.[0]?.message?.content || translatedResponse?.content || null, - thinking: translatedResponse?.choices?.[0]?.message?.reasoning_content || translatedResponse?.reasoning_content || null, - finish_reason: translatedResponse?.choices?.[0]?.finish_reason || "unknown" - }, - pxpipe, - status: "success" - }, { endpoint: clientRawRequest?.endpoint || null })).catch(err => { - console.error("[RequestDetail] Failed to save:", err.message); - }); + if (shouldPersistRequestDetail(persistUsage, "success")) { + saveRequestDetail(buildRequestDetail({ + provider, model, connectionId, apiKey, + latency: { ttft: totalLatency, total: totalLatency }, + tokens: tokensForDetail(usage), + request: extractRequestConfig(body, stream), + providerRequest: finalBody || translatedBody || null, + providerResponse: responseBody || null, + response: { + content: translatedResponse?.choices?.[0]?.message?.content || translatedResponse?.content || null, + thinking: translatedResponse?.choices?.[0]?.message?.reasoning_content || translatedResponse?.reasoning_content || null, + finish_reason: translatedResponse?.choices?.[0]?.finish_reason || "unknown" + }, + pxpipe, + status: "success" + }, { endpoint: clientRawRequest?.endpoint || null })).catch(err => { + console.error("[RequestDetail] Failed to save:", err.message); + }); + } return { success: true, diff --git a/open-sse/handlers/chatCore/requestDetail.js b/open-sse/handlers/chatCore/requestDetail.js index 16371034..8d4f88dc 100644 --- a/open-sse/handlers/chatCore/requestDetail.js +++ b/open-sse/handlers/chatCore/requestDetail.js @@ -110,6 +110,29 @@ export function formatDoneLine({ usage, latency }) { return `DONE ${latency?.total ?? 0}ms${ttftStr} · ${inStr} · OUT ${outTok}`; } +// Request-details storage convention: always prompt_tokens / completion_tokens. +// Translators often hand Claude `{input_tokens, output_tokens}` (or Gemini +// counts) to onStreamComplete; the Details tab only reads the OpenAI names, +// so an uncanonicalized object shows up as input=0 / output=0. +export function tokensForDetail(usage) { + if (!usage || typeof usage !== "object") { + return { prompt_tokens: 0, completion_tokens: 0 }; + } + return canonicalizeUsage(usage) || { + prompt_tokens: usage.prompt_tokens ?? usage.input_tokens ?? 0, + completion_tokens: usage.completion_tokens ?? usage.output_tokens ?? 0, + }; +} + +// Combo fallback/account hops must not inflate Details with 0-token rows. +// `streaming-start` is never persisted: the placeholder was status=success at +// tokens=0, and nested/fusion paths often abandon the stream before complete. +export function shouldPersistRequestDetail(persistUsage, kind) { + if (kind === "streaming-start") return false; + if (persistUsage === "success-only") return kind === "success"; + return true; +} + export function saveUsageStats({ provider, model, tokens, connectionId, apiKey, endpoint, label = "USAGE", silent = false }) { if (!tokens || typeof tokens !== "object") return; diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index 9836f244..5ca14be9 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/open-sse/handlers/chatCore/streamingHandler.js @@ -13,6 +13,8 @@ import { extractRequestConfig, saveUsageStats, formatDoneLine, + tokensForDetail, + shouldPersistRequestDetail, } from "./requestDetail.js"; import { streamStatusForContent } from "../../utils/streamErrorPatterns.js"; import { saveRequestDetail } from "@/lib/usageDb.js"; @@ -60,7 +62,7 @@ function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, /** * Handle streaming response — pipe provider SSE through transform stream to client. */ -export async function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, customToolNames, streamController, onStreamComplete, streamDetailId, pxpipe, reqTag, log, credentials }) { +export async function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, customToolNames, streamController, onStreamComplete, pxpipe, reqTag, log, credentials }) { if (onRequestSuccess) { Promise.resolve() .then(onRequestSuccess) @@ -145,35 +147,6 @@ export async function handleStreamingResponse({ providerResponse, provider, mode stallTimeoutMs, ); - saveRequestDetail( - buildRequestDetail( - { - provider, - model, - connectionId, - apiKey, - latency: { ttft: 0, total: Date.now() - requestStartTime }, - tokens: { prompt_tokens: 0, completion_tokens: 0 }, - request: extractRequestConfig(body, stream), - providerRequest: finalBody || translatedBody || null, - providerResponse: "[Streaming - raw response not captured]", - response: { - content: "[Streaming in progress...]", - thinking: null, - type: "streaming", - }, - pxpipe, - status: "success", - }, - { id: streamDetailId }, - ), - ).catch((err) => { - console.error( - "[RequestDetail] Failed to save streaming request:", - err.message, - ); - }); - return { success: true, response: new Response(transformedBody, { headers: SSE_HEADERS }), @@ -198,6 +171,7 @@ export function buildOnStreamComplete({ reqTag, log, streamErrorPatterns, + persistUsage = "all", }) { const streamDetailId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`; @@ -210,37 +184,39 @@ export function buildOnStreamComplete({ const safeThinking = contentObj?.thinking || null; const rawProviderText = typeof contentObj?.rawProviderText === "string" ? contentObj.rawProviderText : ""; - saveRequestDetail( - buildRequestDetail( - { - provider, - model, - connectionId, - apiKey, - latency, - tokens: usage || { prompt_tokens: 0, completion_tokens: 0 }, - request: extractRequestConfig(body, stream), - providerRequest: finalBody || translatedBody || null, - providerResponse: rawProviderText || safeContent, - response: { - content: safeContent, - thinking: safeThinking, - type: "streaming", + if (shouldPersistRequestDetail(persistUsage, "success")) { + saveRequestDetail( + buildRequestDetail( + { + provider, + model, + connectionId, + apiKey, + latency, + tokens: tokensForDetail(usage), + request: extractRequestConfig(body, stream), + providerRequest: finalBody || translatedBody || null, + providerResponse: rawProviderText || safeContent, + response: { + content: safeContent, + thinking: safeThinking, + type: "streaming", + }, + pxpipe, + status: streamStatusForContent( + streamErrorPatterns?.[provider], + safeContent, + ), }, - pxpipe, - status: streamStatusForContent( - streamErrorPatterns?.[provider], - safeContent, - ), - }, - { id: streamDetailId }, - ), - ).catch((err) => { - console.error( - "[RequestDetail] Failed to update streaming content:", - err.message, - ); - }); + { id: streamDetailId }, + ), + ).catch((err) => { + console.error( + "[RequestDetail] Failed to update streaming content:", + err.message, + ); + }); + } // Persist stream usage to DB (no console line; the "📊 done" line below is authoritative) saveUsageStats({ diff --git a/open-sse/services/combo.js b/open-sse/services/combo.js index e3b9a8e6..24536304 100644 --- a/open-sse/services/combo.js +++ b/open-sse/services/combo.js @@ -246,23 +246,39 @@ export function resetComboRotation(comboName) { } /** - * Get combo models from combos data + * Cancel an unused Response body so its SSE transform (and usage callback) + * is not left hanging. Fusion timeouts and combo fallback both drop Response + * objects without reading them; an unread stream never fires onStreamComplete, + * which is how request-details rows get stuck at input/output tokens = 0. + */ +export function discardResponse(res) { + if (!res || res.__timeout || res.__error) return; + const cancel = res.body?.cancel; + if (typeof cancel === "function") { + try { cancel.call(res.body); } catch { /* already closed / consumed */ } + } +} + +/** + * Get combo models from combos data. + * Nested combo names are kept as-is (comboA listing comboB yields ["comboB", ...]). + * Flattening them into leaves would explode fusion/round-robin: every nested + * leaf becomes its own panel/rotation slot. Nested combos run as one unit + * (see handleSingleModelChat: inner strategy forced to fallback). + * * @param {string} modelStr - Model string to check * @param {Array|Object} combosData - Array of combos or object with combos - * @returns {string[]|null} Array of models or null if not a combo + * @returns {string[]|null} Direct members, or null if not a combo */ export function getComboModelsFromData(modelStr, combosData) { - // Don't check if it's in provider/model format - if (modelStr.includes("/")) return null; - - // Handle both array and object formats + if (typeof modelStr !== "string" || modelStr.includes("/")) return null; + const combos = Array.isArray(combosData) ? combosData : (combosData?.combos || []); - - const combo = combos.find(c => c.name === modelStr); - if (combo && combo.models && combo.models.length > 0) { - return combo.models; + const combo = combos.find((c) => c.name === modelStr); + if (!combo || combo.enabled === false || !Array.isArray(combo.models) || combo.models.length === 0) { + return null; } - return null; + return combo.models; } /** @@ -339,6 +355,9 @@ export async function handleComboChat({ body, models, handleSingleModel, log, co return result; } + // Falling through: this body will never be read by the client. + discardResponse(result); + // For transient errors (503/502/504), wait for cooldown before falling through // so a briefly-overloaded provider gets a chance to recover rather than being // skipped immediately (fixes: combo falls through on transient 503) @@ -509,8 +528,13 @@ function collectPanel(calls, { minPanel, stragglerGraceMs, panelHardTimeoutMs }) const hardTimer = setTimeout(finish, panelHardTimeoutMs); calls.forEach((p, i) => { Promise.resolve(p) - .then((v) => { out[i] = v; }) - .catch((e) => { out[i] = { __error: e }; }) + .then((v) => { + // Quorum/timeout already moved on: cancel the late body so its + // streaming usage callback is not left at tokens=0 forever. + if (finished) discardResponse(v); + else out[i] = v; + }) + .catch((e) => { if (!finished) out[i] = { __error: e }; }) .finally(() => { settled++; if (out[i] && out[i].ok) ok++; @@ -590,7 +614,7 @@ export async function handleFusionChat({ body, models, handleSingleModel, log, c if (!res) { log.warn("FUSION", `Panel ${model} dropped (straggler/timeout)`); continue; } if (res.__timeout) { log.warn("FUSION", `Panel ${model} timed out`); continue; } if (res.__error) { log.warn("FUSION", `Panel ${model} threw`, { error: res.__error?.message || String(res.__error) }); continue; } - if (!res.ok) { log.warn("FUSION", `Panel ${model} failed`, { status: res.status }); continue; } + if (!res.ok) { log.warn("FUSION", `Panel ${model} failed`, { status: res.status }); discardResponse(res); continue; } try { const json = await res.clone().json(); const text = extractPanelText(json); @@ -602,6 +626,11 @@ export async function handleFusionChat({ body, models, handleSingleModel, log, c } } catch (e) { log.warn("FUSION", `Panel ${model} unparseable`, { error: e.message || String(e) }); + } finally { + // Panel responses are never returned to the client; cancel the original + // body (clone().json() already captured the payload) so streaming + // placeholders are not left as success/0-token rows. + discardResponse(res); } } diff --git a/src/app/(dashboard)/dashboard/usage/components/RequestDetailsTab.js b/src/app/(dashboard)/dashboard/usage/components/RequestDetailsTab.js index 59008df9..f8284cc1 100644 --- a/src/app/(dashboard)/dashboard/usage/components/RequestDetailsTab.js +++ b/src/app/(dashboard)/dashboard/usage/components/RequestDetailsTab.js @@ -92,7 +92,7 @@ function getCacheCreationTokens(tokens) { } function getInputTokens(tokens) { - const prompt = tokens?.prompt_tokens || tokens?.input_tokens || 0; + const prompt = tokens?.prompt_tokens || tokens?.input_tokens || tokens?.promptTokenCount || 0; // Canonical storage keeps prompt cache-inclusive. Legacy Claude rows may have // stored prompt cache-exclusive; fall back to cache when it's larger so old // rows don't under-report input. @@ -100,6 +100,10 @@ function getInputTokens(tokens) { return prompt < cache ? cache : prompt; } +function getOutputTokens(tokens) { + return tokens?.completion_tokens || tokens?.output_tokens || tokens?.candidatesTokenCount || 0; +} + export default function RequestDetailsTab() { const [details, setDetails] = useState([]); const [pagination, setPagination] = useState({ @@ -427,13 +431,15 @@ export default function RequestDetailsTab() { "inline-flex items-center gap-1 rounded-full px-2 py-0.5 text-xs font-medium", detail.status === "error" ? "bg-red-500/10 text-red-600" - : "bg-green-500/10 text-green-600" + : detail.status === "streaming" + ? "bg-amber-500/10 text-amber-700 dark:text-amber-400" + : "bg-green-500/10 text-green-600" )}> - {detail.status === "error" ? "Error" : "Success"} + {detail.status === "error" ? "Error" : detail.status === "streaming" ? "Streaming" : "Success"} @@ -446,7 +452,7 @@ export default function RequestDetailsTab() { {getCacheCreationTokens(detail.tokens) > 0 ? getCacheCreationTokens(detail.tokens).toLocaleString() : "—"} - {detail.tokens?.completion_tokens?.toLocaleString() || 0} + {getOutputTokens(detail.tokens).toLocaleString()}
@@ -525,7 +531,9 @@ export default function RequestDetailsTab() { Status:{" "} {selectedDetail.status} @@ -561,7 +569,7 @@ export default function RequestDetailsTab() {
Output Tokens:{" "} - {selectedDetail.tokens?.completion_tokens?.toLocaleString() || 0} + {getOutputTokens(selectedDetail.tokens).toLocaleString()}
diff --git a/src/sse/handlers/chat.js b/src/sse/handlers/chat.js index ba25159b..4d7e9a0c 100644 --- a/src/sse/handlers/chat.js +++ b/src/sse/handlers/chat.js @@ -115,6 +115,7 @@ export async function handleChat(request, clientRawRequest = null) { const augmentedModels = augmentModelsWithCapacityAdapter(comboModels, requiredCapabilities, settings); const adapterAdded = augmentedModels.filter((m) => !comboModels.includes(m)); + const comboOpts = { persistUsage: "success-only", comboStack: new Set([modelStr]) }; if (comboStrategy === "fusion") { log.info("CHAT", `Combo "${modelStr}" with ${comboModels.length} models (strategy: fusion)`); return handleFusionChat({ @@ -126,7 +127,7 @@ export async function handleChat(request, clientRawRequest = null) { const { tools, tool_choice, ...cleanBody } = clientRawRequest.body || {}; cleanRawReq = { ...clientRawRequest, body: cleanBody }; } - return handleSingleModelChat(b, m, cleanRawReq, request, apiKey); + return handleSingleModelChat(b, m, cleanRawReq, request, apiKey, comboOpts); }, log, comboName: modelStr, @@ -141,7 +142,7 @@ export async function handleChat(request, clientRawRequest = null) { body, models: augmentedModels, handleSingleModel: withCapacityAdapterStripping( - (b, m) => handleSingleModelChat(b, m, clientRawRequest, request, apiKey), + (b, m) => handleSingleModelChat(b, m, clientRawRequest, request, apiKey, comboOpts), adapterAdded ), log, @@ -161,7 +162,7 @@ export async function handleChat(request, clientRawRequest = null) { body, models: soloAugmented, handleSingleModel: withCapacityAdapterStripping( - (b, m) => handleSingleModelChat(b, m, clientRawRequest, request, apiKey), + (b, m) => handleSingleModelChat(b, m, clientRawRequest, request, apiKey, { persistUsage: "success-only" }), adapterAdded ), log, @@ -176,24 +177,42 @@ export async function handleChat(request, clientRawRequest = null) { /** * Handle single model chat request */ -async function handleSingleModelChat(body, modelStr, clientRawRequest = null, request = null, apiKey = null) { +async function handleSingleModelChat(body, modelStr, clientRawRequest = null, request = null, apiKey = null, opts = {}) { // Dashboard per-key tests pin the account via x-connection-id (same header // contract as embeddings/images/video). Pinned requests must not rotate. const preferredConnectionId = request?.headers?.get("x-connection-id") || null; const modelInfo = await getModelInfo(modelStr); + const stack = opts.comboStack instanceof Set ? opts.comboStack : new Set(); + const persistUsage = opts.persistUsage || "all"; // If provider is null, this might be a combo name - check and handle if (!modelInfo.provider) { + if (stack.has(modelStr)) { + log.warn("CHAT", `Combo cycle detected: ${modelStr}`); + return errorResponse(HTTP_STATUS.BAD_REQUEST, `Combo cycle detected involving "${modelStr}"`); + } const comboModels = await getComboModels(modelStr); if (comboModels) { const chatSettings = await getSettings(); - // Check for combo-specific strategy first, fallback to global const comboStrategies = chatSettings.comboStrategies || {}; const comboSpecificStrategy = comboStrategies[modelStr]?.fallbackStrategy; - const comboStrategy = comboSpecificStrategy || chatSettings.comboStrategy || "fallback"; + // Nested combo (comboA lists comboB): always fallback. Running the inner + // combo's fusion/round-robin as its own combo multiplies requests and + // leaves abandoned streams at tokens=0 in Details. + const nested = stack.size > 0; + const comboStrategy = nested + ? "fallback" + : (comboSpecificStrategy || chatSettings.comboStrategy || "fallback"); const requiredCapabilities = detectRequiredCapabilities(body); const augmentedModels = augmentModelsWithCapacityAdapter(comboModels, requiredCapabilities, chatSettings); const adapterAdded = augmentedModels.filter((m) => !comboModels.includes(m)); + const nextOpts = { + persistUsage: "success-only", + comboStack: new Set(stack).add(modelStr), + }; + if (nested) { + log.info("CHAT", `Nested combo "${modelStr}" → fallback (${comboModels.length} members)`); + } if (comboStrategy === "fusion") { log.info("CHAT", `Combo "${modelStr}" with ${comboModels.length} models (strategy: fusion)`); @@ -206,7 +225,7 @@ async function handleSingleModelChat(body, modelStr, clientRawRequest = null, re const { tools, tool_choice, ...cleanBody } = clientRawRequest.body || {}; cleanRawReq = { ...clientRawRequest, body: cleanBody }; } - return handleSingleModelChat(b, m, cleanRawReq, request, apiKey); + return handleSingleModelChat(b, m, cleanRawReq, request, apiKey, nextOpts); }, log, comboName: modelStr, @@ -221,7 +240,7 @@ async function handleSingleModelChat(body, modelStr, clientRawRequest = null, re body, models: augmentedModels, handleSingleModel: withCapacityAdapterStripping( - (b, m) => handleSingleModelChat(b, m, clientRawRequest, request, apiKey), + (b, m) => handleSingleModelChat(b, m, clientRawRequest, request, apiKey, nextOpts), adapterAdded ), log, @@ -323,6 +342,7 @@ async function handleSingleModelChat(body, modelStr, clientRawRequest = null, re providerThinking, capsOverride, streamErrorPatterns: chatSettings.streamErrorPatterns || null, + persistUsage, // Detect source format by endpoint + body sourceFormatOverride: request?.url ? detectFormatByEndpoint(new URL(request.url).pathname, body) : null, onCredentialsRefreshed: async (newCreds) => { diff --git a/src/sse/services/model.js b/src/sse/services/model.js index 99248847..d304808b 100644 --- a/src/sse/services/model.js +++ b/src/sse/services/model.js @@ -123,21 +123,16 @@ export async function getModelInfo(modelStr) { } /** - * Check if model is a combo and get models list + * Check if model is a combo and get models list. + * Nested combo names are returned as members (not inlined); chat.js runs them + * as a fallback unit so fusion/round-robin of the parent is not exploded. * @returns {Promise} Array of models or null if not a combo */ export async function getComboModels(modelStr) { - // Only check if it's not in provider/model format - if (modelStr.includes("/")) return null; - + if (typeof modelStr !== "string" || modelStr.includes("/")) return null; const combo = await getComboByName(modelStr); - if ( - combo && - combo.enabled !== false && - combo.models && - combo.models.length > 0 - ) { - return combo.models; + if (!combo || combo.enabled === false || !Array.isArray(combo.models) || combo.models.length === 0) { + return null; } - return null; + return combo.models; } diff --git a/tests/unit/combo-nested.test.js b/tests/unit/combo-nested.test.js new file mode 100644 index 00000000..ea58cbb1 --- /dev/null +++ b/tests/unit/combo-nested.test.js @@ -0,0 +1,128 @@ +import { describe, it, expect, vi } from "vitest"; + +import { getComboModelsFromData, discardResponse, handleFusionChat } from "../../open-sse/services/combo.js"; +import { tokensForDetail, shouldPersistRequestDetail } from "../../open-sse/handlers/chatCore/requestDetail.js"; + +vi.mock("@/lib/usageDb.js", () => ({ + saveRequestUsage: vi.fn(), + saveRequestDetail: vi.fn(), +})); +vi.mock("../../open-sse/utils/stream.js", () => ({ + COLORS: {}, + formatSSE: vi.fn(), +})); + +const log = { info: () => {}, warn: () => {}, debug: () => {} }; + +describe("nested combo members stay units", () => { + it("keeps comboB as a member of comboA (does not flatten leaves into the parent)", () => { + const combos = [ + { name: "comboA", models: ["comboB", "comboC", "x/y"] }, + { name: "comboB", models: ["a/m1", "b/m2"] }, + { name: "comboC", models: ["c/m3"] }, + ]; + expect(getComboModelsFromData("comboA", combos)).toEqual(["comboB", "comboC", "x/y"]); + }); + + it("returns null for a plain provider/model string", () => { + expect(getComboModelsFromData("openai/gpt-4", [{ name: "comboA", models: ["a/m1"] }])).toBeNull(); + }); + + it("still returns direct members of a top-level combo", () => { + const combos = [{ name: "solo", models: ["a/m1", "b/m2"] }]; + expect(getComboModelsFromData("solo", combos)).toEqual(["a/m1", "b/m2"]); + }); + + it("ignores a disabled combo", () => { + const combos = [{ name: "solo", enabled: false, models: ["a/m1"] }]; + expect(getComboModelsFromData("solo", combos)).toBeNull(); + }); +}); + +describe("shouldPersistRequestDetail", () => { + it("never persists the streaming-start placeholder (0-token fake success)", () => { + expect(shouldPersistRequestDetail("all", "streaming-start")).toBe(false); + expect(shouldPersistRequestDetail("success-only", "streaming-start")).toBe(false); + }); + + it("combo hops skip failed attempts so nested fallback does not inflate Details", () => { + expect(shouldPersistRequestDetail("success-only", "error")).toBe(false); + expect(shouldPersistRequestDetail("success-only", "success")).toBe(true); + }); + + it("direct (non-combo) requests still persist errors", () => { + expect(shouldPersistRequestDetail("all", "error")).toBe(true); + expect(shouldPersistRequestDetail("all", "success")).toBe(true); + }); +}); + +describe("tokensForDetail", () => { + it("canonicalizes Claude input_tokens/output_tokens so the Details tab is not 0/0", () => { + expect(tokensForDetail({ input_tokens: 120, output_tokens: 40 })).toEqual( + expect.objectContaining({ prompt_tokens: 120, completion_tokens: 40 }), + ); + }); + + it("keeps OpenAI prompt/completion names", () => { + expect(tokensForDetail({ prompt_tokens: 10, completion_tokens: 3 })).toEqual( + expect.objectContaining({ prompt_tokens: 10, completion_tokens: 3 }), + ); + }); + + it("returns zeros for missing usage (streaming placeholder)", () => { + expect(tokensForDetail(null)).toEqual({ prompt_tokens: 0, completion_tokens: 0 }); + }); +}); + +describe("discardResponse", () => { + it("cancels an unused body so a dropped stream can finish", () => { + const cancel = vi.fn(); + discardResponse({ body: { cancel } }); + expect(cancel).toHaveBeenCalledTimes(1); + }); + + it("is a no-op for timeout/error sentinels and missing bodies", () => { + expect(() => discardResponse({ __timeout: true, body: { cancel: () => { throw new Error("no"); } } })).not.toThrow(); + expect(() => discardResponse(null)).not.toThrow(); + }); +}); + +describe("fusion cancels unused panel bodies", () => { + it("cancels a late panel Response so it is not left as a 0-token streaming row", async () => { + const cancel = vi.fn(); + const ok = (content) => { + const json = { choices: [{ message: { role: "assistant", content } }] }; + const make = () => ({ ok: true, status: 200, clone: make, json: async () => json }); + return make(); + }; + const slow = () => new Promise((resolve) => { + setTimeout(() => { + resolve({ + ok: true, + status: 200, + body: { cancel }, + clone() { return this; }, + json: async () => ({ choices: [{ message: { content: "late" } }] }), + }); + }, 40); + }); + + const handleSingleModel = vi.fn(async (_body, model) => { + if (model === "p/slow") return slow(); + if (model === "p/judge") return ok("FINAL"); + return ok(`ans-${model}`); + }); + + await handleFusionChat({ + body: { messages: [{ role: "user", content: "Q" }] }, + models: ["p/a", "p/b", "p/slow"], + handleSingleModel, + log, + judgeModel: "p/judge", + tuning: { minPanel: 2, stragglerGraceMs: 1, panelHardTimeoutMs: 200 }, + }); + + await new Promise((r) => setTimeout(r, 80)); + expect(cancel).toHaveBeenCalled(); + }); +});