fix(combo): keep nested combos as single units and stop 0-token detail rows

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 <noreply@commandcode.ai>
2026-09-17 14:03:59 +07:00
parent a84ba559f3
commit 81a47f36c6
11 changed files with 332 additions and 144 deletions
--- 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
--- 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,
--- 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;

--- 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({
--- 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);
    }
  }