Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions open-sse/services/sessionManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
*/
Expand Down
89 changes: 89 additions & 0 deletions open-sse/services/toolLatencyTracker.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
/**
* 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<string, ToolLatencyEntry>();

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<string, any> = {};
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.totalRequests++;
entry.ttftAfterToolSum += ttftMs;
entry.ttftCount++;
Comment on lines +81 to +84

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

In recordToolTtft, entry.totalRequests is not incremented when recording the TTFT latency. Since getToolLatencyByProvider uses entry.totalRequests as the measurementCount, any requests that only record TTFT via recordToolTtft will not be counted in the total measurement count, leading to inaccurate metrics reporting.

We should increment entry.totalRequests in recordToolTtft just like we do in recordToolLatency.

  const entry = metrics.get(provider)!;
  entry.totalRequests++;
  entry.ttftAfterToolSum += ttftMs;
  entry.ttftCount++;

}

export function resetToolLatency(): void {
metrics.clear();
}
79 changes: 78 additions & 1 deletion open-sse/utils/stream.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -602,6 +608,21 @@ export function createSSEStream(options: StreamOptions = {}) {
const passthroughResponsesReasoningSummarySeen = new Set<string>();
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<typeof generateSessionId>[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({
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -1285,8 +1307,25 @@ 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) {
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(
Expand All @@ -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 &&
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -1492,6 +1548,27 @@ export function createSSEStream(options: StreamOptions = {}) {
}
}

const translateHasContent =
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;
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
Expand Down
2 changes: 1 addition & 1 deletion src/sse/handlers/chat.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
7 changes: 4 additions & 3 deletions src/sse/handlers/chatHelpers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, unknown>).name : undefined;
if (typeof name === "string" && name.startsWith("auto/")) available.push(name);
}
} catch {
/* DB unavailable */
Expand Down Expand Up @@ -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;
Expand Down
61 changes: 61 additions & 0 deletions tests/unit/tool-latency-tracker.test.ts
Original file line number Diff line number Diff line change
@@ -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);
});