diff --git a/.changeset/ttft-client-server-split.md b/.changeset/ttft-client-server-split.md new file mode 100644 index 0000000000..3e8b5c7efa --- /dev/null +++ b/.changeset/ttft-client-server-split.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Split LLM streaming timing in the session log and `KIMI_CODE_DEBUG=1` output into client vs. API-server portions, so slow turns can be attributed without parsing the wire log. Time-to-first-token splits into the API-server portion (network + server) and the client portion (in-process request building); the decode window splits into time awaiting tokens from the server and time the client spends processing each streamed chunk. diff --git a/apps/kimi-code/src/utils/usage/debug-timing.ts b/apps/kimi-code/src/utils/usage/debug-timing.ts index ab87ebdd87..87f72696c1 100644 --- a/apps/kimi-code/src/utils/usage/debug-timing.ts +++ b/apps/kimi-code/src/utils/usage/debug-timing.ts @@ -10,6 +10,20 @@ interface DebugTokenUsage { export interface StepTimingInput { readonly llmFirstTokenLatencyMs?: number; readonly llmStreamDurationMs?: number; + /** + * Split of `llmFirstTokenLatencyMs` into the client-side request-build + * portion (`llmRequestBuildMs`) and the network + API-server portion + * (`llmServerFirstTokenMs`). Both present together or not at all. + */ + readonly llmRequestBuildMs?: number; + readonly llmServerFirstTokenMs?: number; + /** + * Split of `llmStreamDurationMs` (the decode window) into server time spent + * awaiting parts (`llmServerDecodeMs`) and client time spent processing parts + * (`llmClientConsumeMs`). Both present together or not at all. + */ + readonly llmServerDecodeMs?: number; + readonly llmClientConsumeMs?: number; readonly usage?: DebugTokenUsage; } @@ -26,12 +40,14 @@ export function formatStepDebugTiming(input: StepTimingInput): string | undefine const streamMs = input.llmStreamDurationMs; if (latency === undefined || streamMs === undefined) return undefined; - const parts: string[] = [`TTFT: ${formatDuration(latency)}`]; + const parts: string[] = [`TTFT: ${formatTtft(input)}`]; const outputTokens = input.usage?.output; if (outputTokens !== undefined && outputTokens > 0) { if (streamMs >= MIN_STREAM_MS_FOR_TPS) { const tps = (outputTokens / (streamMs / 1000)).toFixed(1); - parts.push(`TPS: ${tps} tok/s (${outputTokens} tokens in ${formatDuration(streamMs)})`); + parts.push( + `TPS: ${tps} tok/s (${outputTokens} tokens in ${formatDuration(streamMs)}${formatDecodeSplit(input)})`, + ); } else { parts.push( `${outputTokens} tokens in ${formatDuration(streamMs)} (stream too short for TPS)`, @@ -65,6 +81,27 @@ function usageInputTotal(usage: DebugTokenUsage | undefined): number { return (usage.inputOther ?? 0) + (usage.inputCacheRead ?? 0) + (usage.inputCacheCreation ?? 0); } +// Render TTFT, splitting the latency into the network + API-server portion and +// the in-process request-build portion when the provider reported the +// boundary. Falls back to the bare total otherwise. +function formatTtft(input: StepTimingInput): string { + const total = formatDuration(input.llmFirstTokenLatencyMs ?? 0); + const build = input.llmRequestBuildMs; + const server = input.llmServerFirstTokenMs; + if (build === undefined || server === undefined) return total; + return `${total} (api ${formatDuration(server)} + client ${formatDuration(build)})`; +} + +// Render the decode-window split as a trailing clause, e.g. +// `; server 4.6s + client 0.4s`. A large client share means the host's per-part +// processing is throttling decode. Empty when the provider did not report it. +function formatDecodeSplit(input: StepTimingInput): string { + const server = input.llmServerDecodeMs; + const client = input.llmClientConsumeMs; + if (server === undefined || client === undefined) return ''; + return `; server ${formatDuration(server)} + client ${formatDuration(client)}`; +} + function formatDuration(ms: number): string { if (ms < 1000) return `${Math.round(ms)}ms`; return `${(ms / 1000).toFixed(1)}s`; diff --git a/apps/kimi-code/test/utils/usage/debug-timing.test.ts b/apps/kimi-code/test/utils/usage/debug-timing.test.ts index feb8572046..353b10ee53 100644 --- a/apps/kimi-code/test/utils/usage/debug-timing.test.ts +++ b/apps/kimi-code/test/utils/usage/debug-timing.test.ts @@ -89,6 +89,52 @@ describe('formatStepDebugTiming', () => { expect(result).toContain('900ms'); }); + it('splits TTFT into api-server and client portions when both are present', () => { + const result = formatStepDebugTiming({ + llmFirstTokenLatencyMs: 2500, + llmStreamDurationMs: 5000, + llmServerFirstTokenMs: 2400, + llmRequestBuildMs: 100, + usage: { output: 200 }, + }); + expect(result).toBe( + '[Debug] TTFT: 2.5s (api 2.4s + client 100ms) | TPS: 40.0 tok/s (200 tokens in 5.0s)', + ); + }); + + it('falls back to the bare TTFT when only one split component is present', () => { + const result = formatStepDebugTiming({ + llmFirstTokenLatencyMs: 800, + llmStreamDurationMs: 5000, + llmServerFirstTokenMs: 700, + usage: { output: 0 }, + }); + expect(result).toBe('[Debug] TTFT: 800ms'); + }); + + it('appends the decode wait/consume split to the TPS clause', () => { + const result = formatStepDebugTiming({ + llmFirstTokenLatencyMs: 800, + llmStreamDurationMs: 5000, + llmServerDecodeMs: 4600, + llmClientConsumeMs: 400, + usage: { output: 200 }, + }); + expect(result).toBe( + '[Debug] TTFT: 800ms | TPS: 40.0 tok/s (200 tokens in 5.0s; server 4.6s + client 400ms)', + ); + }); + + it('omits the decode split when only one component is present', () => { + const result = formatStepDebugTiming({ + llmFirstTokenLatencyMs: 800, + llmStreamDurationMs: 5000, + llmServerDecodeMs: 4600, + usage: { output: 200 }, + }); + expect(result).toBe('[Debug] TTFT: 800ms | TPS: 40.0 tok/s (200 tokens in 5.0s)'); + }); + it('formats durations at or above 1s as seconds', () => { const result = formatStepDebugTiming({ llmFirstTokenLatencyMs: 1500, diff --git a/apps/vis/web/src/components/analysis/TimelineTab.tsx b/apps/vis/web/src/components/analysis/TimelineTab.tsx index 39c0cca380..1b082142ec 100644 --- a/apps/vis/web/src/components/analysis/TimelineTab.tsx +++ b/apps/vis/web/src/components/analysis/TimelineTab.tsx @@ -290,7 +290,27 @@ function StepRow({ step, turnDurationMs }: { step: StepNode; turnDurationMs?: nu ) : null} {formatDuration(step.durationMs)} {step.llmFirstTokenLatencyMs !== undefined ? ( - ttft {step.llmFirstTokenLatencyMs}ms + + ttft {step.llmFirstTokenLatencyMs}ms + {step.llmServerFirstTokenMs !== undefined && step.llmRequestBuildMs !== undefined + ? ` (api ${step.llmServerFirstTokenMs} + client ${step.llmRequestBuildMs})` + : ''} + + ) : null} + {step.llmServerDecodeMs !== undefined && step.llmClientConsumeMs !== undefined ? ( + + decode {step.llmServerDecodeMs}+{step.llmClientConsumeMs}ms + ) : null} {step.contextTokens !== undefined ? ( ctx {formatTokens(step.contextTokens)} diff --git a/apps/vis/web/src/components/wire/parts.tsx b/apps/vis/web/src/components/wire/parts.tsx index 759c15f8c6..246900bf48 100644 --- a/apps/vis/web/src/components/wire/parts.tsx +++ b/apps/vis/web/src/components/wire/parts.tsx @@ -362,11 +362,31 @@ export function LoopEventDetail({ event }: { event: LoopRecordedEvent }) { {event.llmFirstTokenLatencyMs} ms ) : null} + {event.llmServerFirstTokenMs !== undefined ? ( + + {event.llmServerFirstTokenMs} ms + + ) : null} + {event.llmRequestBuildMs !== undefined ? ( + + {event.llmRequestBuildMs} ms + + ) : null} {event.llmStreamDurationMs !== undefined ? ( {event.llmStreamDurationMs} ms ) : null} + {event.llmServerDecodeMs !== undefined ? ( + + {event.llmServerDecodeMs} ms + + ) : null} + {event.llmClientConsumeMs !== undefined ? ( + + {event.llmClientConsumeMs} ms + + ) : null} {usage !== undefined ? (
diff --git a/apps/vis/web/src/lib/analysis.ts b/apps/vis/web/src/lib/analysis.ts index c8bef2bd5a..fe4f64ba6a 100644 --- a/apps/vis/web/src/lib/analysis.ts +++ b/apps/vis/web/src/lib/analysis.ts @@ -54,6 +54,12 @@ export interface StepNode { contextTokens?: number; llmFirstTokenLatencyMs?: number; llmStreamDurationMs?: number; + /** TTFT split: client-side request-build vs. network + API-server time. */ + llmRequestBuildMs?: number; + llmServerFirstTokenMs?: number; + /** Decode split: server time awaiting parts vs. client time processing them. */ + llmServerDecodeMs?: number; + llmClientConsumeMs?: number; content: ContentSummary; toolCalls: ToolCallNode[]; } @@ -317,6 +323,10 @@ export function analyzeWire(entries: readonly WireEntry[]): Analysis { step.finishReason = ev.finishReason; step.llmFirstTokenLatencyMs = ev.llmFirstTokenLatencyMs; step.llmStreamDurationMs = ev.llmStreamDurationMs; + step.llmRequestBuildMs = ev.llmRequestBuildMs; + step.llmServerFirstTokenMs = ev.llmServerFirstTokenMs; + step.llmServerDecodeMs = ev.llmServerDecodeMs; + step.llmClientConsumeMs = ev.llmClientConsumeMs; if (step.beginTime !== undefined && t !== undefined) step.durationMs = t - step.beginTime; // Steps don't carry a generic 'error' finish reason (errors are // thrown, not recorded). 'filtered' means the provider blocked the diff --git a/packages/agent-core/src/agent/turn/index.ts b/packages/agent-core/src/agent/turn/index.ts index df115b6d5a..0765e6ca4e 100644 --- a/packages/agent-core/src/agent/turn/index.ts +++ b/packages/agent-core/src/agent/turn/index.ts @@ -997,6 +997,10 @@ function mapLoopEvent(event: LoopEvent, turnId: number): AgentEvent | undefined finishReason: event.finishReason, llmFirstTokenLatencyMs: event.llmFirstTokenLatencyMs, llmStreamDurationMs: event.llmStreamDurationMs, + llmRequestBuildMs: event.llmRequestBuildMs, + llmServerFirstTokenMs: event.llmServerFirstTokenMs, + llmServerDecodeMs: event.llmServerDecodeMs, + llmClientConsumeMs: event.llmClientConsumeMs, providerFinishReason: event.providerFinishReason, rawFinishReason: event.rawFinishReason, }; diff --git a/packages/agent-core/src/agent/turn/kosong-llm.ts b/packages/agent-core/src/agent/turn/kosong-llm.ts index ef3e2b8bf6..d5812cf034 100644 --- a/packages/agent-core/src/agent/turn/kosong-llm.ts +++ b/packages/agent-core/src/agent/turn/kosong-llm.ts @@ -25,6 +25,7 @@ import { type GenerateCallbacks, type Message, type ModelCapability, + type StreamDecodeStats, type StreamedMessagePart, } from '@moonshot-ai/kosong'; @@ -87,13 +88,19 @@ export class KosongLLM implements LLM { async chat(params: LLMChatParams): Promise { let requestStartedAt = Date.now(); + let requestSentAt: number | undefined; let firstChunkAt: number | undefined; let streamEndedAt: number | undefined; + let decodeStats: StreamDecodeStats | undefined; const markRequestStart = (): void => { requestStartedAt = Date.now(); }; - const markStreamEnd = (): void => { + const markRequestSent = (): void => { + requestSentAt ??= Date.now(); + }; + const markStreamEnd = (stats?: StreamDecodeStats): void => { streamEndedAt = Date.now(); + decodeStats = stats; }; const markStreamOutput = (): void => { firstChunkAt ??= Date.now(); @@ -113,6 +120,7 @@ export class KosongLLM implements LLM { const options: GenerateOptionsWithRequestLogFields = { signal: params.signal, onRequestStart: markRequestStart, + onRequestSent: markRequestSent, onStreamEnd: markStreamEnd, requestLogFields: params.requestLogFields, }; @@ -147,7 +155,7 @@ export class KosongLLM implements LLM { streamTiming: firstChunkAt === undefined ? undefined - : buildStreamTiming(requestStartedAt, firstChunkAt, streamEndedAt), + : buildStreamTiming(requestStartedAt, requestSentAt, firstChunkAt, streamEndedAt, decodeStats), }; return response; @@ -160,14 +168,34 @@ export class KosongLLM implements LLM { function buildStreamTiming( requestStartedAt: number, + requestSentAt: number | undefined, firstChunkAt: number, streamEndedAt: number | undefined, + decodeStats: StreamDecodeStats | undefined, ): LLMStreamTiming { const outputEndedAt = streamEndedAt ?? Date.now(); - return { - firstTokenLatencyMs: Math.max(0, firstChunkAt - requestStartedAt), + const firstTokenLatencyMs = Math.max(0, firstChunkAt - requestStartedAt); + const timing: { + -readonly [K in keyof LLMStreamTiming]: LLMStreamTiming[K]; + } = { + firstTokenLatencyMs, streamDurationMs: Math.max(0, outputEndedAt - firstChunkAt), }; + // Split TTFT across the request-dispatch boundary when the provider reported + // it. Clamp `requestSentAt` into [requestStartedAt, firstChunkAt] so a stray + // clock reading can never produce a negative or over-long component. + if (requestSentAt !== undefined) { + const sentAt = Math.min(Math.max(requestSentAt, requestStartedAt), firstChunkAt); + timing.requestBuildMs = sentAt - requestStartedAt; + timing.serverFirstTokenMs = firstChunkAt - sentAt; + } + // Split the decode window into server (awaiting parts) vs. client (processing + // parts) time, as accounted by the stream loop. + if (decodeStats !== undefined) { + timing.serverDecodeMs = Math.max(0, decodeStats.serverDecodeMs); + timing.clientConsumeMs = Math.max(0, decodeStats.clientConsumeMs); + } + return timing; } function buildKosongCallbacks( diff --git a/packages/agent-core/src/loop/events.ts b/packages/agent-core/src/loop/events.ts index 5926496585..786afb574c 100644 --- a/packages/agent-core/src/loop/events.ts +++ b/packages/agent-core/src/loop/events.ts @@ -21,6 +21,20 @@ export interface LoopStepEndEvent { readonly finishReason?: LoopStepStopReason | undefined; readonly llmFirstTokenLatencyMs?: number | undefined; readonly llmStreamDurationMs?: number | undefined; + /** + * Split of `llmFirstTokenLatencyMs`: in-process request-building time on the + * client vs. network + API-server time to the first token. Both `undefined` + * when the provider does not report the client/server boundary. + */ + readonly llmRequestBuildMs?: number | undefined; + readonly llmServerFirstTokenMs?: number | undefined; + /** + * Split of `llmStreamDurationMs` (the decode window): time awaiting parts + * from the provider vs. time processing parts in-process. Both `undefined` + * when the provider stream did not report decode accounting. + */ + readonly llmServerDecodeMs?: number | undefined; + readonly llmClientConsumeMs?: number | undefined; /** * Provider diagnostics are optional and must not drive loop control. * Use `finishReason` for normalized behavior. diff --git a/packages/agent-core/src/loop/llm.ts b/packages/agent-core/src/loop/llm.ts index 1749796df5..0650306231 100644 --- a/packages/agent-core/src/loop/llm.ts +++ b/packages/agent-core/src/loop/llm.ts @@ -31,6 +31,27 @@ export interface LLMRequestLogFields { export interface LLMStreamTiming { readonly firstTokenLatencyMs: number; readonly streamDurationMs: number; + /** + * Portion of `firstTokenLatencyMs` spent in-process building the request + * (message serialization, param assembly) before the provider dispatched the + * network call. `undefined` when the provider does not report the + * client/server boundary (no `onRequestSent`). + */ + readonly requestBuildMs?: number; + /** + * Portion of `firstTokenLatencyMs` spent waiting on the network + API server + * from request dispatch to the first streamed token. `undefined` when the + * provider does not report the client/server boundary. + */ + readonly serverFirstTokenMs?: number; + /** + * Split of `streamDurationMs` (the decode window): time spent awaiting parts + * from the provider (`serverDecodeMs`, server + network) vs. time spent + * processing parts in-process (`clientConsumeMs`, host callbacks / merge). + * `undefined` when the provider stream did not report decode accounting. + */ + readonly serverDecodeMs?: number; + readonly clientConsumeMs?: number; } export interface LLMChatParams { diff --git a/packages/agent-core/src/loop/turn-step.ts b/packages/agent-core/src/loop/turn-step.ts index b06cd67df4..2a0fc61328 100644 --- a/packages/agent-core/src/loop/turn-step.ts +++ b/packages/agent-core/src/loop/turn-step.ts @@ -149,9 +149,15 @@ export async function executeLoopStep(deps: ExecuteLoopStepDeps): Promise<{ finishReason: effectiveStopReason, llmFirstTokenLatencyMs: response.streamTiming?.firstTokenLatencyMs, llmStreamDurationMs: response.streamTiming?.streamDurationMs, + llmRequestBuildMs: response.streamTiming?.requestBuildMs, + llmServerFirstTokenMs: response.streamTiming?.serverFirstTokenMs, + llmServerDecodeMs: response.streamTiming?.serverDecodeMs, + llmClientConsumeMs: response.streamTiming?.clientConsumeMs, ...stepEndProviderDiagnostics(response, effectiveStopReason), }); + logStepTiming(log, turnId, currentStep, response); + let stopTurnAfterStep = stopTurnAfterUsage; if (hooks?.afterStep !== undefined) { try { @@ -176,6 +182,36 @@ export async function executeLoopStep(deps: ExecuteLoopStepDeps): Promise<{ }; } +/** + * Emit a per-step completion log with the LLM response timing. TTFT is split + * into the client-side request-build portion and the network + API-server + * portion, and the decode window is split into server (awaiting parts) vs. + * client (processing parts) time, so slow turns can be attributed without + * parsing the wire log. + */ +function logStepTiming( + log: Logger | undefined, + turnId: string, + step: number, + response: LLMChatResponse, +): void { + if (log === undefined) return; + const timing = response.streamTiming; + if (timing === undefined) return; + log.info('llm response', { + turnStep: `${turnId}/${String(step)}`, + ttftMs: timing.firstTokenLatencyMs, + ...(timing.requestBuildMs !== undefined ? { requestBuildMs: timing.requestBuildMs } : {}), + ...(timing.serverFirstTokenMs !== undefined + ? { serverFirstTokenMs: timing.serverFirstTokenMs } + : {}), + streamDurationMs: timing.streamDurationMs, + ...(timing.serverDecodeMs !== undefined ? { serverDecodeMs: timing.serverDecodeMs } : {}), + ...(timing.clientConsumeMs !== undefined ? { clientConsumeMs: timing.clientConsumeMs } : {}), + outputTokens: response.usage.output, + }); +} + function deriveStepStopReason(response: LLMChatResponse): LoopStepStopReason { switch (response.providerFinishReason) { case 'truncated': diff --git a/packages/agent-core/test/agent/harness/snapshots.ts b/packages/agent-core/test/agent/harness/snapshots.ts index dea21e5bc8..a027e58c50 100644 --- a/packages/agent-core/test/agent/harness/snapshots.ts +++ b/packages/agent-core/test/agent/harness/snapshots.ts @@ -314,7 +314,15 @@ function isUuid(value: string): boolean { } function isVolatileDurationKey(key: string): boolean { - return key === 'llmFirstTokenLatencyMs' || key === 'llmStreamDurationMs' || key === 'durationMs'; + return ( + key === 'llmFirstTokenLatencyMs' || + key === 'llmStreamDurationMs' || + key === 'llmRequestBuildMs' || + key === 'llmServerFirstTokenMs' || + key === 'llmServerDecodeMs' || + key === 'llmClientConsumeMs' || + key === 'durationMs' + ); } function isPlanModeReminder(value: string): boolean { diff --git a/packages/agent-core/test/agent/kosong-llm.test.ts b/packages/agent-core/test/agent/kosong-llm.test.ts index 4855f12366..ffb80edb32 100644 --- a/packages/agent-core/test/agent/kosong-llm.test.ts +++ b/packages/agent-core/test/agent/kosong-llm.test.ts @@ -134,6 +134,143 @@ describe('KosongLLM stream timing', () => { expect(response.streamTiming?.firstTokenLatencyMs).toBeGreaterThanOrEqual(0); expect(response.streamTiming?.streamDurationMs).toBeGreaterThanOrEqual(0); }); + + it('splits first-token latency across the request-dispatch boundary', async () => { + const generate: GenerateFn = async ( + _provider, + _systemPrompt, + _tools, + _history, + callbacks, + options, + ) => { + options?.onRequestStart?.(); + options?.onRequestSent?.(); + await callbacks?.onMessagePart?.({ type: 'text', text: 'timed' }); + options?.onStreamEnd?.(); + return { + id: 'response-1', + message: { role: 'assistant', content: [{ type: 'text', text: 'timed' }], toolCalls: [] }, + usage: emptyUsage(), + finishReason: 'completed', + rawFinishReason: 'stop', + }; + }; + const llm = new KosongLLM({ provider, systemPrompt: 'system', generate }); + + const response = await llm.chat({ + messages: [], + tools: [], + signal: new AbortController().signal, + }); + + const timing = response.streamTiming; + expect(timing?.requestBuildMs).toBeGreaterThanOrEqual(0); + expect(timing?.serverFirstTokenMs).toBeGreaterThanOrEqual(0); + // The two components reconstruct the total (allowing for clock granularity). + expect((timing?.requestBuildMs ?? 0) + (timing?.serverFirstTokenMs ?? 0)).toBe( + timing?.firstTokenLatencyMs, + ); + }); + + it('leaves the split undefined when the provider does not report dispatch', async () => { + const generate: GenerateFn = async ( + _provider, + _systemPrompt, + _tools, + _history, + callbacks, + options, + ) => { + options?.onRequestStart?.(); + // No onRequestSent — older providers / stubs that do not mark dispatch. + await callbacks?.onMessagePart?.({ type: 'text', text: 'timed' }); + options?.onStreamEnd?.(); + return { + id: 'response-1', + message: { role: 'assistant', content: [{ type: 'text', text: 'timed' }], toolCalls: [] }, + usage: emptyUsage(), + finishReason: 'completed', + rawFinishReason: 'stop', + }; + }; + const llm = new KosongLLM({ provider, systemPrompt: 'system', generate }); + + const response = await llm.chat({ + messages: [], + tools: [], + signal: new AbortController().signal, + }); + + expect(response.streamTiming?.firstTokenLatencyMs).toBeGreaterThanOrEqual(0); + expect(response.streamTiming?.requestBuildMs).toBeUndefined(); + expect(response.streamTiming?.serverFirstTokenMs).toBeUndefined(); + }); + + it('surfaces the decode wait/consume split reported by the stream', async () => { + const generate: GenerateFn = async ( + _provider, + _systemPrompt, + _tools, + _history, + callbacks, + options, + ) => { + options?.onRequestStart?.(); + options?.onRequestSent?.(); + await callbacks?.onMessagePart?.({ type: 'text', text: 'timed' }); + options?.onStreamEnd?.({ serverDecodeMs: 800, clientConsumeMs: 200 }); + return { + id: 'response-1', + message: { role: 'assistant', content: [{ type: 'text', text: 'timed' }], toolCalls: [] }, + usage: emptyUsage(), + finishReason: 'completed', + rawFinishReason: 'stop', + }; + }; + const llm = new KosongLLM({ provider, systemPrompt: 'system', generate }); + + const response = await llm.chat({ + messages: [], + tools: [], + signal: new AbortController().signal, + }); + + expect(response.streamTiming?.serverDecodeMs).toBe(800); + expect(response.streamTiming?.clientConsumeMs).toBe(200); + }); + + it('leaves the decode split undefined when the stream reports no accounting', async () => { + const generate: GenerateFn = async ( + _provider, + _systemPrompt, + _tools, + _history, + callbacks, + options, + ) => { + options?.onRequestStart?.(); + await callbacks?.onMessagePart?.({ type: 'text', text: 'timed' }); + options?.onStreamEnd?.(); // no decode stats + return { + id: 'response-1', + message: { role: 'assistant', content: [{ type: 'text', text: 'timed' }], toolCalls: [] }, + usage: emptyUsage(), + finishReason: 'completed', + rawFinishReason: 'stop', + }; + }; + const llm = new KosongLLM({ provider, systemPrompt: 'system', generate }); + + const response = await llm.chat({ + messages: [], + tools: [], + signal: new AbortController().signal, + }); + + expect(response.streamTiming?.serverDecodeMs).toBeUndefined(); + expect(response.streamTiming?.clientConsumeMs).toBeUndefined(); + }); }); describe('KosongLLM completion budget', () => { diff --git a/packages/kosong/src/generate.ts b/packages/kosong/src/generate.ts index d17fef39c5..ad475ed527 100644 --- a/packages/kosong/src/generate.ts +++ b/packages/kosong/src/generate.ts @@ -111,53 +111,83 @@ export async function generate( // the stream. await throwIfAborted(options?.signal, stream); + // Decode-phase accounting. We split the window from the first streamed part + // to stream end into time spent awaiting the next part (server + network) vs. + // time spent processing each part in-process (deep copy, host callback, part + // merge). `lastResumeAt` marks the end of the previous part's processing, so + // the gap until the next part arrives is attributed to the server. The + // per-part processing is wrapped in try/finally so the accounting stays + // correct across `continue` and thrown aborts. + let serverDecodeMs = 0; + let clientConsumeMs = 0; + let firstPartAt: number | undefined; + let lastResumeAt = 0; + for await (const part of stream) { - await throwIfAborted(options?.signal, stream); + const arrivedAt = Date.now(); + if (firstPartAt === undefined) { + firstPartAt = arrivedAt; + } else { + serverDecodeMs += arrivedAt - lastResumeAt; + } - // Notify raw part callback (deep copy to avoid aliasing mutations). - if (callbacks?.onMessagePart !== undefined) { - await callbacks.onMessagePart(deepCopyPart(part)); + try { await throwIfAborted(options?.signal, stream); - } - // Index-based routing for parallel tool call argument deltas. - // When a ToolCallPart arrives with an index referring to a tool call - // that is NOT the currently-pending one, append it directly to the - // correct ToolCall in message.toolCalls instead of relying on sequential - // merging. This prevents argument cross-contamination across parallel calls. - if ( - isToolCallPart(part) && - part.index !== undefined && - !isPendingToolCallAtIndex(pendingPart, part.index) - ) { - const arrayIdx = toolCallIndexMap.get(part.index); - if (arrayIdx !== undefined) { - const target = message.toolCalls[arrayIdx]; - if (target !== undefined && part.argumentsPart !== null) { - target.arguments = - target.arguments === null - ? part.argumentsPart - : target.arguments + part.argumentsPart; + // Notify raw part callback (deep copy to avoid aliasing mutations). + if (callbacks?.onMessagePart !== undefined) { + await callbacks.onMessagePart(deepCopyPart(part)); + await throwIfAborted(options?.signal, stream); + } + + // Index-based routing for parallel tool call argument deltas. + // When a ToolCallPart arrives with an index referring to a tool call + // that is NOT the currently-pending one, append it directly to the + // correct ToolCall in message.toolCalls instead of relying on sequential + // merging. This prevents argument cross-contamination across parallel calls. + if ( + isToolCallPart(part) && + part.index !== undefined && + !isPendingToolCallAtIndex(pendingPart, part.index) + ) { + const arrayIdx = toolCallIndexMap.get(part.index); + if (arrayIdx !== undefined) { + const target = message.toolCalls[arrayIdx]; + if (target !== undefined && part.argumentsPart !== null) { + target.arguments = + target.arguments === null + ? part.argumentsPart + : target.arguments + part.argumentsPart; + } + continue; } - continue; + // Unknown index — fall through to the sequential logic as a safety net. } - // Unknown index — fall through to the sequential logic as a safety net. - } - if (pendingPart === null) { - pendingPart = part; - } else if (!mergeInPlace(pendingPart, part)) { - // Could not merge — flush the pending part and start a new one. - // For parallel tool calls this happens when a new ToolCall header arrives - // while a previous ToolCall is still pending; the flush finalizes the - // previous tool call into `message.toolCalls`. - flushPart(message, pendingPart, toolCallIndexMap); - pendingPart = part; + if (pendingPart === null) { + pendingPart = part; + } else if (!mergeInPlace(pendingPart, part)) { + // Could not merge — flush the pending part and start a new one. + // For parallel tool calls this happens when a new ToolCall header arrives + // while a previous ToolCall is still pending; the flush finalizes the + // previous tool call into `message.toolCalls`. + flushPart(message, pendingPart, toolCallIndexMap); + pendingPart = part; + } + } finally { + lastResumeAt = Date.now(); + clientConsumeMs += lastResumeAt - arrivedAt; } } await throwIfAborted(options?.signal, stream); - options?.onStreamEnd?.(); + if (firstPartAt !== undefined) { + // Tail wait: from the last processed part to the stream's done signal. + serverDecodeMs += Date.now() - lastResumeAt; + } + options?.onStreamEnd?.( + firstPartAt === undefined ? undefined : { serverDecodeMs, clientConsumeMs }, + ); // Flush the last pending part. if (pendingPart !== null) { diff --git a/packages/kosong/src/provider.ts b/packages/kosong/src/provider.ts index 1782f03e45..2e08f6c24d 100644 --- a/packages/kosong/src/provider.ts +++ b/packages/kosong/src/provider.ts @@ -118,11 +118,41 @@ export interface GenerateOptions { * provider adapter's generate call. */ onRequestStart?: () => void; + /** + * Host-side instrumentation hook fired by the provider adapter immediately + * before it dispatches the network request to the upstream API. The window + * between {@link onRequestStart} and this hook is in-process request-building + * time (message serialization, param assembly) spent by the client; the + * window between this hook and the first streamed part is network + server + * time. Splitting time-to-first-token across this boundary lets hosts + * attribute latency to the client vs. the API server. + */ + onRequestSent?: () => void; /** * Host-side instrumentation hook fired after the provider stream is fully - * drained, before post-processing the assembled response. + * drained, before post-processing the assembled response. Receives the + * {@link StreamDecodeStats} accounting accumulated across the stream when at + * least one part was streamed, or `undefined` for an empty stream. */ - onStreamEnd?: () => void; + onStreamEnd?: (stats?: StreamDecodeStats) => void; +} + +/** + * Decode-phase accounting for a single streamed generation. Splits the window + * from the first streamed part to stream end into the time spent waiting on the + * provider for the next part (server + network) versus the time spent + * processing each part in-process (deep copy, host callbacks, part merging). + * + * Because both buckets are wall-clock measured on the single JS thread, a + * stop-the-world GC pause that lands while awaiting the next part is counted in + * {@link serverDecodeMs}; a non-trivial {@link clientConsumeMs} share is the + * unambiguous signal that the host's per-part processing is throttling decode. + */ +export interface StreamDecodeStats { + /** Cumulative time spent awaiting the next streamed part (server + network). */ + readonly serverDecodeMs: number; + /** Cumulative time spent processing streamed parts in-process (client). */ + readonly clientConsumeMs: number; } /** diff --git a/packages/kosong/src/providers/anthropic.ts b/packages/kosong/src/providers/anthropic.ts index 1b43abdda6..9a615ed882 100644 --- a/packages/kosong/src/providers/anthropic.ts +++ b/packages/kosong/src/providers/anthropic.ts @@ -1102,6 +1102,7 @@ export class AnthropicChatProvider implements ChatProvider { } const finalRequestOptions = Object.keys(requestOptions).length > 0 ? requestOptions : undefined; const client = this._createClient(options?.auth); + options?.onRequestSent?.(); if (this._stream) { // Use the raw Messages stream instead of the SDK MessageStream helper. diff --git a/packages/kosong/src/providers/google-genai.ts b/packages/kosong/src/providers/google-genai.ts index 290645fd97..522c48a52f 100644 --- a/packages/kosong/src/providers/google-genai.ts +++ b/packages/kosong/src/providers/google-genai.ts @@ -790,6 +790,7 @@ export class GoogleGenAIChatProvider implements ChatProvider { // the initial SDK request against the caller's abort signal ourselves. // Once we have a response/stream object, the wrapper below continues to // check the signal at each chunk boundary. + options?.onRequestSent?.(); if (this._stream) { const stream = await Promise.race([ models.generateContentStream(params), diff --git a/packages/kosong/src/providers/kimi.ts b/packages/kosong/src/providers/kimi.ts index 3a120c0782..e7c25c3846 100644 --- a/packages/kosong/src/providers/kimi.ts +++ b/packages/kosong/src/providers/kimi.ts @@ -488,6 +488,7 @@ export class KimiChatProvider implements ChatProvider { const client = this._createClient(options?.auth); // Use type assertion via unknown because we pass Moonshot-proprietary fields // (reasoning_effort, thinking) that don't exist in the OpenAI type definitions. + options?.onRequestSent?.(); const response = (await client.chat.completions.create( createParams as unknown as OpenAI.Chat.ChatCompletionCreateParamsNonStreaming, options?.signal ? { signal: options.signal } : undefined, diff --git a/packages/kosong/src/providers/openai-legacy.ts b/packages/kosong/src/providers/openai-legacy.ts index 6caa55e24c..50187fd7ac 100644 --- a/packages/kosong/src/providers/openai-legacy.ts +++ b/packages/kosong/src/providers/openai-legacy.ts @@ -571,6 +571,7 @@ export class OpenAILegacyChatProvider implements ChatProvider { try { const client = this._createClient(options?.auth); + options?.onRequestSent?.(); const response = (await client.chat.completions.create( createParams as unknown as OpenAI.Chat.ChatCompletionCreateParamsNonStreaming, options?.signal ? { signal: options.signal } : undefined, diff --git a/packages/kosong/src/providers/openai-responses.ts b/packages/kosong/src/providers/openai-responses.ts index bb010538dd..48544b81ba 100644 --- a/packages/kosong/src/providers/openai-responses.ts +++ b/packages/kosong/src/providers/openai-responses.ts @@ -1084,6 +1084,7 @@ export class OpenAIResponsesChatProvider implements ChatProvider { ); } + options?.onRequestSent?.(); const response = await ( client.responses as { create(params: unknown, opts?: unknown): Promise; diff --git a/packages/kosong/test/generate.test.ts b/packages/kosong/test/generate.test.ts index 6c27779471..7b3266c211 100644 --- a/packages/kosong/test/generate.test.ts +++ b/packages/kosong/test/generate.test.ts @@ -932,4 +932,84 @@ describe('generate()', () => { expect(result.rawFinishReason).toBe('content_filter'); }); }); + + describe('decode accounting', () => { + const sleep = (ms: number): Promise => new Promise((r) => setTimeout(r, ms)); + + function createDelayedStream( + parts: StreamedMessagePart[], + perPartWaitMs: number, + ): StreamedMessage { + return { + get id(): string | null { + return null; + }, + get usage(): TokenUsage | null { + return null; + }, + finishReason: 'completed', + rawFinishReason: 'stop', + async *[Symbol.asyncIterator](): AsyncIterator { + let first = true; + for (const part of parts) { + // Simulate the provider taking time to produce each part after the + // first (the first part's wait is time-to-first-token, not decode). + if (!first && perPartWaitMs > 0) await sleep(perPartWaitMs); + first = false; + yield part; + } + }, + }; + } + + it('attributes per-part processing time to the client bucket', async () => { + const stream = createDelayedStream( + [ + { type: 'text', text: 'a' }, + { type: 'text', text: 'b' }, + { type: 'text', text: 'c' }, + ], + 0, // provider yields instantly — all measurable time is client-side + ); + const provider = createMockProvider(stream); + let stats: { serverDecodeMs: number; clientConsumeMs: number } | undefined; + await generate(provider, '', [], [], { + async onMessagePart(): Promise { + await sleep(25); + }, + }, { + onStreamEnd: (s) => { + stats = s; + }, + }); + expect(stats).toBeDefined(); + expect(stats!.clientConsumeMs).toBeGreaterThan(stats!.serverDecodeMs); + expect(stats!.clientConsumeMs).toBeGreaterThanOrEqual(50); + }); + + it('attributes time spent awaiting parts to the server bucket', async () => { + const stream = createDelayedStream( + [ + { type: 'text', text: 'a' }, + { type: 'text', text: 'b' }, + { type: 'text', text: 'c' }, + ], + 25, // provider stalls before each part after the first + ); + const provider = createMockProvider(stream); + let stats: { serverDecodeMs: number; clientConsumeMs: number } | undefined; + await generate(provider, '', [], [], { + onMessagePart(): void { + // instant client processing + }, + }, { + onStreamEnd: (s) => { + stats = s; + }, + }); + expect(stats).toBeDefined(); + expect(stats!.serverDecodeMs).toBeGreaterThan(stats!.clientConsumeMs); + expect(stats!.serverDecodeMs).toBeGreaterThanOrEqual(40); + }); + }); }); diff --git a/packages/node-sdk/test/session-event-types.test.ts b/packages/node-sdk/test/session-event-types.test.ts index e57bda3881..19786a598d 100644 --- a/packages/node-sdk/test/session-event-types.test.ts +++ b/packages/node-sdk/test/session-event-types.test.ts @@ -28,6 +28,18 @@ describe('Event public types', () => { expectTypeOf['llmStreamDurationMs']>().toEqualTypeOf< number | undefined >(); + expectTypeOf['llmRequestBuildMs']>().toEqualTypeOf< + number | undefined + >(); + expectTypeOf['llmServerFirstTokenMs']>().toEqualTypeOf< + number | undefined + >(); + expectTypeOf['llmServerDecodeMs']>().toEqualTypeOf< + number | undefined + >(); + expectTypeOf['llmClientConsumeMs']>().toEqualTypeOf< + number | undefined + >(); }); it('narrows subagent lifecycle events by type', () => { diff --git a/packages/protocol/src/events.ts b/packages/protocol/src/events.ts index 7bd9be73d9..9347da5434 100644 --- a/packages/protocol/src/events.ts +++ b/packages/protocol/src/events.ts @@ -430,6 +430,20 @@ export interface TurnStepCompletedEvent { readonly finishReason?: string; readonly llmFirstTokenLatencyMs?: number; readonly llmStreamDurationMs?: number; + /** + * Split of `llmFirstTokenLatencyMs`: in-process request-building time on the + * client vs. network + API-server time to the first token. Both omitted when + * the provider does not report the client/server boundary. + */ + readonly llmRequestBuildMs?: number; + readonly llmServerFirstTokenMs?: number; + /** + * Split of `llmStreamDurationMs` (the decode window): time awaiting parts from + * the provider vs. time processing parts in-process. Both omitted when the + * provider stream did not report decode accounting. + */ + readonly llmServerDecodeMs?: number; + readonly llmClientConsumeMs?: number; readonly providerFinishReason?: FinishReason; readonly rawFinishReason?: string; } @@ -1096,6 +1110,10 @@ export const turnStepCompletedEventSchema = z.object({ finishReason: z.string().optional(), llmFirstTokenLatencyMs: z.number().optional(), llmStreamDurationMs: z.number().optional(), + llmRequestBuildMs: z.number().optional(), + llmServerFirstTokenMs: z.number().optional(), + llmServerDecodeMs: z.number().optional(), + llmClientConsumeMs: z.number().optional(), providerFinishReason: finishReasonSchema.optional(), rawFinishReason: z.string().optional(), }) satisfies z.ZodType;