From a625ea9fd8d663e8219039464df9478dada66771 Mon Sep 17 00:00:00 2001 From: decolua Date: Fri, 10 Jul 2026 18:01:20 +0700 Subject: [PATCH] refactor(log): unify request lifecycle logging with session-colored tags Collapse scattered per-request console lines (request/routing/auth/pending/ usage/stream-usage/stream) into 3 correlated lines: request, transform, done. Add stable per-session color tag so concurrent request lines are easy to follow, surface thinking intent, always-on full error logging for debug, re-enable warn level, and uppercase keyword labels. Also fix usage overview cards wrapping (5 cards -> grid-cols-5). Co-authored-by: Cursor --- open-sse/handlers/chatCore.js | 88 ++++++++++++++----- .../handlers/chatCore/nonStreamingHandler.js | 7 +- open-sse/handlers/chatCore/requestDetail.js | 28 +++++- .../handlers/chatCore/sseToJsonHandler.js | 10 ++- .../handlers/chatCore/streamingHandler.js | 13 +-- open-sse/utils/streamHandler.js | 21 ++--- open-sse/utils/usageTracking.js | 7 ++ .../usage/components/OverviewCards.js | 2 +- src/lib/db/repos/usageRepo.js | 3 +- src/sse/handlers/chat.js | 21 +---- src/sse/utils/logger.js | 45 +++++++++- 11 files changed, 176 insertions(+), 69 deletions(-) diff --git a/open-sse/handlers/chatCore.js b/open-sse/handlers/chatCore.js index a1a54d60..e72235f6 100644 --- a/open-sse/handlers/chatCore.js +++ b/open-sse/handlers/chatCore.js @@ -3,7 +3,6 @@ import { translateRequest } from "../translator/index.js"; import { stripThinkingSuffix } from "../translator/concerns/thinkingUnified.js"; import { FORMATS } from "../translator/formats.js"; import { normalizeClaudePassthrough } from "../translator/formats/claude.js"; -import { COLORS } from "../utils/stream.js"; import { createStreamController } from "../utils/streamHandler.js"; import { refreshWithRetry } from "../services/tokenRefresh.js"; import { createRequestLogger } from "../utils/requestLogger.js"; @@ -22,12 +21,14 @@ import { detectClientTool, isNativePassthrough } from "../utils/clientDetector.j import { dedupeTools } from "../utils/toolDeduper.js"; import { injectCaveman } from "../rtk/caveman.js"; import { injectPonytail } from "../rtk/ponytail.js"; -import { compressMessages, formatRtkLog } from "../rtk/index.js"; -import { compressWithHeadroom, formatHeadroomLog, formatHeadroomSizeLog, isHeadroomPhantomSavings } from "../rtk/headroom.js"; -import { compressWithPxpipe, formatPxpipeLog } from "../rtk/pxpipe.js"; +import { compressMessages } from "../rtk/index.js"; +import { compressWithHeadroom, formatHeadroomSizeLog, isHeadroomPhantomSavings } from "../rtk/headroom.js"; +import { compressWithPxpipe } from "../rtk/pxpipe.js"; import { getCapabilitiesForModel } from "../providers/capabilities.js"; import { stripUnsupportedModalities } from "../translator/concerns/modality.js"; import { prefetchRemoteImages } from "../translator/concerns/prefetch.js"; +import { extractThinking } from "../translator/concerns/thinkingUnified.js"; +import { resolveSessionId } from "../utils/sessionManager.js"; /** * Core chat handler - shared between SSE and Worker @@ -39,6 +40,15 @@ import { prefetchRemoteImages } from "../translator/concerns/prefetch.js"; export async function handleChatCore({ body, modelInfo, credentials, log, onCredentialsRefreshed, onRequestSuccess, onDisconnect, clientRawRequest, connectionId, userAgent, apiKey, ccFilterNaming, rtkEnabled, headroomEnabled, headroomUrl, headroomCompressUserMessages, cavemanEnabled, cavemanLevel, ponytailEnabled, ponytailLevel, pxpipeEnabled, pxpipeMinChars, pxpipeTimeoutMs, pxpipeTransform, onPxpipeEvent, sourceFormatOverride, providerThinking }) { const { provider, model } = modelInfo; const requestStartTime = Date.now(); + // Stable per-session color so all lines of one CLI conversation share a tag + const sessionSeed = (() => { + try { + return resolveSessionId({ headers: clientRawRequest?.headers, body, connectionId, scope: provider }); + } catch { + return connectionId || ""; + } + })(); + const reqTag = log?.tagForSession ? log.tagForSession(sessionSeed) : (log?.nextTag ? log.nextTag() : ""); const sourceFormat = sourceFormatOverride || detectFormat(body); @@ -152,39 +162,68 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred // Covers both passthrough (source shape) and translated (target shape) flows const finalFormat = passthrough ? sourceFormat : targetFormat; + // Request line: one correlated summary (fmt + thinking + counts + account) + if (log?.line) { + const clientModel = clientRawRequest?.body?.model || `${provider}/${model}`; + const msgN = translatedBody.messages?.length || translatedBody.input?.length || translatedBody.contents?.length || body.messages?.length || body.input?.length || 0; + const toolN = translatedBody.tools?.length || body.tools?.length || 0; + const fmtStr = passthrough ? `FMT: ${sourceFormat} (passthrough)` : `FMT: ${sourceFormat}→${targetFormat}`; + const think = log.fmtThink?.(extractThinking(translatedBody)); + const acc = credentials?.connectionName || credentials?.connectionId?.slice(0, 8) || "-"; + const parts = [ + `POST ${clientModel} → ${provider}/${model}`, + fmtStr, + stream ? "STREAM" : "JSON", + `${msgN} MSG`, + ]; + if (toolN) parts.push(`${toolN} TOOL`); + if (think) parts.push(`THINK:${think}`); + parts.push(`ACC:${acc}`); + log.line(reqTag, "▶", parts.join(" · ")); + } + // TTS models don't support tool messages/function calling if (getModelType(alias, model) === "tts" && translatedBody.messages) { translatedBody.messages = translatedBody.messages.filter(msg => msg.role !== "tool"); delete translatedBody.tools; } + // Token-saver summary parts, printed as one "⚙" line at the end (only active ones) + const xf = []; + // RTK: compress tool_result content const rtkStats = compressMessages(translatedBody, rtkEnabled); - const rtkLine = formatRtkLog(rtkStats); - if (rtkLine) console.log(rtkLine); + if (rtkStats?.hits?.length) { + const saved = rtkStats.bytesBefore - rtkStats.bytesAfter; + const pct = rtkStats.bytesBefore > 0 ? ((saved / rtkStats.bytesBefore) * 100).toFixed(0) : "0"; + xf.push(`RTK −${saved}B(${pct}%)`); + } // Headroom: optional external proxy compression; fail open if proxy is absent. const headroomDiagnostics = {}; const headroomStats = await compressWithHeadroom(translatedBody, { enabled: headroomEnabled, url: headroomUrl, model: upstreamModel, format: finalFormat, compressUserMessages: headroomCompressUserMessages, diagnostics: headroomDiagnostics }); - const headroomLine = formatHeadroomLog(headroomStats); - const headroomSizeLine = formatHeadroomSizeLog(headroomDiagnostics); - if (headroomLine) { - log?.info?.("HEADROOM", `${headroomLine}${headroomSizeLine ? ` | ${headroomSizeLine}` : ""}`); + if (headroomStats) { + const before = headroomStats.tokens_before || 0; + const delta = headroomStats.tokens_saved || 0; + const pct = before > 0 ? ((delta / before) * 100).toFixed(1) : "0"; + xf.push(`HEADROOM −${delta}tok(${pct}%)`); if (isHeadroomPhantomSavings(headroomStats, headroomDiagnostics)) { - log?.warn?.("HEADROOM", `reported token delta, but outbound JSON shrank <5%; provider may bill near-original payload | ${headroomSizeLine}`); + log?.warn?.("HEADROOM", `reported token delta, but outbound JSON shrank <5%; provider may bill near-original payload | ${formatHeadroomSizeLog(headroomDiagnostics)}`); } - } else if (headroomEnabled) log?.warn?.("HEADROOM", `skipped: ${headroomDiagnostics.reason || "compression unavailable"}${headroomDiagnostics.endpoint ? ` (${headroomDiagnostics.endpoint})` : ""}`); + } else if (headroomEnabled) { + log?.warn?.("HEADROOM", `skipped: ${headroomDiagnostics.reason || "compression unavailable"}${headroomDiagnostics.endpoint ? ` (${headroomDiagnostics.endpoint})` : ""}`); + } // Caveman: inject terse-style system prompt if (cavemanEnabled && cavemanLevel) { injectCaveman(translatedBody, finalFormat, cavemanLevel); - log?.debug?.("CAVEMAN", `${cavemanLevel} | ${finalFormat}`); + xf.push(`CAVEMAN:${cavemanLevel}`); } // Ponytail: inject lazy-senior-dev system prompt if (ponytailEnabled && ponytailLevel) { injectPonytail(translatedBody, finalFormat, ponytailLevel); - log?.debug?.("PONYTAIL", `${ponytailLevel} | ${finalFormat}`); + xf.push(`PONYTAIL:${ponytailLevel}`); } // PXPIPE: image bulky context (Claude-format bodies only), last saver before dispatch @@ -196,12 +235,12 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred }); pxpipeSummary = pxpipeResult.summary; if (pxpipeResult.body) translatedBody = pxpipeResult.body; - const pxpipeLine = formatPxpipeLog(pxpipeSummary); - if (pxpipeLine) log?.info?.("PXPIPE", pxpipeLine); - else log?.debug?.("PXPIPE", `skipped: ${pxpipeSummary.reason}${pxpipeSummary.detail ? ` (${pxpipeSummary.detail})` : ""}`); + if (pxpipeSummary?.applied) xf.push(`PXPIPE:${pxpipeSummary.imageCount}img`); try { onPxpipeEvent?.({ provider, model, ...pxpipeSummary }); } catch { /* stats must not break requests */ } } + if (xf.length && log?.line) log.line(reqTag, "⚙", xf.join(" · ")); + const executor = getExecutor(provider); trackPendingRequest(model, provider, connectionId, true); appendRequestLog({ model, provider, connectionId, status: "PENDING" }).catch(() => { }); @@ -215,7 +254,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred if (onDisconnect) onDisconnect(reason); }, onError: () => trackPendingRequest(model, provider, connectionId, false), - log, provider, model + log, provider, model, reqTag }); const proxyOptions = { @@ -279,7 +318,9 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred return createErrorResult(499, "Request aborted"); } const errMsg = formatProviderError(error, provider, model, HTTP_STATUS.BAD_GATEWAY); - console.log(`${COLORS.red}[ERROR] ${errMsg}${COLORS.reset}`); + if (log?.errorLine) { + log.errorLine(reqTag, "✗", `ERROR 502 · ${provider}/${model} · ${Date.now() - requestStartTime}ms\n ${errMsg}${error.stack ? `\n ${error.stack}` : ""}`); + } return createErrorResult(HTTP_STATUS.BAD_GATEWAY, errMsg); } @@ -288,7 +329,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred try { const newCredentials = await refreshWithRetry(() => executor.refreshCredentials(credentials, log), 3, log); if (newCredentials?.accessToken || newCredentials?.copilotToken) { - log?.info?.("TOKEN", `${provider.toUpperCase()} | refreshed`); + if (log?.line) log.line(reqTag, "🔑", `TOKEN REFRESHED · ${provider}/${model}`); Object.assign(credentials, newCredentials); if (onCredentialsRefreshed) { try { await onCredentialsRefreshed(newCredentials); } catch (e) { log?.warn?.("TOKEN", `onCredentialsRefreshed failed: ${e.message}`); } @@ -322,12 +363,15 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred })).catch(() => { }); const errMsg = formatProviderError(new Error(message), provider, model, statusCode); - console.log(`${COLORS.red}[ERROR] ${errMsg}${COLORS.reset}`); + if (log?.errorLine) { + const urlStr = providerUrl ? `\n URL: ${providerUrl}` : ""; + log.errorLine(reqTag, "✗", `ERROR ${statusCode} · ${provider}/${model} · ${Date.now() - requestStartTime}ms${urlStr}\n ${errMsg}`); + } reqLogger.logError(new Error(message), finalBody || translatedBody); return createErrorResult(statusCode, errMsg, resetsAtMs); } - const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, pxpipe: pxpipeSummary }; + const sharedCtx = { provider, model, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, pxpipe: pxpipeSummary, reqTag, log }; const appendLog = (extra) => appendRequestLog({ model, provider, connectionId, ...extra }).catch(() => { }); const trackDone = () => trackPendingRequest(model, provider, connectionId, false); diff --git a/open-sse/handlers/chatCore/nonStreamingHandler.js b/open-sse/handlers/chatCore/nonStreamingHandler.js index 654ead88..f93058fd 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 } from "./requestDetail.js"; +import { buildRequestDetail, extractRequestConfig, extractUsageFromResponse, saveUsageStats, formatDoneLine } from "./requestDetail.js"; import { appendRequestLog, saveRequestDetail } from "@/lib/usageDb.js"; import { decloakToolNames } from "../../utils/claudeCloaking.js"; @@ -198,7 +198,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, pxpipe }) { +export async function handleNonStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, trackDone, appendLog, pxpipe, reqTag, log }) { trackDone(); const contentType = providerResponse.headers.get("content-type") || ""; let responseBody; @@ -235,7 +235,8 @@ 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, silent: true }); + if (log?.line) log.line(reqTag, "📊", formatDoneLine({ usage, latency: { total: Date.now() - requestStartTime } })); const translatedResponse = needsTranslation(targetFormat, sourceFormat) ? translateNonStreamingResponse(responseBody, targetFormat, sourceFormat) diff --git a/open-sse/handlers/chatCore/requestDetail.js b/open-sse/handlers/chatCore/requestDetail.js index 33122da8..b3d24115 100644 --- a/open-sse/handlers/chatCore/requestDetail.js +++ b/open-sse/handlers/chatCore/requestDetail.js @@ -75,7 +75,25 @@ export function buildRequestDetail(base, overrides = {}) { }; } -export function saveUsageStats({ provider, model, tokens, connectionId, apiKey, endpoint, label = "USAGE" }) { +// Build the "done" summary: duration, ttft, in/out tokens with cache breakdown +export function formatDoneLine({ usage, latency }) { + const u = usage || {}; + const inTok = u.prompt_tokens ?? u.input_tokens ?? 0; + const outTok = u.completion_tokens ?? u.output_tokens ?? 0; + const cacheRead = u.cache_read_input_tokens ?? u.cached_tokens ?? u.prompt_tokens_details?.cached_tokens ?? 0; + const cacheCreate = u.cache_creation_input_tokens ?? 0; + let inStr = `IN ${inTok}`; + if (cacheRead || cacheCreate) { + const parts = []; + if (cacheRead) parts.push(`↻${cacheRead}`); + if (cacheCreate) parts.push(`+${cacheCreate}`); + inStr += ` (CACHE ${parts.join(" ")})`; + } + const ttftStr = latency?.ttft ? ` · TTFT ${latency.ttft}ms` : ""; + return `DONE ${latency?.total ?? 0}ms${ttftStr} · ${inStr} · OUT ${outTok}`; +} + +export function saveUsageStats({ provider, model, tokens, connectionId, apiKey, endpoint, label = "USAGE", silent = false }) { if (!tokens || typeof tokens !== "object") return; const inTokens = tokens.input_tokens ?? tokens.prompt_tokens ?? 0; @@ -83,9 +101,11 @@ export function saveUsageStats({ provider, model, tokens, connectionId, apiKey, if (inTokens === 0 && outTokens === 0) return; - 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}`); + if (!silent) { + 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}`); + } // Canonicalize to one storage convention (prompt_tokens cache-inclusive) so // cached/cache-creation tokens survive to cost calc + stats. See canonicalizeUsage. diff --git a/open-sse/handlers/chatCore/sseToJsonHandler.js b/open-sse/handlers/chatCore/sseToJsonHandler.js index 1e0edeba..f25986c5 100644 --- a/open-sse/handlers/chatCore/sseToJsonHandler.js +++ b/open-sse/handlers/chatCore/sseToJsonHandler.js @@ -3,7 +3,7 @@ import { createErrorResult } from "../../utils/error.js"; import { HTTP_STATUS } from "../../config/runtimeConfig.js"; import { FORMATS } from "../../translator/formats.js"; import { PROVIDERS } from "../../config/providers.js"; -import { buildRequestDetail, extractRequestConfig, saveUsageStats } from "./requestDetail.js"; +import { buildRequestDetail, extractRequestConfig, saveUsageStats, formatDoneLine } from "./requestDetail.js"; // Responses-API providers (e.g. codex) may emit SSE without content-type + use Responses output shape const isResponsesProvider = (p) => PROVIDERS[p]?.format === FORMATS.OPENAI_RESPONSES; @@ -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, clientRawRequest, onRequestSuccess, trackDone, appendLog, reqTag, log }) { const contentType = providerResponse.headers.get("content-type") || ""; const isSSE = contentType.includes("text/event-stream") || (contentType === "" && isResponsesProvider(provider)); if (!isSSE) return null; // not handled here @@ -124,7 +124,8 @@ 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, silent: true }); + if (log?.line) log.line(reqTag, "📊", formatDoneLine({ usage, latency: { total: Date.now() - requestStartTime } })); const { msgItem, textContent } = pickAssistantMessageForChatCompletion(jsonResponse.output); const totalLatency = Date.now() - requestStartTime; @@ -200,7 +201,8 @@ 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, silent: true }); + if (log?.line) log.line(reqTag, "📊", formatDoneLine({ usage, latency: { total: Date.now() - requestStartTime } })); const totalLatency = Date.now() - requestStartTime; saveRequestDetail(buildRequestDetail({ diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index 29ee39f3..1fb8a322 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/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, saveUsageStats, formatDoneLine } from "./requestDetail.js"; import { saveRequestDetail } from "@/lib/usageDb.js"; import { SSE_HEADERS_CORS as SSE_HEADERS } from "../../utils/sseConstants.js"; @@ -43,7 +43,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, streamController, onStreamComplete, streamDetailId, pxpipe }) { +export async function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, streamController, onStreamComplete, streamDetailId, pxpipe, reqTag, log }) { if (onRequestSuccess) { Promise.resolve() .then(onRequestSuccess) @@ -67,7 +67,8 @@ export async function handleStreamingResponse({ providerResponse, provider, mode const shortMsg = sanitizedTitle || (bodyText.length < 200 ? bodyText.replace(/<[^>]*>/g, '').trim().slice(0, 160) : `Upstream returned non-SSE response (${upstreamContentType})`); const status = providerResponse.status || 502; - console.warn(`[STREAM] ${provider} | ${model} | blocked pipe: ${shortMsg} [${status}]`); + if (log?.errorLine) log.errorLine(reqTag, "✗", `BLOCKED ${status} · ${provider}/${model} · non-SSE (${upstreamContentType})\n ${shortMsg}`); + else console.warn(`[STREAM] ${provider} | ${model} | blocked pipe: ${shortMsg} [${status}]`); streamController?.handleError?.(new Error(`upstream non-SSE: ${status}`)); return { success: false, @@ -109,7 +110,7 @@ export async function handleStreamingResponse({ providerResponse, provider, mode /** * Build onStreamComplete callback for streaming usage tracking. */ -export function buildOnStreamComplete({ provider, model, connectionId, apiKey, requestStartTime, body, stream, finalBody, translatedBody, clientRawRequest, pxpipe }) { +export function buildOnStreamComplete({ provider, model, connectionId, apiKey, requestStartTime, body, stream, finalBody, translatedBody, clientRawRequest, pxpipe, reqTag, log }) { const streamDetailId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`; const onStreamComplete = (contentObj, usage, ttftAt) => { @@ -134,7 +135,9 @@ export function buildOnStreamComplete({ provider, model, connectionId, apiKey, r console.error("[RequestDetail] Failed to update streaming content:", err.message); }); - saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint, label: "STREAM USAGE" }); + // Persist stream usage to DB (no console line; the "📊 done" line below is authoritative) + saveUsageStats({ provider, model, tokens: usage, connectionId, apiKey, endpoint: clientRawRequest?.endpoint, label: "STREAM USAGE", silent: true }); + if (log?.line) log.line(reqTag, "📊", formatDoneLine({ usage, latency })); }; return { onStreamComplete, streamDetailId }; diff --git a/open-sse/utils/streamHandler.js b/open-sse/utils/streamHandler.js index b8a06e2f..7f04427d 100644 --- a/open-sse/utils/streamHandler.js +++ b/open-sse/utils/streamHandler.js @@ -15,16 +15,19 @@ function getTimeString() { * @param {string} options.provider - Provider name * @param {string} options.model - Model name */ -export function createStreamController({ onDisconnect, onError, log, provider, model } = {}) { +export function createStreamController({ onDisconnect, onError, log, provider, model, reqTag = "" } = {}) { const abortController = new AbortController(); const startTime = Date.now(); let disconnected = false; let abortTimeout = null; - const logStream = (status) => { + // Only abnormal terminations are logged; normal completion is covered by "📊 done". + // isError uses errorLine (always shown, ignores LOG_LEVEL) so failures survive quiet levels. + const logStream = (symbol, status, isError = false) => { const duration = Date.now() - startTime; - const p = provider?.toUpperCase() || "UNKNOWN"; - console.log(`[${getTimeString()}] 🌊 [STREAM] ${p} | ${model || "unknown"} | ${duration}ms | ${status}`); + const emit = isError ? log?.errorLine : log?.line; + if (emit) emit(reqTag, symbol, `${status} · ${provider}/${model} · ${duration}ms`); + else console.log(`[${getTimeString()}] ${symbol} ${provider}/${model} · ${status} · ${duration}ms`); }; return { @@ -38,7 +41,7 @@ export function createStreamController({ onDisconnect, onError, log, provider, m if (disconnected) return; disconnected = true; - logStream(`disconnect: ${reason}`); + logStream("⚡", `DISCONNECT: ${reason}`); dbg("CTRL", `${provider}/${model} | disconnect=${reason} | dur=${Date.now() - startTime}ms`); // Delay abort to allow cleanup @@ -49,13 +52,11 @@ export function createStreamController({ onDisconnect, onError, log, provider, m onDisconnect?.({ reason, duration: Date.now() - startTime }); }, - // Call when stream completes normally + // Call when stream completes normally (no line here — "📊 done" is authoritative) handleComplete: () => { if (disconnected) return; disconnected = true; - logStream("complete"); - if (abortTimeout) { clearTimeout(abortTimeout); abortTimeout = null; @@ -73,11 +74,11 @@ export function createStreamController({ onDisconnect, onError, log, provider, m } if (error.name === "AbortError") { - logStream("aborted"); + logStream("⚡", "ABORTED"); return; } - logStream(`error: ${error.message}`); + logStream("✗", `ERROR: ${error.message}${error.stack ? `\n ${error.stack}` : ""}`, true); onError?.(error); }, diff --git a/open-sse/utils/usageTracking.js b/open-sse/utils/usageTracking.js index 663f2eaf..24518ef3 100644 --- a/open-sse/utils/usageTracking.js +++ b/open-sse/utils/usageTracking.js @@ -4,6 +4,9 @@ import { FORMATS } from "../translator/formats.js"; +// Legacy per-chunk usage console line; off by default (superseded by "📊 done") +const DEBUG_USAGE = process.env.LOG_USAGE_VERBOSE === "1"; + // ANSI color codes export const COLORS = { reset: "\x1b[0m", @@ -401,6 +404,10 @@ export function estimateUsage(body, contentLength, targetFormat = FORMATS.OPENAI export function logUsage(provider, usage, model = null, connectionId = null, apiKey = null) { if (!usage || typeof usage !== "object") return; + // Console output moved to the unified "📊 done" line (streamingHandler). Kept as + // a no-op hook so callers stay unchanged; usage persistence happens via saveUsageStats. + if (!DEBUG_USAGE) return; + const p = provider?.toUpperCase() || "UNKNOWN"; // Support both formats: diff --git a/src/app/(dashboard)/dashboard/usage/components/OverviewCards.js b/src/app/(dashboard)/dashboard/usage/components/OverviewCards.js index 6bc65580..22020ff2 100644 --- a/src/app/(dashboard)/dashboard/usage/components/OverviewCards.js +++ b/src/app/(dashboard)/dashboard/usage/components/OverviewCards.js @@ -8,7 +8,7 @@ const fmtCost = (n) => `$${(n || 0).toFixed(2)}`; export default function OverviewCards({ stats }) { return ( -
+
Total Requests {fmt(stats.totalRequests)} diff --git a/src/lib/db/repos/usageRepo.js b/src/lib/db/repos/usageRepo.js index ce6c4761..6b1966cb 100644 --- a/src/lib/db/repos/usageRepo.js +++ b/src/lib/db/repos/usageRepo.js @@ -189,8 +189,7 @@ export function trackPendingRequest(model, provider, connectionId, started, erro lastErrorProvider.ts = Date.now(); } - const t = new Date().toLocaleTimeString("en-US", { hour12: false, hour: "2-digit", minute: "2-digit", second: "2-digit" }); - console.log(`[${t}] [PENDING] ${started ? "START" : "END"}${error ? " (ERROR)" : ""} | provider=${provider} | model=${model}`); + // [PENDING] console line removed; lifecycle is visible via "▶" and "📊 done" lines scheduleStatsEvent("pending"); } diff --git a/src/sse/handlers/chat.js b/src/sse/handlers/chat.js index a0c187f5..77f450c1 100644 --- a/src/sse/handlers/chat.js +++ b/src/sse/handlers/chat.js @@ -48,15 +48,9 @@ export async function handleChat(request, clientRawRequest = null) { } cacheClaudeHeaders(clientRawRequest.headers); - // Log request endpoint and model - const url = new URL(request.url); const modelStr = body.model; - // Count messages (support both messages[] and input[] formats) - const msgCount = body.messages?.length || body.input?.length || 0; - const toolCount = body.tools?.length || 0; - const effort = body.reasoning_effort || body.reasoning?.effort || null; - log.request("POST", `${url.pathname} | ${modelStr} | ${msgCount} msgs${toolCount ? ` | ${toolCount} tools` : ""}${effort ? ` | effort=${effort}` : ""}`); + // Request summary is emitted as the unified "▶" line in chatCore (has fmt/thinking/account) // Log API key (masked) const authHeader = request.headers.get("Authorization"); @@ -191,12 +185,7 @@ async function handleSingleModelChat(body, modelStr, clientRawRequest = null, re const { provider, model } = modelInfo; - // Log model routing (alias → actual model) - if (modelStr !== `${provider}/${model}`) { - log.info("ROUTING", `${modelStr} → ${provider}/${model}`); - } else { - log.info("ROUTING", `Provider: ${provider}, Model: ${model}`); - } + // Routing shown in the unified "▶" line (client model → provider/model) // Extract userAgent from request const userAgent = request?.headers?.get("user-agent") || ""; @@ -225,9 +214,7 @@ async function handleSingleModelChat(body, modelStr, clientRawRequest = null, re return errorResponse(lastStatus || HTTP_STATUS.SERVICE_UNAVAILABLE, lastError || "All accounts unavailable"); } - // Log account selection - log.info("AUTH", `\x1b[32mUsing ${provider} account: ${credentials.connectionName}\x1b[0m`); - + // Account selection shown in the unified "▶" line (acc:...) const refreshedCredentials = await checkAndRefreshToken(provider, credentials); // Ensure real project ID is available for providers that need it (P0 fix: cold miss) @@ -288,7 +275,7 @@ async function handleSingleModelChat(body, modelStr, clientRawRequest = null, re const { shouldFallback } = await markAccountUnavailable(credentials.connectionId, result.status, result.error, provider, model, result.resetsAtMs); if (shouldFallback) { - log.warn("AUTH", `Account ${credentials.connectionName} unavailable (${result.status}), trying fallback`); + log.warn("FALLBACK", `⇄ ACC:${credentials.connectionName} UNAVAILABLE (${result.status}) → NEXT ACCOUNT`); excludeConnectionIds.add(credentials.connectionId); lastError = result.error; lastStatus = result.status; diff --git a/src/sse/utils/logger.js b/src/sse/utils/logger.js index bb47ba5d..a9c631ed 100644 --- a/src/sse/utils/logger.js +++ b/src/sse/utils/logger.js @@ -13,6 +13,49 @@ function formatTime() { return new Date().toLocaleTimeString("en-US", { hour12: false }); } +// Colored-dot tags to correlate request lines by session (same session → same color) +const REQ_TAGS = ["🟢", "🔵", "🟣", "🟡", "🟠", "🔴", "⚪", "🟤"]; +let tagCursor = 0; + +// Allocate next rotating tag (fallback when no session seed available) +export function nextTag() { + const tag = REQ_TAGS[tagCursor % REQ_TAGS.length]; + tagCursor++; + return tag; +} + +// Stable tag derived from a session/connection seed: same seed always maps to the same color +export function tagForSession(seed) { + if (!seed) return nextTag(); + let h = 0; + for (let i = 0; i < seed.length; i++) h = (h * 31 + seed.charCodeAt(i)) | 0; + return REQ_TAGS[Math.abs(h) % REQ_TAGS.length]; +} + +// Print one correlated line: [time] tag symbol message +export function line(tag, symbol, message) { + if (LEVEL > LOG_LEVELS.INFO) return; + console.log(`[${formatTime()}] ${tag} ${symbol} ${message}`); +} + +// Like line() but always printed regardless of LOG_LEVEL (errors must never be hidden) +export function errorLine(tag, symbol, message) { + console.log(`[${formatTime()}] ${tag} ${symbol} ${message}`); +} + +// Format thinking intent for the request line ("high(10k)" / "off" / "auto") +export function fmtThink(intent) { + if (!intent || !intent.mode) return null; + if (intent.mode === "none") return "off"; + if (intent.mode === "auto") return "auto"; + if (intent.mode === "budget") { + const k = intent.budget >= 1000 ? `${Math.round(intent.budget / 1000)}k` : `${intent.budget}`; + return k; + } + if (intent.mode === "level") return intent.level; + return null; +} + function formatData(data) { if (!data) return ""; if (typeof data === "string") return data; @@ -40,7 +83,7 @@ export function info(tag, message, data) { export function warn(tag, message, data) { if (LEVEL <= LOG_LEVELS.WARN) { const dataStr = data ? ` ${formatData(data)}` : ""; - // console.warn(`[${formatTime()}] ⚠️ [${tag}] ${message}${dataStr}`); + console.warn(`[${formatTime()}] ⚠️ [${tag}] ${message}${dataStr}`); } }