Merge branch 'master' of https://github.com/decolua/9router

# Conflicts:
#	open-sse/utils/usageTracking.js

open-sse/executors/qoder.js

@@ -223,22 +223,75 @@ 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.
  */
-function wrapQoderSSE(response, model) {
+async function wrapQoderSSE(response, model) {
   if (!response.ok || !response.body) return response;
 
   const decoder = new TextDecoder();
   const encoder = new TextEncoder();
+
+  // Peek at first chunk to detect errors early
+  const reader = response.body.getReader();
+  const firstRead = await reader.read();
+  
+  if (firstRead.done) {
+    // Empty stream
+    return new Response("data: [DONE]\n\n", {
+      status: response.status,
+      statusText: response.statusText,
+      headers: {
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache",
+      },
+    });
+  }
+
+  // Parse first line to check for error
+  const firstText = decoder.decode(firstRead.value, { stream: true });
+  const nlIndex = firstText.indexOf("\n");
+  const firstLine = nlIndex !== -1 ? firstText.slice(0, nlIndex) : firstText;
+  const trimmed = firstLine.replace(/\r$/, "").trim();
+  
+  if (trimmed.startsWith("data:")) {
+    const data = trimmed.slice(5).trimStart();
+    if (data !== "[DONE]") {
+      try {
+        const envelope = JSON.parse(data);
+        const statusVal = typeof envelope.statusCodeValue === "number" ? envelope.statusCodeValue : 200;
+        
+        if (statusVal !== 200) {
+          // Error detected - return error Response to trigger failover
+          const msg = envelope.body || `upstream status ${statusVal}`;
+          const errorResponse = new Response(
+            JSON.stringify({ 
+              error: { 
+                message: `qoder error ${statusVal}: ${truncate(msg, 500)}`,
+                type: "upstream_error",
+                code: String(statusVal)
+              } 
+            }),
+            { 
+              status: statusVal >= 400 && statusVal < 600 ? statusVal : 502,
+              headers: { "Content-Type": "application/json" }
+            }
+          );
+          reader.cancel();
+          return errorResponse;
+        }
+      } catch (e) {
+        // Not JSON, continue as normal stream
+      }
+    }
+  }
+
+  // No error detected - proceed with normal TransformStream
   let buffer = "";
   let doneEmitted = false;
 
-  // Process one already-extracted SSE line (no trailing newline). Returns
-  // false when the line indicated end-of-stream so the caller can stop
-  // forwarding any remaining chunks after [DONE].
   const processLine = (line, controller) => {
     const trimmed = line.replace(/\r$/, "").trim();
     if (!trimmed) return;
     if (!trimmed.startsWith("data:")) return;
-    if (doneEmitted) return; // never forward chunks past stream end
+    if (doneEmitted) return;
 
     const data = trimmed.slice(5).trimStart();
     if (data === "[DONE]") {
@@ -271,10 +324,6 @@ function wrapQoderSSE(response, model) {
       doneEmitted = true;
       return;
     }
-    // Inner is an OpenAI-shaped chunk. Strip any embedded newlines so the
-    // SSE frame stays a single event (a literal "\n" inside `inner` would
-    // otherwise split the frame across multiple data: lines and downstream
-    // parsers would reassemble them as separate events).
     const sanitized = inner.replace(/\r?\n/g, "");
     controller.enqueue(encoder.encode(`data: ${sanitized}\n\n`));
   };
@@ -290,13 +339,7 @@ function wrapQoderSSE(response, model) {
       }
     },
     flush(controller) {
-      // Finalize the decoder so any pending multi-byte sequence is
-      // released into `buffer` instead of being silently dropped.
       buffer += decoder.decode();
-      // Drain any trailing line that arrived without a terminating newline
-      // (e.g. upstream closed the socket immediately after the last write,
-      // or a CDN stripped the final CRLF). Without this, the chunk that
-      // carries finish_reason is silently lost.
       if (buffer.length > 0) {
         processLine(buffer, controller);
         buffer = "";
@@ -308,9 +351,25 @@ function wrapQoderSSE(response, model) {
     },
   });
 
-  const transformed = response.body.pipeThrough(transform);
-  // Build a Response with passable headers; the streaming handler reads
-  // `.body` as a ReadableStream regardless of Content-Type.
+  // Create a ReadableStream that emits the first chunk + remaining chunks
+  const combinedStream = new ReadableStream({
+    start(controller) {
+      controller.enqueue(firstRead.value);
+    },
+    async pull(controller) {
+      const { done, value } = await reader.read();
+      if (done) {
+        controller.close();
+      } else {
+        controller.enqueue(value);
+      }
+    },
+    cancel() {
+      reader.cancel();
+    }
+  });
+
+  const transformed = combinedStream.pipeThrough(transform);
   return new Response(transformed, {
     status: response.status,
     statusText: response.statusText,
@@ -431,7 +490,7 @@ export class QoderExecutor extends BaseExecutor {
       return { response, url, headers, transformedBody: payload };
     }
 
-    const wrapped = wrapQoderSSE(response, `qoder/${qoderKey}`);
+    const wrapped = await wrapQoderSSE(response, `qoder/${qoderKey}`);
     return { response: wrapped, url, headers, transformedBody: payload };
   }
 

open-sse/handlers/chatCore/requestDetail.js

@@ -80,14 +80,29 @@ export function saveUsageStats({ provider, model, tokens, connectionId, apiKey,
 
   if (inTokens === 0 && outTokens === 0) 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;
+  const cacheCreation = tokens.cache_creation_input_tokens || 0;
+  const reasoning = tokens.reasoning_tokens || 0;
+
   const time = new Date().toLocaleTimeString("en-US", { hour12: false, hour: "2-digit", minute: "2-digit", second: "2-digit" });
   const accountSuffix = connectionId ? ` | account=${connectionId.slice(0, 8)}...` : "";
-  console.log(`${COLORS.green}[${time}] 📊 [${label}] ${provider.toUpperCase()} | in=${inTokens} | out=${outTokens}${accountSuffix}${COLORS.reset}`);
+  
+  let msg = `${COLORS.green}[${time}] 📊 [${label}] ${provider?.toUpperCase() || "UNKNOWN"} | in=${inTokens} | out=${outTokens}${accountSuffix}`;
+  if (tokens.estimated) msg += ` ${COLORS.yellow}(estimated)${COLORS.reset}`;
+  if (cacheRead) msg += ` | cache_read=${cacheRead}`;
+  if (cacheCreation) msg += ` | cache_create=${cacheCreation}`;
+  if (reasoning) msg += ` | reasoning=${reasoning}`;
+  msg += `${COLORS.reset}`;
+  console.log(msg);
 
-  // Normalize to OpenAI token shape for storage
+  // Normalize to OpenAI token shape for storage (include all token types)
   const normalized = {
     prompt_tokens: tokens.prompt_tokens ?? tokens.input_tokens ?? 0,
-    completion_tokens: tokens.completion_tokens ?? tokens.output_tokens ?? 0
+    completion_tokens: tokens.completion_tokens ?? tokens.output_tokens ?? 0,
+    cache_read_input_tokens: cacheRead,
+    cache_creation_input_tokens: cacheCreation,
+    reasoning_tokens: reasoning,
   };
 
   saveRequestUsage({

open-sse/handlers/chatCore/streamingHandler.js

@@ -5,7 +5,7 @@ import { pipeWithDisconnect } from "../../utils/streamHandler.js";
 import { PROVIDERS } from "../../config/providers.js";
 import { STREAM_STALL_TIMEOUT_MS } from "../../config/runtimeConfig.js";
 import { buildAbortedResponsesTerminalBytes } from "../../utils/responsesStreamHelpers.js";
-import { buildRequestDetail, extractRequestConfig, saveUsageStats } from "./requestDetail.js";
+import { buildRequestDetail, extractRequestConfig } from "./requestDetail.js";
 import { saveRequestDetail } from "@/lib/usageDb.js";
 import { SSE_HEADERS_CORS as SSE_HEADERS } from "../../utils/sseConstants.js";
 
@@ -114,8 +114,6 @@ export function buildOnStreamComplete({ provider, model, connectionId, apiKey, r
     }, { id: streamDetailId })).catch(err => {
       console.error("[RequestDetail] Failed to update streaming content:", err.message);
     });
-
-    saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint, label: "STREAM USAGE" });
   };
 
   return { onStreamComplete, streamDetailId };

open-sse/utils/stream.js

@@ -1,7 +1,8 @@
 import { translateResponse, initState } from "../translator/index.js";
 import { FORMATS } from "../translator/formats.js";
 import { trackPendingRequest, appendRequestLog } from "@/lib/usageDb.js";
-import { extractUsage, hasValidUsage, estimateUsage, logUsage, addBufferToUsage, filterUsageForFormat, COLORS } from "./usageTracking.js";
+import { extractUsage, hasValidUsage, estimateUsage, addBufferToUsage, filterUsageForFormat, COLORS } from "./usageTracking.js";
+import { saveUsageStats } from "../handlers/chatCore/requestDetail.js";
 import { parseSSELine, hasValuableContent, fixInvalidId, formatSSE } from "./streamHelpers.js";
 import { getOpenAIResponsesEventName, isOpenAIResponsesTerminalEvent, formatIncompleteOpenAIResponsesStreamFailure } from "./responsesStreamHelpers.js";
 import { dbg, isDebugEnabled } from "./debugLog.js";
@@ -341,7 +342,7 @@ export function createSSEStream(options = {}) {
           }
 
           if (hasValidUsage(usage)) {
-            logUsage(provider, usage, model, connectionId, apiKey);
+            saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey });
           } else {
             appendRequestLog({ model, provider, connectionId, tokens: null, status: "200 OK" }).catch(() => { });
           }
@@ -423,7 +424,7 @@ export function createSSEStream(options = {}) {
         }
 
         if (hasValidUsage(state?.usage)) {
-          logUsage(state.provider || targetFormat, state.usage, model, connectionId, apiKey);
+          saveUsageStats({ provider: state.provider || targetFormat, model, tokens: state.usage, connectionId, apiKey });
         } else {
           appendRequestLog({ model, provider, connectionId, tokens: null, status: "200 OK" }).catch(() => { });
         }

open-sse/utils/usageTracking.js

@@ -298,7 +298,6 @@ export function estimateUsage(body, contentLength, targetFormat = FORMATS.OPENAI
     targetFormat
   );
 }
-
 /**
  * Log usage with cache info (green color)
  */