From 8e5094fcd9123e47e41f220e15f922effc5501e6 Mon Sep 17 00:00:00 2001 From: John Date: Thu, 2 Jul 2026 07:45:47 +0800 Subject: [PATCH] feat: record first token latency in runtime router --- .../portal/scripts/runtime-slo-report.mjs | 5 +++ .runtime/portal/server.mjs | 36 +++++++++++++++- scripts/runtime-slo-report.mjs | 5 +++ tkmind-proxy.mjs | 43 ++++++++++++++++++- 4 files changed, 86 insertions(+), 3 deletions(-) diff --git a/.runtime/portal/scripts/runtime-slo-report.mjs b/.runtime/portal/scripts/runtime-slo-report.mjs index cb74e19..c07e7ec 100755 --- a/.runtime/portal/scripts/runtime-slo-report.mjs +++ b/.runtime/portal/scripts/runtime-slo-report.mjs @@ -155,6 +155,10 @@ function summarizeRuntime(runtimeJson) { streamOpenCount: worker.streamOpenCount, streamAbortCount: worker.streamAbortCount, streamErrorCount: worker.streamErrorCount, + ewmaFirstTokenMs: worker.ewmaFirstTokenMs, + lastFirstTokenMs: worker.lastFirstTokenMs, + lastFirstTokenAt: worker.lastFirstTokenAt, + firstTokenCount: worker.firstTokenCount, cpuLoad: worker.cpuLoad, memoryPressure: worker.memoryPressure, fdPressure: worker.fdPressure, @@ -213,6 +217,7 @@ for (const worker of runtimeSummary?.workers ?? []) { if (worker.containerHealth && worker.containerHealth !== 'healthy') failures.push(`${worker.id}_container_${worker.containerHealth}`); if (!worker.metricsFresh) failures.push(`${worker.id}_metrics_stale`); if (worker.streamErrorCount > 0) failures.push(`${worker.id}_stream_errors_${worker.streamErrorCount}`); + if (worker.firstTokenCount > 0 && worker.ewmaFirstTokenMs <= 0) failures.push(`${worker.id}_first_token_ewma_missing`); } if (runtimeSummary?.toolRuntime?.chatInjectsCodeTools !== false) failures.push('chat_injects_code_tools'); diff --git a/.runtime/portal/server.mjs b/.runtime/portal/server.mjs index e0f0ed4..ac8dca2 100644 --- a/.runtime/portal/server.mjs +++ b/.runtime/portal/server.mjs @@ -8768,6 +8768,17 @@ function createRuntimeRouter({ if (streamKey) multi.set(streamKey, status, { EX: 600 }); await multi.exec().catch(() => null); }, + async firstTokenObserved(target, latencyMs) { + const client = await getClient(); + if (!client || !target) return; + const workerId = workerIdForTarget(target); + const sample = Math.max(0, Math.round(Number(latencyMs) || 0)); + const ewmaKey = key("worker", workerId, "ewma_first_token_ms"); + const prev = readNumber(await client.get(ewmaKey).catch(() => null)); + const next = prev > 0 ? Math.round(prev * 0.8 + sample * 0.2) : sample; + const now = String(Date.now()); + await client.multi().set(ewmaKey, String(next)).set(key("worker", workerId, "last_first_token_ms"), String(sample)).set(key("worker", workerId, "last_first_token_at"), now).incr(key("worker", workerId, "first_token_count")).set(key("worker", workerId, "heartbeat"), now, { EX: 30 }).exec().catch(() => null); + }, async getStatus() { const client = await getClient(); const workers = []; @@ -8791,7 +8802,10 @@ function createRuntimeRouter({ key("worker", workerId, "fd_count"), key("worker", workerId, "container_pids"), key("worker", workerId, "container_health"), - key("worker", workerId, "metrics_sampled_at") + key("worker", workerId, "metrics_sampled_at"), + key("worker", workerId, "last_first_token_ms"), + key("worker", workerId, "last_first_token_at"), + key("worker", workerId, "first_token_count") ]).catch(() => []) : []; workers.push({ id: workerId, @@ -8814,6 +8828,9 @@ function createRuntimeRouter({ containerPids: readNumber(values?.[15]), containerHealth: values?.[16] ?? null, metricsSampledAt: values?.[17] ? Number(values[17]) : null, + lastFirstTokenMs: readNumber(values?.[18]), + lastFirstTokenAt: values?.[19] ? Number(values[19]) : null, + firstTokenCount: readNumber(values?.[20]), score: workerScoreFromValues(values) }); } @@ -9344,6 +9361,10 @@ function createTkmindProxy({ if (!runtimeRouter || !sessionId || !target) return; await runtimeRouter.streamEnded(sessionId, target, options).catch(() => null); } + function markFirstTokenObserved(target, latencyMs) { + if (!runtimeRouter || !target) return; + void runtimeRouter.firstTokenObserved(target, latencyMs).catch(() => null); + } async function getRuntimeStatus() { const targetStatuses = []; for (const target of targets) { @@ -9859,6 +9880,7 @@ function createTkmindProxy({ ]; const proxySessionEvents = async (req, res, sessionId, { onAfterFinish, onEvent } = {}) => { const upstreamAbort = new AbortController(); + const streamRequestedAt = Date.now(); let clientClosed = false; const abortUpstream = () => { clientClosed = true; @@ -9916,6 +9938,16 @@ function createTkmindProxy({ } }); const source = Readable.fromWeb(upstream.body); + let firstChunkSeen = false; + const firstTokenProbe = new Transform2({ + transform(chunk, _encoding, callback) { + if (!firstChunkSeen && chunk?.length > 0) { + firstChunkSeen = true; + markFirstTokenObserved(sessionTarget, Date.now() - streamRequestedAt); + } + callback(null, chunk); + } + }); const linkSanitizer = createSessionEventSanitizer(req.currentUser, { onEvent }); const waitForDrain = () => new Promise((resolve) => res.once("drain", resolve)); const writeClientChunk = async (chunk) => { @@ -9945,7 +9977,7 @@ function createTkmindProxy({ let streamCloseStatus = "closed"; try { try { - await pipeline(source, linkSanitizer, billingTransform, clientSink); + await pipeline(source, firstTokenProbe, linkSanitizer, billingTransform, clientSink); } catch (err) { streamCloseStatus = clientClosed || upstreamAbort.signal.aborted ? "aborted" : "error"; throw err; diff --git a/scripts/runtime-slo-report.mjs b/scripts/runtime-slo-report.mjs index cb74e19..c07e7ec 100755 --- a/scripts/runtime-slo-report.mjs +++ b/scripts/runtime-slo-report.mjs @@ -155,6 +155,10 @@ function summarizeRuntime(runtimeJson) { streamOpenCount: worker.streamOpenCount, streamAbortCount: worker.streamAbortCount, streamErrorCount: worker.streamErrorCount, + ewmaFirstTokenMs: worker.ewmaFirstTokenMs, + lastFirstTokenMs: worker.lastFirstTokenMs, + lastFirstTokenAt: worker.lastFirstTokenAt, + firstTokenCount: worker.firstTokenCount, cpuLoad: worker.cpuLoad, memoryPressure: worker.memoryPressure, fdPressure: worker.fdPressure, @@ -213,6 +217,7 @@ for (const worker of runtimeSummary?.workers ?? []) { if (worker.containerHealth && worker.containerHealth !== 'healthy') failures.push(`${worker.id}_container_${worker.containerHealth}`); if (!worker.metricsFresh) failures.push(`${worker.id}_metrics_stale`); if (worker.streamErrorCount > 0) failures.push(`${worker.id}_stream_errors_${worker.streamErrorCount}`); + if (worker.firstTokenCount > 0 && worker.ewmaFirstTokenMs <= 0) failures.push(`${worker.id}_first_token_ewma_missing`); } if (runtimeSummary?.toolRuntime?.chatInjectsCodeTools !== false) failures.push('chat_injects_code_tools'); diff --git a/tkmind-proxy.mjs b/tkmind-proxy.mjs index f063abd..8ca0ca1 100644 --- a/tkmind-proxy.mjs +++ b/tkmind-proxy.mjs @@ -219,6 +219,25 @@ function createRuntimeRouter({ if (streamKey) multi.set(streamKey, status, { EX: 600 }); await multi.exec().catch(() => null); }, + async firstTokenObserved(target, latencyMs) { + const client = await getClient(); + if (!client || !target) return; + const workerId = workerIdForTarget(target); + const sample = Math.max(0, Math.round(Number(latencyMs) || 0)); + const ewmaKey = key('worker', workerId, 'ewma_first_token_ms'); + const prev = readNumber(await client.get(ewmaKey).catch(() => null)); + const next = prev > 0 ? Math.round(prev * 0.8 + sample * 0.2) : sample; + const now = String(Date.now()); + await client + .multi() + .set(ewmaKey, String(next)) + .set(key('worker', workerId, 'last_first_token_ms'), String(sample)) + .set(key('worker', workerId, 'last_first_token_at'), now) + .incr(key('worker', workerId, 'first_token_count')) + .set(key('worker', workerId, 'heartbeat'), now, { EX: 30 }) + .exec() + .catch(() => null); + }, async getStatus() { const client = await getClient(); const workers = []; @@ -245,6 +264,9 @@ function createRuntimeRouter({ key('worker', workerId, 'container_pids'), key('worker', workerId, 'container_health'), key('worker', workerId, 'metrics_sampled_at'), + key('worker', workerId, 'last_first_token_ms'), + key('worker', workerId, 'last_first_token_at'), + key('worker', workerId, 'first_token_count'), ]) .catch(() => []) : []; @@ -269,6 +291,9 @@ function createRuntimeRouter({ containerPids: readNumber(values?.[15]), containerHealth: values?.[16] ?? null, metricsSampledAt: values?.[17] ? Number(values[17]) : null, + lastFirstTokenMs: readNumber(values?.[18]), + lastFirstTokenAt: values?.[19] ? Number(values[19]) : null, + firstTokenCount: readNumber(values?.[20]), score: workerScoreFromValues(values), }); } @@ -909,6 +934,11 @@ export function createTkmindProxy({ await runtimeRouter.streamEnded(sessionId, target, options).catch(() => null); } + function markFirstTokenObserved(target, latencyMs) { + if (!runtimeRouter || !target) return; + void runtimeRouter.firstTokenObserved(target, latencyMs).catch(() => null); + } + async function getRuntimeStatus() { const targetStatuses = []; for (const target of targets) { @@ -1503,6 +1533,7 @@ export function createTkmindProxy({ const proxySessionEvents = async (req, res, sessionId, { onAfterFinish, onEvent } = {}) => { const upstreamAbort = new AbortController(); + const streamRequestedAt = Date.now(); let clientClosed = false; const abortUpstream = () => { clientClosed = true; @@ -1564,6 +1595,16 @@ export function createTkmindProxy({ }); const source = Readable.fromWeb(upstream.body); + let firstChunkSeen = false; + const firstTokenProbe = new Transform({ + transform(chunk, _encoding, callback) { + if (!firstChunkSeen && chunk?.length > 0) { + firstChunkSeen = true; + markFirstTokenObserved(sessionTarget, Date.now() - streamRequestedAt); + } + callback(null, chunk); + }, + }); const linkSanitizer = createSessionEventSanitizer(req.currentUser, { onEvent }); const waitForDrain = () => new Promise((resolve) => res.once('drain', resolve)); const writeClientChunk = async (chunk) => { @@ -1595,7 +1636,7 @@ export function createTkmindProxy({ let streamCloseStatus = 'closed'; try { try { - await pipeline(source, linkSanitizer, billingTransform, clientSink); + await pipeline(source, firstTokenProbe, linkSanitizer, billingTransform, clientSink); } catch (err) { streamCloseStatus = clientClosed || upstreamAbort.signal.aborted ? 'aborted' : 'error'; throw err;