From 269bd973b4c6e35be48bcfbadafc21b9976eb85f Mon Sep 17 00:00:00 2001 From: pizzav-xyz Date: Thu, 4 Jun 2026 15:32:56 +0200 Subject: [PATCH 1/3] feat(sse): add cross-request tool-call latency tracking - Add toolLatencyTracker service for per-provider TTFT/gap metrics - Track tool finish timestamps across requests via session manager - Record latency from last tool chunk and from tool finish to first content - Add markToolFinish/consumeToolFinishTime to sessionManager - Integrate latency recording into SSE stream pipeline --- open-sse/services/sessionManager.ts | 25 +++++++ open-sse/services/toolLatencyTracker.ts | 88 +++++++++++++++++++++++++ open-sse/utils/stream.ts | 77 ++++++++++++++++++++++ src/sse/handlers/chat.ts | 2 +- src/sse/handlers/chatHelpers.ts | 7 +- 5 files changed, 195 insertions(+), 4 deletions(-) create mode 100644 open-sse/services/toolLatencyTracker.ts diff --git a/open-sse/services/sessionManager.ts b/open-sse/services/sessionManager.ts index 7dd67758597b..9abb4282e7db 100644 --- a/open-sse/services/sessionManager.ts +++ b/open-sse/services/sessionManager.ts @@ -12,6 +12,8 @@ interface SessionEntry { lastActive: number; requestCount: number; connectionId: string | null; + /** Timestamp when the last tool_calls finish_reason was emitted (cross-request TTFT tracking) */ + lastToolFinishAt?: number; } interface SessionFingerprintOptions { @@ -131,6 +133,29 @@ export function touchSession(sessionId: string | null, connectionId: string | nu } } +/** + * Persist the tool-finish timestamp so Request 2 (the follow-up after tool execution) + * can measure cross-request TTFT. + */ +export function markToolFinish(sessionId: string | null): void { + if (!sessionId) return; + const session = sessions.get(sessionId); + if (session) session.lastToolFinishAt = Date.now(); +} + +/** + * Consume the previously persisted tool-finish timestamp (one-shot: clears after read). + * Returns null if no pending timestamp or session not found. + */ +export function consumeToolFinishTime(sessionId: string | null): number | null { + if (!sessionId) return null; + const session = sessions.get(sessionId); + if (!session?.lastToolFinishAt) return null; + const ts = session.lastToolFinishAt; + session.lastToolFinishAt = undefined; + return ts; +} + /** * Get session info (for sticky routing decisions) */ diff --git a/open-sse/services/toolLatencyTracker.ts b/open-sse/services/toolLatencyTracker.ts new file mode 100644 index 000000000000..5973254a1dc4 --- /dev/null +++ b/open-sse/services/toolLatencyTracker.ts @@ -0,0 +1,88 @@ +/** + * Tracks per-provider tool call latency metrics. + * Resets on server restart. + */ + +interface ToolLatencyEntry { + totalRequests: number; + ttftAfterToolSum: number; + gapAfterToolSum: number; + ttftCount: number; + gapCount: number; +} + +const metrics = new Map(); + +export function recordToolLatency( + provider: string, + ttftAfterToolMs: number | null, + gapAfterToolMs: number | null +): void { + if (!provider) return; + + if (!metrics.has(provider)) { + metrics.set(provider, { + totalRequests: 0, + ttftAfterToolSum: 0, + gapAfterToolSum: 0, + ttftCount: 0, + gapCount: 0, + }); + } + + const entry = metrics.get(provider)!; + entry.totalRequests++; + + if (ttftAfterToolMs != null && ttftAfterToolMs >= 0) { + entry.ttftAfterToolSum += ttftAfterToolMs; + entry.ttftCount++; + } + + if (gapAfterToolMs != null && gapAfterToolMs >= 0) { + entry.gapAfterToolSum += gapAfterToolMs; + entry.gapCount++; + } +} + +export function getToolLatencyByProvider(): Record< + string, + { + avgTtftAfterToolMs: number; + avgGapAfterToolMs: number; + measurementCount: number; + } +> { + const result: Record = {}; + for (const [provider, entry] of metrics) { + result[provider] = { + avgTtftAfterToolMs: + entry.ttftCount > 0 ? Math.round(entry.ttftAfterToolSum / entry.ttftCount) : 0, + avgGapAfterToolMs: + entry.gapCount > 0 ? Math.round(entry.gapAfterToolSum / entry.gapCount) : 0, + measurementCount: entry.totalRequests, + }; + } + return result; +} + +export function recordToolTtft(provider: string, ttftMs: number): void { + if (!provider || ttftMs < 0) return; + + if (!metrics.has(provider)) { + metrics.set(provider, { + totalRequests: 0, + ttftAfterToolSum: 0, + gapAfterToolSum: 0, + ttftCount: 0, + gapCount: 0, + }); + } + + const entry = metrics.get(provider)!; + entry.ttftAfterToolSum += ttftMs; + entry.ttftCount++; +} + +export function resetToolLatency(): void { + metrics.clear(); +} diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 062cb808bbf6..daf1713591d9 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -29,6 +29,12 @@ import { extractThinkingFromContent, } from "../handlers/responseSanitizer.ts"; import { buildErrorBody } from "./error.ts"; +import { recordToolLatency } from "../services/toolLatencyTracker.ts"; +import { + generateSessionId, + markToolFinish, + consumeToolFinishTime, +} from "../services/sessionManager.ts"; /** * Race a response body read against a timeout. @@ -602,6 +608,21 @@ export function createSSEStream(options: StreamOptions = {}) { const passthroughResponsesReasoningSummarySeen = new Set(); const streamStartedAt = Date.now(); + let lastToolCallChunkTime: number | null = null; + let toolFinishTime: number | null = null; + let contentAfterToolSeen = false; + + // Cross-request tool latency: fingerprint the session from the request body + // so Request 2 can pick up the tool-finish timestamp left by Request 1. + const sessionId = generateSessionId(body as Parameters[0], { + provider: provider ?? undefined, + connectionId: connectionId ?? undefined, + }); + let pendingToolFinishTime: number | null = null; + try { + pendingToolFinishTime = consumeToolFinishTime(sessionId); + } catch {} + // Guard against duplicate [DONE] events — ensures exactly one per stream let doneSent = false; const providerPayloadCollector = createStructuredSSECollector({ @@ -1252,6 +1273,7 @@ export function createSSEStream(options: StreamOptions = {}) { // T18: Track if we saw tool calls & accumulate for call log if (delta?.tool_calls && delta.tool_calls.length > 0) { passthroughHasToolCalls = true; + lastToolCallChunkTime = Date.now(); for (const tc of delta.tool_calls) { // Key by index first — id only appears on the first delta in OpenAI streaming let key: string; @@ -1287,6 +1309,23 @@ export function createSSEStream(options: StreamOptions = {}) { const content = delta?.content || delta?.reasoning_content; if (content && typeof content === "string") { totalContentLength += content.length; + + if (!contentAfterToolSeen) { + const toolTs = toolFinishTime || pendingToolFinishTime; + const lastChunkTs = lastToolCallChunkTime; + if (toolTs || lastChunkTs) { + contentAfterToolSeen = true; + const now = Date.now(); + try { + recordToolLatency( + provider || "unknown", + toolTs ? now - toolTs : null, + lastChunkTs ? now - lastChunkTs : null + ); + } catch {} + pendingToolFinishTime = null; + } + } } if (typeof delta?.content === "string") passthroughAccumulatedContent = appendBoundedText( @@ -1306,6 +1345,13 @@ export function createSSEStream(options: StreamOptions = {}) { const isFinishChunk = parsed.choices?.[0]?.finish_reason; + if (isFinishChunk && passthroughHasToolCalls) { + toolFinishTime = Date.now(); + try { + markToolFinish(sessionId); + } catch {} + } + // T18: Normalize finish_reason to 'tool_calls' if tool calls were used if ( isFinishChunk && @@ -1395,6 +1441,16 @@ export function createSSEStream(options: StreamOptions = {}) { continue; } + if (parsed.choices?.[0]?.delta?.tool_calls) { + lastToolCallChunkTime = Date.now(); + } + if (parsed.choices?.[0]?.finish_reason === "tool_calls") { + toolFinishTime = Date.now(); + try { + markToolFinish(sessionId); + } catch {} + } + // Track content length and accumulate for call log (from raw provider chunk, so content is never missed) // Do this before translation so we capture content regardless of translator output shape @@ -1492,6 +1548,27 @@ export function createSSEStream(options: StreamOptions = {}) { } } + const translateHasContent = + parsed.delta?.text || + parsed.choices?.[0]?.delta?.content || + parsed.choices?.[0]?.delta?.reasoning_content; + if (translateHasContent && !contentAfterToolSeen) { + const toolTs = toolFinishTime || pendingToolFinishTime; + const lastChunkTs = lastToolCallChunkTime; + if (toolTs || lastChunkTs) { + contentAfterToolSeen = true; + const now = Date.now(); + try { + recordToolLatency( + provider || "unknown", + toolTs ? now - toolTs : null, + lastChunkTs ? now - lastChunkTs : null + ); + } catch {} + pendingToolFinishTime = null; + } + } + // Extract usage const extracted = extractUsage(parsed); if (extracted) state.usage = extracted; // Keep original usage for logging diff --git a/src/sse/handlers/chat.ts b/src/sse/handlers/chat.ts index 7c31d8507f3d..18d38704d8b1 100644 --- a/src/sse/handlers/chat.ts +++ b/src/sse/handlers/chat.ts @@ -902,7 +902,7 @@ async function handleSingleModelChat( refreshedCredentials ); } - const proxyInfo = await safeResolveProxy(credentials.connectionId); + const proxyInfo = await safeResolveProxy(credentials.connectionId, apiKeyInfo?.id); const proxyStartTime = Date.now(); // 4. Execute chat via core after breaker gate checks (with optional TLS tracking) diff --git a/src/sse/handlers/chatHelpers.ts b/src/sse/handlers/chatHelpers.ts index 658e4a4cb8c1..f3e29d41b427 100644 --- a/src/sse/handlers/chatHelpers.ts +++ b/src/sse/handlers/chatHelpers.ts @@ -159,7 +159,8 @@ export async function resolveModelOrError( const { getCombos } = await import("@/lib/localDb"); const all = await getCombos(); for (const c of all) { - if (c.name?.startsWith("auto/")) available.push(c.name); + const name = typeof c === "object" && c !== null ? (c as Record).name : undefined; + if (typeof name === "string" && name.startsWith("auto/")) available.push(name); } } catch { /* DB unavailable */ @@ -507,9 +508,9 @@ export function handleNoCredentials( ); } -export async function safeResolveProxy(connectionId: string) { +export async function safeResolveProxy(connectionId: string, apiKeyId?: string) { try { - return await resolveProxyForConnection(connectionId); + return await resolveProxyForConnection(connectionId, apiKeyId); } catch (proxyErr: any) { log.debug("PROXY", `Failed to resolve proxy: ${proxyErr.message}`); return null; From d3aa15bbe1b714c08b3ec3b2282a9943ef5b00ba Mon Sep 17 00:00:00 2001 From: pizzav-xyz Date: Thu, 4 Jun 2026 22:57:52 +0200 Subject: [PATCH 2/3] fix(tool-latency): fix empty-string content detection + increment totalRequests in recordToolTtft --- open-sse/services/toolLatencyTracker.ts | 1 + open-sse/utils/stream.ts | 8 ++++---- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/open-sse/services/toolLatencyTracker.ts b/open-sse/services/toolLatencyTracker.ts index 5973254a1dc4..85af5fa9185f 100644 --- a/open-sse/services/toolLatencyTracker.ts +++ b/open-sse/services/toolLatencyTracker.ts @@ -79,6 +79,7 @@ export function recordToolTtft(provider: string, ttftMs: number): void { } const entry = metrics.get(provider)!; + entry.totalRequests++; entry.ttftAfterToolSum += ttftMs; entry.ttftCount++; } diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index daf1713591d9..730ad5ba5110 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -1307,7 +1307,7 @@ export function createSSEStream(options: StreamOptions = {}) { } const content = delta?.content || delta?.reasoning_content; - if (content && typeof content === "string") { + if (typeof content === "string") { totalContentLength += content.length; if (!contentAfterToolSeen) { @@ -1549,9 +1549,9 @@ export function createSSEStream(options: StreamOptions = {}) { } const translateHasContent = - parsed.delta?.text || - parsed.choices?.[0]?.delta?.content || - parsed.choices?.[0]?.delta?.reasoning_content; + typeof parsed.delta?.text === "string" || + typeof parsed.choices?.[0]?.delta?.content === "string" || + typeof parsed.choices?.[0]?.delta?.reasoning_content === "string"; if (translateHasContent && !contentAfterToolSeen) { const toolTs = toolFinishTime || pendingToolFinishTime; const lastChunkTs = lastToolCallChunkTime; From 947e57907785d70f69ccebfbee2388e9b60e34e8 Mon Sep 17 00:00:00 2001 From: diegosouzapw Date: Thu, 4 Jun 2026 18:18:04 -0300 Subject: [PATCH 3/3] test(metrics): cover tool latency tracker + session tool-finish one-shot (#3173) --- tests/unit/tool-latency-tracker.test.ts | 61 +++++++++++++++++++++++++ 1 file changed, 61 insertions(+) create mode 100644 tests/unit/tool-latency-tracker.test.ts diff --git a/tests/unit/tool-latency-tracker.test.ts b/tests/unit/tool-latency-tracker.test.ts new file mode 100644 index 000000000000..5c0378e56534 --- /dev/null +++ b/tests/unit/tool-latency-tracker.test.ts @@ -0,0 +1,61 @@ +/** + * #3173 — cross-request tool latency metrics. + * + * toolLatencyTracker aggregates per-provider TTFT/gap after tool calls; + * sessionManager persists a one-shot tool-finish timestamp so the follow-up + * request (Request 2) can measure cross-request TTFT. + */ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { + recordToolLatency, + getToolLatencyByProvider, + recordToolTtft, + resetToolLatency, +} = await import("../../open-sse/services/toolLatencyTracker.ts"); +const { markToolFinish, consumeToolFinishTime } = await import( + "../../open-sse/services/sessionManager.ts" +); + +test("recordToolLatency averages TTFT/gap per provider and counts requests", () => { + resetToolLatency(); + recordToolLatency("openai", 100, 40); + recordToolLatency("openai", 300, 60); + recordToolLatency("anthropic", 200, null); + + const m = getToolLatencyByProvider(); + assert.equal(m.openai.avgTtftAfterToolMs, 200); // (100+300)/2 + assert.equal(m.openai.avgGapAfterToolMs, 50); // (40+60)/2 + assert.equal(m.openai.measurementCount, 2); + assert.equal(m.anthropic.avgTtftAfterToolMs, 200); + assert.equal(m.anthropic.avgGapAfterToolMs, 0); // no gap samples + assert.equal(m.anthropic.measurementCount, 1); +}); + +test("recordToolLatency ignores null/negative samples but still counts the request", () => { + resetToolLatency(); + recordToolLatency("glm", null, null); + recordToolLatency("glm", -5, -5); + const m = getToolLatencyByProvider(); + assert.equal(m.glm.measurementCount, 2); + assert.equal(m.glm.avgTtftAfterToolMs, 0); + assert.equal(m.glm.avgGapAfterToolMs, 0); +}); + +test("recordToolTtft accumulates and resetToolLatency clears", () => { + resetToolLatency(); + recordToolTtft("codex", 120); + recordToolTtft("codex", 80); + assert.equal(getToolLatencyByProvider().codex.avgTtftAfterToolMs, 100); + resetToolLatency(); + assert.deepEqual(getToolLatencyByProvider(), {}); +}); + +test("markToolFinish/consumeToolFinishTime is a one-shot per session", () => { + // null session is a no-op / null + markToolFinish(null); + assert.equal(consumeToolFinishTime(null), null); + // unknown session (never touched) → null + assert.equal(consumeToolFinishTime("sess-never-seen-xyz"), null); +});