diff --git a/packages/core/src/agents/codex/cli-middleware-runtime.ts b/packages/core/src/agents/codex/cli-middleware-runtime.ts index c78ab872..ae7aa70c 100644 --- a/packages/core/src/agents/codex/cli-middleware-runtime.ts +++ b/packages/core/src/agents/codex/cli-middleware-runtime.ts @@ -4819,6 +4819,9 @@ function configRequirementsRead(existingResult) { ? { ...requirements.featureRequirements } : {}; featureRequirements.fast_mode = true; + if (!Object.prototype.hasOwnProperty.call(requirements, "application")) { + requirements.application = null; + } requirements.featureRequirements = featureRequirements; result.requirements = requirements; return result; diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index 30b01af5..919a19fd 100644 --- a/packages/core/src/config/config.ts +++ b/packages/core/src/config/config.ts @@ -875,6 +875,10 @@ function pickConfig(value: Partial): LoadedAppConfig { if (trayIcon) { config.trayIcon = trayIcon; } + const trayShowTokenRate = (value as Record).trayShowTokenRate; + if (typeof trayShowTokenRate === "boolean") { + config.trayShowTokenRate = trayShowTokenRate; + } const trayBalanceProgress = parseTrayBalanceProgress((value as Record).trayBalanceProgress); if (trayBalanceProgress) { config.trayBalanceProgress = trayBalanceProgress; diff --git a/packages/core/src/config/default-config.ts b/packages/core/src/config/default-config.ts index 20da9644..ac03d389 100644 --- a/packages/core/src/config/default-config.ts +++ b/packages/core/src/config/default-config.ts @@ -216,6 +216,7 @@ export function createDefaultAppConfig(options: DefaultAppConfigOptions = {}): A theme: "system", trayComponentVariants: DEFAULT_TRAY_COMPONENT_VARIANTS, trayIcon: "random", + trayShowTokenRate: false, trayProgressTargetTokens: 100000, trayWidgets: DEFAULT_TRAY_WIDGETS, trayWindowModules: DEFAULT_TRAY_WINDOW_MODULES, diff --git a/packages/core/src/contracts/app.ts b/packages/core/src/contracts/app.ts index 29ec35db..a99cfa7a 100644 --- a/packages/core/src/contracts/app.ts +++ b/packages/core/src/contracts/app.ts @@ -1854,6 +1854,7 @@ export type AppConfig = { routerEndpoint: string; theme: "system" | "light" | "dark"; trayBalanceProgress?: TrayBalanceProgressConfig; + trayShowTokenRate: boolean; trayProgressTargetTokens: number; trayComponentVariants: TrayComponentVariants; trayIcon: TrayIconPreference; @@ -2190,7 +2191,33 @@ export type RequestRouteTrace = { version: 1 | 2; }; +export type StreamSpeedSampleStatus = + | "complete" + | "partial" + | "usage_missing" + | "insufficient_tokens" + | "unsupported_protocol" + | "hidden_reasoning" + | "batched_output"; + +export type RequestStreamMetrics = { + activeOutputMs?: number; + estimatedOutputTokens: number; + maxInterEventGapMs?: number; + p95InterEventGapMs?: number; + reasoningObserved: boolean; + responseHeadersMs?: number; + sampleStatus: StreamSpeedSampleStatus; + tailMs?: number; + textObserved: boolean; + timeToFirstSignalMs?: number; + timeToFirstTextMs?: number; + toolObserved: boolean; + upstreamTimeToFirstSignalMs?: number; +}; + export type RequestLogEntry = { + activeOutputMs?: number; cacheReadTokens: number; cacheWriteTokens: number; client: string; @@ -2205,13 +2232,17 @@ export type RequestLogEntry = { id: number; inputTokens: number; isStream: boolean; + maxInterEventGapMs?: number; method: string; model: string; ok: boolean; outputTokens: number; + outputTokensPerSecond?: number; path: string; + p95InterEventGapMs?: number; provider: string; reasoningTokens: number; + responseHeadersMs?: number; requestedModel?: string; requestBody: RequestLogBody; requestHeaders: Record; @@ -2226,7 +2257,12 @@ export type RequestLogEntry = { responseModel?: string; responseHeaders: Record; statusCode: number; + streamSpeedSampleStatus?: StreamSpeedSampleStatus; + tailMs?: number; + timeToFirstSignalMs?: number; + timeToFirstTextMs?: number; totalTokens: number; + upstreamTimeToFirstSignalMs?: number; url: string; }; diff --git a/packages/core/src/gateway/application/gateway-service.ts b/packages/core/src/gateway/application/gateway-service.ts index d0b8ce67..014a51b4 100644 --- a/packages/core/src/gateway/application/gateway-service.ts +++ b/packages/core/src/gateway/application/gateway-service.ts @@ -20,7 +20,11 @@ import { GatewayBillingSynchronizer } from "@ccr/core/usage/billing-sync"; import { assertLoopbackCoreHost, endpoint, formatCoreGatewayChildExit, gatewayNetworkEndpoints, gatewayRuntimeSupportsRouterPlugin, generateCoreGatewayAuthToken, isCoreGatewayHealthy, loopbackCoreHostError, removeManagedCoreGatewayMarker, shouldRunGatewayRuntime, shouldRunUnifiedServer, spawnGatewayProcess, stopPreviousManagedCoreGateway, waitForCoreGatewayStop, waitForManagedCoreGatewayReady, writeManagedCoreGatewayMarker } from "@ccr/core/gateway/core-runtime/supervisor"; import { coreGatewayAuthHeader } from "@ccr/core/gateway/internal/shared"; import type { BrowserAutomationMcpIntegration, BrowserWebSearchMcpIntegration, GatewayStopOptions } from "@ccr/core/gateway/internal/shared"; -import { ccrRuntimeConfigReloadMessageType } from "@ccr/core/gateway/core-runtime/router-plugin-contract"; +import { + ccrLiveTokenRateConfigMessageType, + ccrLiveTokenRateSnapshotMessageType, + ccrRuntimeConfigReloadMessageType +} from "@ccr/core/gateway/core-runtime/router-plugin-contract"; import { GatewayRequestPipeline } from "@ccr/core/gateway/request/pipeline"; import { GatewayHttpRequestHandler } from "@ccr/core/gateway/http/request-handler"; import { gatewayRuntimeConfigRevision } from "@ccr/core/gateway/runtime-config-control"; @@ -34,9 +38,12 @@ import { mediaToolsGatewayEndpoint } from "@ccr/core/mcp/grok-media-config"; import { browserAutomationMcpEnabled } from "@ccr/core/mcp/toolhub-config"; import { installSocketTypeOfServiceCompat } from "@ccr/core/platform/socket-compat"; import { profileApiKeyId } from "@ccr/core/profiles/api-key"; +import { setExternalLiveTokenRateSnapshot } from "@ccr/core/observability/stream-experience"; installSocketTypeOfServiceCompat(); +const coreGatewayLiveTokenRateSourceId = "core-gateway"; + type RouteScriptTestHeaders = Record; type SingleGatewayRuntimeBlockerOptions = { gatewayRuntimeSupportsRouterPlugin?: () => boolean; @@ -343,6 +350,7 @@ class GatewayService { const child = this.child; const childCoreEndpoint = child ? this.status.coreEndpoint : ""; this.child = undefined; + setExternalLiveTokenRateSnapshot(coreGatewayLiveTokenRateSourceId); this.coreAuthToken = ""; this.externalGatewayApiKey = undefined; if (child && !child.killed) { @@ -427,6 +435,7 @@ class GatewayService { gatewayManagedExternally: undefined, networkEndpoints: gatewayNetworkEndpoints(config.gateway.host, config.gateway.port) }; + this.sendCoreGatewayLiveTokenRateConfig(config.trayShowTokenRate); } validateRouteScript(request: RouteScriptValidationRequest): Promise { @@ -525,6 +534,7 @@ class GatewayService { return; } this.child = undefined; + setExternalLiveTokenRateSnapshot(coreGatewayLiveTokenRateSourceId); this.coreAuthToken = ""; await removeManagedCoreGatewayMarkerBestEffort(); this.status = { @@ -537,10 +547,34 @@ class GatewayService { } private handleCoreGatewayMessage(child: ChildProcess, message: unknown): void { - if (this.child !== child || !isRuntimeConfigReloadMessage(message)) { + if (this.child !== child) { return; } - this.schedulePersistedRuntimeConfigReload(message.configRevision, message.forceRestart === true); + if (isLiveTokenRateSnapshotMessage(message)) { + setExternalLiveTokenRateSnapshot(coreGatewayLiveTokenRateSourceId, { + activeRequests: message.activeRequests, + tokensPerSecond: message.tokensPerSecond + }); + return; + } + if (isRuntimeConfigReloadMessage(message)) { + this.schedulePersistedRuntimeConfigReload(message.configRevision, message.forceRestart === true); + } + } + + private sendCoreGatewayLiveTokenRateConfig(enabled: boolean): void { + if (!this.child?.connected || !this.child.send) { + return; + } + try { + this.child.send({ + enabled, + protocolVersion: 1, + type: ccrLiveTokenRateConfigMessageType + }); + } catch { + // The runtime may exit between the connected check and send(). + } } private async handleRequest(request: IncomingMessage, response: ServerResponse): Promise { @@ -609,6 +643,7 @@ class GatewayService { this.child = undefined; this.server = undefined; this.coreAuthToken = ""; + setExternalLiveTokenRateSnapshot(coreGatewayLiveTokenRateSourceId); this.externalGatewayApiKey = apiKey; this.status = { coreEndpoint: endpoint(config.gateway.coreHost, config.gateway.corePort), @@ -713,6 +748,19 @@ function isRuntimeConfigReloadMessage(message: unknown): message is { configRevi (record.forceRestart === undefined || typeof record.forceRestart === "boolean"); } +function isLiveTokenRateSnapshotMessage( + message: unknown +): message is { activeRequests: number; tokensPerSecond: number } { + if (!message || typeof message !== "object" || Array.isArray(message)) { + return false; + } + const record = message as Record; + return record.type === ccrLiveTokenRateSnapshotMessageType && + record.protocolVersion === 1 && + typeof record.activeRequests === "number" && Number.isFinite(record.activeRequests) && + typeof record.tokensPerSecond === "number" && Number.isFinite(record.tokensPerSecond); +} + function withCoreRuntimeEndpoint(config: AppConfig, host: string, port: number): AppConfig { return { ...config, diff --git a/packages/core/src/gateway/core-runtime/router-plugin-contract.ts b/packages/core/src/gateway/core-runtime/router-plugin-contract.ts index 9861efe9..40421b6c 100644 --- a/packages/core/src/gateway/core-runtime/router-plugin-contract.ts +++ b/packages/core/src/gateway/core-runtime/router-plugin-contract.ts @@ -12,6 +12,22 @@ export const ccrRouterHttpRouteKey = "ccr-router-route"; export const ccrRouterHttpRoutePath = "/__ccr/route"; export const ccrRawTraceSyncAckRouteKey = "ccr-raw-trace-sync-ack"; export const ccrRuntimeConfigReloadMessageType = "ccr:runtime-config-reload"; +export const ccrLiveTokenRateConfigMessageType = "ccr:live-token-rate-config"; +export const ccrLiveTokenRateSnapshotMessageType = "ccr:live-token-rate-snapshot"; +export const ccrLiveTokenRateStreamHookKey = "ccr-live-token-rate-stream-hook"; + +export type CcrLiveTokenRateConfigMessage = { + enabled: boolean; + protocolVersion: 1; + type: typeof ccrLiveTokenRateConfigMessageType; +}; + +export type CcrLiveTokenRateSnapshotMessage = { + activeRequests: number; + protocolVersion: 1; + tokensPerSecond: number; + type: typeof ccrLiveTokenRateSnapshotMessageType; +}; export const ccrRouteStageHeader = "x-ccr-route-stage"; export const ccrRouteReasonHeader = "x-ccr-route-reason"; diff --git a/packages/core/src/gateway/core-runtime/router-plugin.ts b/packages/core/src/gateway/core-runtime/router-plugin.ts index f509c4ef..7be74431 100644 --- a/packages/core/src/gateway/core-runtime/router-plugin.ts +++ b/packages/core/src/gateway/core-runtime/router-plugin.ts @@ -1,4 +1,4 @@ -import { timingSafeEqual } from "node:crypto"; +import { randomUUID, timingSafeEqual } from "node:crypto"; import type { IncomingMessage, ServerResponse } from "node:http"; import { Readable } from "node:stream"; import type { ApiKeyConfig, AppConfig, GatewayProviderConfig, GatewayProviderProtocol, ProfileConfig, RouterRule } from "@ccr/core/contracts/app"; @@ -18,6 +18,9 @@ import { ccrCodexBridgeResponseHookKey, ccrCodexBridgeStreamHookKey, ccrCodexMultiAgentBridgeHeader, + ccrLiveTokenRateConfigMessageType, + ccrLiveTokenRateSnapshotMessageType, + ccrLiveTokenRateStreamHookKey, ccrOpenRouterDiscountFinalizeResponseHookKey, ccrOpenRouterDiscountFinalizeStreamHookKey, ccrOpenRouterDiscountRequestIdHeader, @@ -57,6 +60,7 @@ import { transformCodexMultiAgentBridgeResponseValue } from "@ccr/core/gateway/features/codex-multi-agent-bridge"; import { requestLogRequestedModel } from "@ccr/core/observability/request-log-model"; +import { createStreamExperienceMeter, LiveTokenRateTracker } from "@ccr/core/observability/stream-experience"; import { finalizeOpenRouterDiscountProviderRouterSelection, openRouterDiscountProviderRouterTransform @@ -170,6 +174,8 @@ type GatewayStreamHookInput = { method?: string; url?: string; }; + targetProvider?: string; + targetProviderConfig?: Pick; upstreamRequest?: UpstreamRequest; upstreamResponse: Response; }; @@ -202,6 +208,7 @@ export async function createGatewayPlugin(input: GatewayPluginFactoryInput = {}) scriptValidationErrors }); const openRouterDiscountContext = openRouterDiscountTransformContext(config); + const liveTokenRatePublisher = coreGatewayLiveTokenRatePublisherFor(config.trayShowTokenRate); return { httpRoutes: [{ @@ -457,6 +464,10 @@ export async function createGatewayPlugin(input: GatewayPluginFactoryInput = {}) finalizeOpenRouterDiscountSelection(streamInput); return undefined; } + }, { + key: ccrLiveTokenRateStreamHookKey, + transformResponse: (streamInput: GatewayStreamHookInput) => + applyLiveTokenRateStreamTransform(streamInput, liveTokenRatePublisher) }], routeResolvers: [{ key: ccrRouterRouteResolverKey, @@ -466,6 +477,127 @@ export async function createGatewayPlugin(input: GatewayPluginFactoryInput = {}) }; } +const liveTokenRatePublishIntervalMs = 250; +let sharedCoreGatewayLiveTokenRatePublisher: CoreGatewayLiveTokenRatePublisher | undefined; + +class CoreGatewayLiveTokenRatePublisher { + readonly tracker = new LiveTokenRateTracker(); + private enabled = false; + private lastPublishedAt = 0; + private publishTimer?: NodeJS.Timeout; + + constructor(enabled: boolean) { + process.on("message", this.handleMessage); + this.configure(enabled); + } + + configure(enabled: boolean): void { + const nextEnabled = enabled && typeof process.send === "function"; + if (this.enabled === nextEnabled) { + return; + } + this.enabled = nextEnabled; + if (!nextEnabled) { + this.tracker.clear(); + } + this.publishNow(); + } + + isEnabled(): boolean { + return this.enabled; + } + + notifyActivity(urgent: boolean): void { + if (!this.enabled) { + return; + } + const delayMs = liveTokenRatePublishIntervalMs - (Date.now() - this.lastPublishedAt); + if (urgent || delayMs <= 0) { + this.publishNow(); + return; + } + if (this.publishTimer) { + return; + } + this.publishTimer = setTimeout(() => { + this.publishTimer = undefined; + this.publishNow(); + }, delayMs); + this.publishTimer.unref?.(); + } + + private readonly handleMessage = (message: unknown): void => { + if (!isRecord(message) || message.type !== ccrLiveTokenRateConfigMessageType || message.protocolVersion !== 1) { + return; + } + this.configure(message.enabled === true); + }; + + private publishNow(): void { + if (this.publishTimer) { + clearTimeout(this.publishTimer); + this.publishTimer = undefined; + } + this.lastPublishedAt = Date.now(); + const snapshot = this.enabled + ? this.tracker.snapshot() + : { activeRequests: 0, tokensPerSecond: 0 }; + try { + process.send?.({ + ...snapshot, + protocolVersion: 1, + type: ccrLiveTokenRateSnapshotMessageType + }); + } catch { + this.enabled = false; + this.tracker.clear(); + return; + } + if (this.enabled && snapshot.activeRequests > 0) { + this.publishTimer = setTimeout(() => { + this.publishTimer = undefined; + this.publishNow(); + }, liveTokenRatePublishIntervalMs); + this.publishTimer.unref?.(); + } + } +} + +function coreGatewayLiveTokenRatePublisherFor(enabled: boolean): CoreGatewayLiveTokenRatePublisher { + sharedCoreGatewayLiveTokenRatePublisher ??= new CoreGatewayLiveTokenRatePublisher(enabled); + sharedCoreGatewayLiveTokenRatePublisher.configure(enabled); + return sharedCoreGatewayLiveTokenRatePublisher; +} + +function applyLiveTokenRateStreamTransform( + streamInput: GatewayStreamHookInput, + publisher: CoreGatewayLiveTokenRatePublisher +): Response | undefined { + if (!publisher.isEnabled() || !streamInput.upstreamResponse.body) { + return undefined; + } + const protocol = normalizeProviderProtocol(streamInput.targetProviderConfig?.type) ?? + normalizeProviderProtocol(streamInput.targetProviderConfig?.provider) ?? + normalizeProviderProtocol(streamInput.targetProvider); + const meter = createStreamExperienceMeter({ + contentType: streamInput.upstreamResponse.headers.get("content-type") ?? undefined, + liveRateTracker: publisher.tracker, + onLiveRateActivity: (urgent) => publisher.notifyActivity(urgent), + protocol, + publishLiveRate: true, + requestId: readGatewayRequestId(streamInput) ?? randomUUID() + }); + const source = Readable.fromWeb( + streamInput.upstreamResponse.body as unknown as Parameters[0] + ); + const metered = source.pipe(meter.stream); + return new Response(Readable.toWeb(metered) as ReadableStream, { + headers: new Headers(streamInput.upstreamResponse.headers), + status: streamInput.upstreamResponse.status, + statusText: streamInput.upstreamResponse.statusText + }); +} + async function handleRuntimeConfigControlRoute( config: AppConfig, request: GatewayPluginHttpRequest, diff --git a/packages/core/src/gateway/internal/shared.ts b/packages/core/src/gateway/internal/shared.ts index a07777a6..c2e29e28 100644 --- a/packages/core/src/gateway/internal/shared.ts +++ b/packages/core/src/gateway/internal/shared.ts @@ -206,6 +206,9 @@ export type UpstreamFetchResult = { attempt: UpstreamAttempt; failedAttempts: UpstreamFailedAttempt[]; response: Response; + timing: { + attemptStartedAtMonoMs: number; + }; }; diff --git a/packages/core/src/gateway/request/pipeline.ts b/packages/core/src/gateway/request/pipeline.ts index d531a5c1..df250b63 100644 --- a/packages/core/src/gateway/request/pipeline.ts +++ b/packages/core/src/gateway/request/pipeline.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import type { IncomingMessage, ServerResponse } from "node:http"; import { Readable } from "node:stream"; -import type { ApiKeyConfig, AppConfig, ProfileConfig, RequestRouteTraceChange, RouterFallbackConfig } from "@ccr/core/contracts/app"; +import type { ApiKeyConfig, AppConfig, ProfileConfig, RequestRouteTraceChange, RequestStreamMetrics, RouterFallbackConfig } from "@ccr/core/contracts/app"; import { createSseErrorDetector, markGatewayRequestLogDropped, @@ -60,6 +60,7 @@ import { ccrRouterHttpRoutePath } from "@ccr/core/gateway/core-runtime/router-plugin-contract"; import { isRecord } from "@ccr/core/gateway/internal/value"; +import { combineStreamExperienceMetrics, createStreamExperienceMeter, monotonicNowMs } from "@ccr/core/observability/stream-experience"; export type GatewayRequestPipelineDependencies = { getBrowserWebSearchMcpIntegration: () => BrowserWebSearchMcpIntegration | undefined; @@ -126,11 +127,12 @@ export class GatewayRequestPipeline { const activeConfig = this.config; const method = request.method ?? "GET"; - const requestBody = await readRequestBody(request); - const requestedModel = requestLogRequestedModel(requestBody, path); + const requestStartedAtMonoMs = monotonicNowMs(); const startedAt = Date.now(); const startedAtIso = new Date(startedAt).toISOString(); const requestId = randomUUID(); + const requestBody = await readRequestBody(request); + const requestedModel = requestLogRequestedModel(requestBody, path); const requestUrl = new URL(request.url || path, this.status.endpoint || "http://127.0.0.1").toString(); const routeTrace = shouldRecordRequestLogs(this.config) ? new RequestRouteTraceRecorder(startedAt) @@ -258,6 +260,7 @@ export class GatewayRequestPipeline { let responseCompleted = false; let onClientDisconnect: (() => void) | undefined; let onResponseFinish: (() => void) | undefined; + let streamMetrics: RequestStreamMetrics | undefined; const handleClientDisconnect = () => { if (responseCompleted || response.writableEnded) { return; @@ -327,6 +330,7 @@ export class GatewayRequestPipeline { requestId, resolvedModel: routedModel, routeTrace: routeTrace?.finish({ captureBodyValues: captureBody }), + streamMetrics, responseBodyText, responseBodySizeBytes, responseBodyTruncated, @@ -903,6 +907,7 @@ export class GatewayRequestPipeline { return; } response.writeHead(upstreamResponse.status, Object.fromEntries(filteredResponseHeaders(responseHeaders))); + const responseHeadersAtMonoMs = monotonicNowMs(); if (!upstreamResponse.body) { finalizeOpenRouterDiscountSelection(upstreamResponse.ok); if (shouldCaptureUsage) { @@ -925,7 +930,12 @@ export class GatewayRequestPipeline { return; } - const upstreamBody = Readable.fromWeb(upstreamResponse.body as unknown as import("node:stream/web").ReadableStream); + const upstreamBodySource = Readable.fromWeb(upstreamResponse.body as unknown as import("node:stream/web").ReadableStream); + const upstreamExperienceMeter = createStreamExperienceMeter({ + contentType: contextArchiveSourceContentType, + protocol: responseProtocol + }); + const upstreamBody = upstreamBodySource.pipe(upstreamExperienceMeter.stream); const patchedResponseBody = codexApplyPatchBridgeActive ? codexApplyPatchBridgeResponseStream(upstreamBody, responseHeaders) : upstreamBody; @@ -959,10 +969,31 @@ export class GatewayRequestPipeline { const clientResponseBody = rewriteAnthropicResponseModel && clientVisibleResponseModel ? rewriteAnthropicMessageStartModelStream(responseBody, clientVisibleResponseModel) : responseBody; - const responseStreams = uniqueStreams([upstreamBody, patchedResponseBody, multiAgentResponseBody, hostedWebSearchResponseBody, responseBody, clientResponseBody]); const sampler = createBodySampler(); const sseErrorDetector = createSseErrorDetector(responseHeaders.get("content-type") ?? undefined); let streamDetectedError: string | undefined; + const clientExperienceMeter = createStreamExperienceMeter({ + contentType: responseHeaders.get("content-type") ?? undefined, + onChunk: (chunk) => { + sampler.append(chunk); + streamDetectedError ??= sseErrorDetector.append(chunk); + }, + protocol: responseProtocol, + publishLiveRate: true, + requestId + }); + const meteredClientResponseBody = clientResponseBody.pipe(clientExperienceMeter.stream); + const responseStreams = uniqueStreams([ + upstreamBodySource, + upstreamExperienceMeter.stream, + upstreamBody, + patchedResponseBody, + multiAgentResponseBody, + hostedWebSearchResponseBody, + responseBody, + clientResponseBody, + meteredClientResponseBody + ]); let upstreamStreamEnded = false; let logRecorded = false; const writeStreamLog = (error?: string) => { @@ -977,6 +1008,18 @@ export class GatewayRequestPipeline { terminalEventSeen: sseErrorDetector.hasTerminalEvent(), upstreamStatus: upstreamResponse.status }); + const clientExperience = clientExperienceMeter.snapshot(); + const upstreamExperience = upstreamExperienceMeter.snapshot(); + streamMetrics = clientExperience.active || upstreamExperience.active + ? combineStreamExperienceMetrics({ + client: clientExperience, + requestStartedAtMs: requestStartedAtMonoMs, + responseHeadersAtMs: responseHeadersAtMonoMs, + sampleStatus: outcome.error ? "partial" : "complete", + upstream: upstreamExperience, + upstreamAttemptStartedAtMs: upstreamResult.timing.attemptStartedAtMonoMs + }) + : undefined; finalizeOpenRouterDiscountSelection(outcome.statusCode >= 200 && outcome.statusCode < 400 && !outcome.error); writeRequestLog( outcome.statusCode, @@ -988,9 +1031,11 @@ export class GatewayRequestPipeline { ); }; onClientDisconnect = () => { + upstreamExperienceMeter.finish(); + clientExperienceMeter.finish(); streamDetectedError ??= sseErrorDetector.finish(); writeStreamLog(); - clientResponseBody.unpipe(response); + meteredClientResponseBody.unpipe(response); destroyResponseStreams(responseStreams); }; onResponseFinish = () => { @@ -1000,6 +1045,8 @@ export class GatewayRequestPipeline { }; const onResponseStreamError = (error: Error) => { failContextArchiveRequest(contextArchiveRecord, contextArchiveRequestConfig); + upstreamExperienceMeter.finish(); + clientExperienceMeter.finish(); streamDetectedError ??= sseErrorDetector.finish(); writeStreamLog(clientDisconnected ? clientDisconnectMessage : formatUpstreamErrorForLog(error, { attempts: upstreamResult.failedAttempts.length + 1, @@ -1016,11 +1063,7 @@ export class GatewayRequestPipeline { for (const stream of responseStreams) { stream.on("error", onResponseStreamError); } - clientResponseBody.on("data", (chunk) => { - sampler.append(chunk); - streamDetectedError ??= sseErrorDetector.append(chunk); - }); - clientResponseBody.once("end", () => { + meteredClientResponseBody.once("end", () => { upstreamStreamEnded = true; streamDetectedError ??= sseErrorDetector.finish(); if (responseCompleted || response.writableEnded) { @@ -1028,7 +1071,7 @@ export class GatewayRequestPipeline { } }); if (shouldCaptureUsage) { - clientResponseBody.once("end", () => { + meteredClientResponseBody.once("end", () => { recordUsage({ bodyText: sampler.read(), client, @@ -1048,7 +1091,7 @@ export class GatewayRequestPipeline { onClientDisconnect(); return; } - clientResponseBody.pipe(response); + meteredClientResponseBody.pipe(response); } async replayContextArchive(input: ContextArchiveReplayInput): Promise { diff --git a/packages/core/src/gateway/upstream/executor.ts b/packages/core/src/gateway/upstream/executor.ts index c4db6588..cea16886 100644 --- a/packages/core/src/gateway/upstream/executor.ts +++ b/packages/core/src/gateway/upstream/executor.ts @@ -22,6 +22,7 @@ import { retryDelayAfterNetworkError, retryDelayAfterStatus, shouldFallbackAfter import { claudeCodeOauthBetaHeader, claudeCodeOauthRequiredBeta, UpstreamRequestError } from "@ccr/core/gateway/internal/shared"; import type { ApiKeyLimitUsage, ProviderCredentialRoutingTarget, UpstreamAttempt, UpstreamFailedAttempt, UpstreamFetchResult } from "@ccr/core/gateway/internal/shared"; import type { RouteTraceObserver } from "@ccr/core/observability/route-trace"; +import { monotonicNowMs } from "@ccr/core/observability/stream-experience"; const providerCredentialSpilloverThreshold = 0.8; const openRouterDiscountModelHeader = "x-ccr-openrouter-discount-model"; @@ -431,6 +432,7 @@ export async function fetchUpstreamWithFallback(input: { attempt.target?.kind === "provider" ? attempt.target.provider.name : undefined ); const attemptStartedAt = Date.now(); + const attemptStartedAtMonoMs = monotonicNowMs(); input.trace?.capture({ attempt: attemptNumber, changes: [ @@ -529,7 +531,10 @@ export async function fetchUpstreamWithFallback(input: { return { attempt, failedAttempts, - response + response, + timing: { + attemptStartedAtMonoMs + } }; } catch (error) { const message = formatError(error); diff --git a/packages/core/src/observability/request-log-runtime.ts b/packages/core/src/observability/request-log-runtime.ts index cad097f8..203ddbde 100644 --- a/packages/core/src/observability/request-log-runtime.ts +++ b/packages/core/src/observability/request-log-runtime.ts @@ -1086,7 +1086,8 @@ function plainHeaderRecord(value: Headers | Record; @@ -252,7 +260,12 @@ type StoredRequestLogEntry = { responseModel: string; responseHeaders: Record; statusCode: number; + streamSpeedSampleStatus?: StreamSpeedSampleStatus; + tailMs?: number; + timeToFirstSignalMs?: number; + timeToFirstTextMs?: number; totalTokens: number; + upstreamTimeToFirstSignalMs?: number; url: string; }; @@ -662,6 +675,7 @@ export class RequestLogStore { responseHeaders, url: input.url }); + const streamMetrics = resolveStoredStreamMetrics(input.streamMetrics, outputTokens, reasoningTokens); const statement = this.insertRequestStatement ??= database.prepare(` INSERT OR IGNORE INTO request_logs ( @@ -711,8 +725,9 @@ export class RequestLogStore { response_body_size_bytes, response_body_truncated, response_body_ref, + stream_metrics_json, error - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `); let inserted = false; @@ -764,6 +779,7 @@ export class RequestLogStore { responseBody.sizeBytes, responseBody.truncated ? 1 : 0, responseBody.bodyRef ?? "", + streamMetrics ? JSON.stringify(streamMetrics) : "", responseError ?? "" ); if (result.changes === 0) return; @@ -1086,6 +1102,7 @@ export class RequestLogStore { status_code, ok, duration_ms, + stream_metrics_json, input_tokens, output_tokens, reasoning_tokens, @@ -1434,6 +1451,7 @@ export class RequestLogStore { response_body_size_bytes INTEGER NOT NULL DEFAULT 0, response_body_truncated INTEGER NOT NULL DEFAULT 0, response_body_ref TEXT NOT NULL DEFAULT '', + stream_metrics_json TEXT NOT NULL DEFAULT '', error TEXT NOT NULL DEFAULT '' ); @@ -4465,6 +4483,7 @@ function ensureRequestLogSchema(database: SqlDatabase): void { addColumn("response_body_size_bytes", "INTEGER NOT NULL DEFAULT 0"); addColumn("response_body_truncated", "INTEGER NOT NULL DEFAULT 0"); addColumn("response_body_ref", "TEXT NOT NULL DEFAULT ''"); + addColumn("stream_metrics_json", "TEXT NOT NULL DEFAULT ''"); addColumn("error", "TEXT NOT NULL DEFAULT ''"); if (needsModelSummaryMigration) { @@ -5042,6 +5061,7 @@ function readRequestLogById(database: SqlDatabase, id: number): StoredRequestLog status_code, ok, duration_ms, + stream_metrics_json, input_tokens, output_tokens, reasoning_tokens, @@ -5075,6 +5095,12 @@ function readRequestLogById(database: SqlDatabase, id: number): StoredRequestLog function toRequestLogEntry(row: Record): StoredRequestLogEntry { const costUsd = asFloat(row.cost_usd); + const outputTokens = normalizeCount(row.output_tokens); + const streamMetrics = parseStoredStreamMetrics(row.stream_metrics_json); + const outputTokensPerSecond = streamMetrics?.sampleStatus === "complete" && + streamMetrics.activeOutputMs !== undefined && streamMetrics.activeOutputMs > 0 && outputTokens >= 2 + ? Math.round(((outputTokens - 1) * 1_000 / streamMetrics.activeOutputMs) * 10) / 10 + : undefined; const requestBody = bodyFromRow(row, "request") ?? emptyBody(); const responseBody = bodyFromRow(row, "response"); const requestHeaders = parseHeaderJson(row.request_headers); @@ -5088,6 +5114,7 @@ function toRequestLogEntry(row: Record): StoredRequestLogEntry url: String(row.url ?? "") }); return { + ...(streamMetrics?.activeOutputMs !== undefined ? { activeOutputMs: streamMetrics.activeOutputMs } : {}), cacheReadTokens: normalizeCount(row.cache_read_tokens), cacheWriteTokens: normalizeCount(row.cache_write_tokens), client: normalizeLabel(String(row.client ?? ""), "unknown"), @@ -5102,13 +5129,17 @@ function toRequestLogEntry(row: Record): StoredRequestLogEntry id: normalizeCount(row.id), inputTokens: normalizeCount(row.input_tokens), isStream, + ...(streamMetrics?.maxInterEventGapMs !== undefined ? { maxInterEventGapMs: streamMetrics.maxInterEventGapMs } : {}), method: String(row.method ?? ""), model: normalizeLabel(String(row.model ?? ""), "unknown"), ok: normalizeCount(row.ok) === 1, - outputTokens: normalizeCount(row.output_tokens), + outputTokens, + ...(outputTokensPerSecond !== undefined ? { outputTokensPerSecond } : {}), path: normalizeLabel(String(row.path ?? ""), "/"), + ...(streamMetrics?.p95InterEventGapMs !== undefined ? { p95InterEventGapMs: streamMetrics.p95InterEventGapMs } : {}), provider: normalizeLabel(String(row.provider ?? ""), "unknown"), reasoningTokens: normalizeCount(row.reasoning_tokens), + ...(streamMetrics?.responseHeadersMs !== undefined ? { responseHeadersMs: streamMetrics.responseHeadersMs } : {}), requestedModel: normalizeLabel(String(row.requested_model ?? ""), ""), requestBody, requestHeaders, @@ -5122,11 +5153,80 @@ function toRequestLogEntry(row: Record): StoredRequestLogEntry responseHeaders, responseModel: normalizeLabel(String(row.response_model ?? ""), ""), statusCode: normalizeCount(row.status_code), + ...(streamMetrics ? { streamSpeedSampleStatus: streamMetrics.sampleStatus } : {}), + ...(streamMetrics?.tailMs !== undefined ? { tailMs: streamMetrics.tailMs } : {}), + ...(streamMetrics?.timeToFirstSignalMs !== undefined ? { timeToFirstSignalMs: streamMetrics.timeToFirstSignalMs } : {}), + ...(streamMetrics?.timeToFirstTextMs !== undefined ? { timeToFirstTextMs: streamMetrics.timeToFirstTextMs } : {}), totalTokens: normalizeCount(row.total_tokens), + ...(streamMetrics?.upstreamTimeToFirstSignalMs !== undefined + ? { upstreamTimeToFirstSignalMs: streamMetrics.upstreamTimeToFirstSignalMs } + : {}), url: String(row.url ?? "") }; } +function resolveStoredStreamMetrics( + metrics: RequestStreamMetrics | undefined, + outputTokens: number, + reasoningTokens: number +): RequestStreamMetrics | undefined { + if (!metrics) { + return undefined; + } + let sampleStatus = metrics.sampleStatus; + if (sampleStatus === "complete") { + if (outputTokens === 0) { + sampleStatus = "usage_missing"; + } else if (outputTokens < 2) { + sampleStatus = "insufficient_tokens"; + } else if (reasoningTokens > 0 && !metrics.reasoningObserved) { + sampleStatus = "hidden_reasoning"; + } else if (metrics.activeOutputMs === undefined || metrics.activeOutputMs <= 0) { + sampleStatus = "batched_output"; + } + } + return { ...metrics, sampleStatus }; +} + +function parseStoredStreamMetrics(value: SqlValue): RequestStreamMetrics | undefined { + if (typeof value !== "string" || !value.trim()) { + return undefined; + } + const parsed = parseJson(value); + if (!isRecord(parsed) || !isStreamSpeedSampleStatus(parsed.sampleStatus)) { + return undefined; + } + return { + ...optionalStreamMetricNumber("activeOutputMs", parsed.activeOutputMs), + estimatedOutputTokens: normalizeCount(parsed.estimatedOutputTokens), + ...optionalStreamMetricNumber("maxInterEventGapMs", parsed.maxInterEventGapMs), + ...optionalStreamMetricNumber("p95InterEventGapMs", parsed.p95InterEventGapMs), + reasoningObserved: parsed.reasoningObserved === true, + ...optionalStreamMetricNumber("responseHeadersMs", parsed.responseHeadersMs), + sampleStatus: parsed.sampleStatus, + ...optionalStreamMetricNumber("tailMs", parsed.tailMs), + textObserved: parsed.textObserved === true, + ...optionalStreamMetricNumber("timeToFirstSignalMs", parsed.timeToFirstSignalMs), + ...optionalStreamMetricNumber("timeToFirstTextMs", parsed.timeToFirstTextMs), + toolObserved: parsed.toolObserved === true, + ...optionalStreamMetricNumber("upstreamTimeToFirstSignalMs", parsed.upstreamTimeToFirstSignalMs) + }; +} + +function optionalStreamMetricNumber( + key: Key, + value: unknown +): Partial> { + const number = asNumber(value); + return number === undefined ? {} : { [key]: number } as Partial>; +} + +function isStreamSpeedSampleStatus(value: unknown): value is StreamSpeedSampleStatus { + return value === "complete" || value === "partial" || value === "usage_missing" || + value === "insufficient_tokens" || value === "unsupported_protocol" || + value === "hidden_reasoning" || value === "batched_output"; +} + function bodyFromRow(row: Record, prefix: "request" | "response"): RequestLogBody | undefined { const text = String(row[`${prefix}_body_text`] ?? ""); const sizeBytes = normalizeCount(row[`${prefix}_body_size_bytes`]); diff --git a/packages/core/src/observability/stream-experience.ts b/packages/core/src/observability/stream-experience.ts new file mode 100644 index 00000000..1156830f --- /dev/null +++ b/packages/core/src/observability/stream-experience.ts @@ -0,0 +1,580 @@ +import { EventEmitter } from "node:events"; +import { performance } from "node:perf_hooks"; +import { StringDecoder } from "node:string_decoder"; +import { Transform } from "node:stream"; +import type { GatewayProviderProtocol, RequestStreamMetrics, StreamSpeedSampleStatus } from "@ccr/core/contracts/app"; + +const liveRateWindowMs = 2_000; +const maxRecordedGaps = 1_024; + +type SemanticFragmentKind = "reasoning" | "text" | "tool"; + +type SemanticFragment = { + kind: SemanticFragmentKind; + text: string; +}; + +type TokenRateSample = { + atMs: number; + tokens: number; +}; + +type ActiveTokenStream = { + firstTokenAtMs?: number; + samples: TokenRateSample[]; +}; + +export type LiveTokenRateSnapshot = { + activeRequests: number; + tokensPerSecond: number; +}; + +export type StreamExperienceSnapshot = { + active: boolean; + estimatedOutputTokens: number; + firstSignalAtMs?: number; + firstTextAtMs?: number; + lastSignalAtMs?: number; + maxInterEventGapMs?: number; + p95InterEventGapMs?: number; + reasoningObserved: boolean; + streamEndAtMs?: number; + textObserved: boolean; + toolObserved: boolean; +}; + +export type StreamExperienceMeter = { + finish: () => void; + snapshot: () => StreamExperienceSnapshot; + stream: Transform; +}; + +export class LiveTokenRateTracker { + private readonly active = new Map(); + + constructor(private readonly windowMs = liveRateWindowMs) {} + + start(requestId: string): void { + this.active.set(requestId, { samples: [] }); + } + + record(requestId: string, tokens: number, atMs = monotonicNowMs()): void { + const stream = this.active.get(requestId); + if (!stream || !Number.isFinite(tokens) || tokens <= 0) { + return; + } + stream.firstTokenAtMs ??= atMs; + stream.samples.push({ atMs, tokens }); + this.prune(stream, atMs); + } + + finish(requestId: string): void { + this.active.delete(requestId); + } + + clear(): void { + this.active.clear(); + } + + snapshot(atMs = monotonicNowMs()): LiveTokenRateSnapshot { + let tokensPerSecond = 0; + for (const stream of this.active.values()) { + this.prune(stream, atMs); + if (stream.firstTokenAtMs === undefined || stream.samples.length === 0) { + continue; + } + const tokenCount = stream.samples.reduce((total, sample) => total + sample.tokens, 0); + const elapsedMs = Math.max(250, Math.min(this.windowMs, atMs - stream.firstTokenAtMs)); + tokensPerSecond += tokenCount * 1_000 / elapsedMs; + } + return { + activeRequests: this.active.size, + tokensPerSecond: Number.isFinite(tokensPerSecond) ? Math.max(0, tokensPerSecond) : 0 + }; + } + + private prune(stream: ActiveTokenStream, atMs: number): void { + const cutoff = atMs - this.windowMs; + while (stream.samples.length > 0 && stream.samples[0].atMs < cutoff) { + stream.samples.shift(); + } + } +} + +const liveTokenRateTracker = new LiveTokenRateTracker(); +const externalLiveTokenRates = new Map(); +const liveTokenRateEvents = new EventEmitter(); + +export function getLiveTokenRateSnapshot(): LiveTokenRateSnapshot { + const local = liveTokenRateTracker.snapshot(); + let activeRequests = local.activeRequests; + let tokensPerSecond = local.tokensPerSecond; + for (const snapshot of externalLiveTokenRates.values()) { + activeRequests += snapshot.activeRequests; + tokensPerSecond += snapshot.tokensPerSecond; + } + return { + activeRequests, + tokensPerSecond: Number.isFinite(tokensPerSecond) ? Math.max(0, tokensPerSecond) : 0 + }; +} + +export function onLiveTokenRateChanged(listener: (snapshot: LiveTokenRateSnapshot) => void): () => void { + liveTokenRateEvents.on("change", listener); + return () => liveTokenRateEvents.off("change", listener); +} + +export function setExternalLiveTokenRateSnapshot(sourceId: string, snapshot?: LiveTokenRateSnapshot): void { + if (!sourceId.trim()) { + return; + } + if (!snapshot) { + externalLiveTokenRates.delete(sourceId); + } else { + externalLiveTokenRates.set(sourceId, { + activeRequests: nonNegativeInteger(snapshot.activeRequests), + tokensPerSecond: nonNegativeNumber(snapshot.tokensPerSecond) + }); + } + emitLiveRateChanged(); +} + +export function monotonicNowMs(): number { + return performance.now(); +} + +export function createStreamExperienceMeter(input: { + contentType?: string; + liveRateTracker?: LiveTokenRateTracker; + now?: () => number; + onChunk?: (chunk: Buffer) => void; + onLiveRateActivity?: (urgent: boolean) => void; + protocol?: GatewayProviderProtocol; + publishLiveRate?: boolean; + requestId?: string; +}): StreamExperienceMeter { + const parser = new StreamExperienceParser(input.contentType, input.protocol, input.now ?? monotonicNowMs); + const requestId = input.requestId; + const publishLiveRate = Boolean(input.publishLiveRate && requestId && parser.active); + const rateTracker = input.liveRateTracker ?? liveTokenRateTracker; + const publishesGlobalRate = rateTracker === liveTokenRateTracker; + let finished = false; + + if (publishLiveRate && requestId) { + rateTracker.start(requestId); + if (publishesGlobalRate) { + emitLiveRateChanged(); + } + input.onLiveRateActivity?.(true); + } + + const finish = () => { + if (finished) { + return; + } + finished = true; + parser.finish(); + if (publishLiveRate && requestId) { + rateTracker.finish(requestId); + if (publishesGlobalRate) { + emitLiveRateChanged(); + } + input.onLiveRateActivity?.(true); + } + }; + + const stream = new Transform({ + transform(chunk: Buffer | string, _encoding, callback) { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + try { + input.onChunk?.(buffer); + const estimatedTokens = parser.append(buffer); + if (publishLiveRate && requestId && estimatedTokens > 0) { + rateTracker.record(requestId, estimatedTokens); + input.onLiveRateActivity?.(false); + } + } catch { + // Metrics are fail-open: malformed provider events must never interrupt forwarding. + } + callback(null, buffer); + }, + flush(callback) { + finish(); + callback(); + } + }); + stream.once("close", finish); + + return { + finish, + snapshot: () => parser.snapshot(), + stream + }; +} + +export function combineStreamExperienceMetrics(input: { + client: StreamExperienceSnapshot; + requestStartedAtMs: number; + responseHeadersAtMs?: number; + sampleStatus: StreamSpeedSampleStatus; + upstream: StreamExperienceSnapshot; + upstreamAttemptStartedAtMs?: number; +}): RequestStreamMetrics { + const firstOutputAtMs = input.upstream.firstSignalAtMs; + const lastOutputAtMs = input.upstream.lastSignalAtMs; + const firstClientSignalAtMs = input.client.firstSignalAtMs; + const clientEndAtMs = input.client.streamEndAtMs; + + return { + ...(firstOutputAtMs !== undefined && lastOutputAtMs !== undefined + ? { activeOutputMs: roundedDuration(lastOutputAtMs - firstOutputAtMs) } + : {}), + estimatedOutputTokens: input.upstream.estimatedOutputTokens, + ...(input.client.maxInterEventGapMs !== undefined + ? { maxInterEventGapMs: roundedDuration(input.client.maxInterEventGapMs) } + : {}), + ...(input.client.p95InterEventGapMs !== undefined + ? { p95InterEventGapMs: roundedDuration(input.client.p95InterEventGapMs) } + : {}), + reasoningObserved: input.upstream.reasoningObserved, + ...(input.responseHeadersAtMs !== undefined + ? { responseHeadersMs: roundedDuration(input.responseHeadersAtMs - input.requestStartedAtMs) } + : {}), + sampleStatus: input.client.active && input.upstream.active ? input.sampleStatus : "unsupported_protocol", + ...(clientEndAtMs !== undefined && input.client.lastSignalAtMs !== undefined + ? { tailMs: roundedDuration(clientEndAtMs - input.client.lastSignalAtMs) } + : {}), + textObserved: input.upstream.textObserved, + ...(firstClientSignalAtMs !== undefined + ? { timeToFirstSignalMs: roundedDuration(firstClientSignalAtMs - input.requestStartedAtMs) } + : {}), + ...(input.client.firstTextAtMs !== undefined + ? { timeToFirstTextMs: roundedDuration(input.client.firstTextAtMs - input.requestStartedAtMs) } + : {}), + toolObserved: input.upstream.toolObserved, + ...(firstOutputAtMs !== undefined && input.upstreamAttemptStartedAtMs !== undefined + ? { upstreamTimeToFirstSignalMs: roundedDuration(firstOutputAtMs - input.upstreamAttemptStartedAtMs) } + : {}) + }; +} + +class StreamExperienceParser { + readonly active: boolean; + private currentEvent = ""; + private dataLines: string[] = []; + private readonly decoder = new StringDecoder("utf8"); + private estimatedOutputTokens = 0; + private firstSignalAtMs?: number; + private firstTextAtMs?: number; + private readonly gaps: number[] = []; + private lastSignalAtMs?: number; + private maxInterEventGapMs?: number; + private pendingLine = ""; + private reasoningObserved = false; + private readonly sse: boolean; + private streamEndAtMs?: number; + private textObserved = false; + private readonly tokenEstimator = new StreamingTokenEstimator(); + private toolObserved = false; + + constructor( + contentType: string | undefined, + private readonly protocol: GatewayProviderProtocol | undefined, + private readonly now: () => number + ) { + const normalizedContentType = contentType?.toLowerCase() ?? ""; + this.sse = normalizedContentType.includes("text/event-stream"); + this.active = this.sse || + normalizedContentType.includes("application/x-ndjson") || + normalizedContentType.includes("application/json-seq"); + } + + append(chunk: Buffer): number { + if (!this.active) { + return 0; + } + return this.processText(this.decoder.write(chunk)); + } + + finish(): void { + if (this.streamEndAtMs !== undefined) { + return; + } + if (this.active) { + this.processText(this.decoder.end()); + if (this.pendingLine) { + this.processLine(this.pendingLine.endsWith("\r") ? this.pendingLine.slice(0, -1) : this.pendingLine); + this.pendingLine = ""; + } + if (this.currentEvent || this.dataLines.length > 0) { + this.flushEvent(); + } + } + this.streamEndAtMs = this.now(); + } + + snapshot(): StreamExperienceSnapshot { + return { + active: this.active, + estimatedOutputTokens: this.estimatedOutputTokens, + ...(this.firstSignalAtMs !== undefined ? { firstSignalAtMs: this.firstSignalAtMs } : {}), + ...(this.firstTextAtMs !== undefined ? { firstTextAtMs: this.firstTextAtMs } : {}), + ...(this.lastSignalAtMs !== undefined ? { lastSignalAtMs: this.lastSignalAtMs } : {}), + ...(this.maxInterEventGapMs !== undefined ? { maxInterEventGapMs: this.maxInterEventGapMs } : {}), + ...(this.gaps.length > 0 ? { p95InterEventGapMs: percentile(this.gaps, 0.95) } : {}), + reasoningObserved: this.reasoningObserved, + ...(this.streamEndAtMs !== undefined ? { streamEndAtMs: this.streamEndAtMs } : {}), + textObserved: this.textObserved, + toolObserved: this.toolObserved + }; + } + + private processText(text: string): number { + let estimatedTokens = 0; + this.pendingLine += text; + while (true) { + const newlineIndex = this.pendingLine.indexOf("\n"); + if (newlineIndex === -1) { + break; + } + const rawLine = this.pendingLine.slice(0, newlineIndex); + this.pendingLine = this.pendingLine.slice(newlineIndex + 1); + estimatedTokens += this.processLine(rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine); + } + return estimatedTokens; + } + + private processLine(line: string): number { + if (!this.sse) { + return line.trim() ? this.processPayload(line, "") : 0; + } + if (line === "") { + return this.flushEvent(); + } + if (line.startsWith(":")) { + return 0; + } + const separator = line.indexOf(":"); + const field = separator === -1 ? line : line.slice(0, separator); + const rawValue = separator === -1 ? "" : line.slice(separator + 1); + const value = rawValue.startsWith(" ") ? rawValue.slice(1) : rawValue; + if (field === "event") { + this.currentEvent = value.trim(); + return 0; + } + if (field === "data") { + this.dataLines.push(value); + return 0; + } + return 0; + } + + private flushEvent(): number { + const data = this.dataLines.join("\n"); + const event = this.currentEvent; + this.currentEvent = ""; + this.dataLines = []; + if (!data || data.trim() === "[DONE]") { + return 0; + } + return this.processPayload(data, event); + } + + private processPayload(data: string, event: string): number { + let payload: unknown; + try { + payload = JSON.parse(data) as unknown; + } catch { + return 0; + } + const fragments = semanticFragments(payload, event, this.protocol) + .filter((item) => item.text.length > 0); + if (fragments.length === 0) { + return 0; + } + const atMs = this.now(); + this.firstSignalAtMs ??= atMs; + if (this.lastSignalAtMs !== undefined && atMs > this.lastSignalAtMs) { + const gap = atMs - this.lastSignalAtMs; + this.maxInterEventGapMs = Math.max(this.maxInterEventGapMs ?? 0, gap); + this.gaps.push(gap); + if (this.gaps.length > maxRecordedGaps) { + this.gaps.shift(); + } + } + this.lastSignalAtMs = atMs; + for (const item of fragments) { + if (item.kind === "text") { + this.textObserved = true; + if (/\S/u.test(item.text)) { + this.firstTextAtMs ??= atMs; + } + } else if (item.kind === "reasoning") { + this.reasoningObserved = true; + } else { + this.toolObserved = true; + } + } + const estimatedTokens = this.tokenEstimator.add(fragments.map((item) => item.text).join("")); + this.estimatedOutputTokens += estimatedTokens; + return estimatedTokens; + } +} + +class StreamingTokenEstimator { + private emittedTokens = 0; + private weightedCharacters = 0; + + add(text: string): number { + for (const character of text) { + const codePoint = character.codePointAt(0) ?? 0; + if (codePoint >= 0x3400 && codePoint <= 0x9fff) { + this.weightedCharacters += 1; + } else if (codePoint <= 0x7f) { + this.weightedCharacters += 0.25; + } else { + this.weightedCharacters += 0.5; + } + } + const nextTotal = Math.max(text.length > 0 ? 1 : 0, Math.floor(this.weightedCharacters)); + const delta = Math.max(0, nextTotal - this.emittedTokens); + this.emittedTokens = nextTotal; + return delta; + } +} + +function semanticFragments(payload: unknown, eventName: string, protocol: GatewayProviderProtocol | undefined): SemanticFragment[] { + if (!isRecord(payload)) { + return []; + } + if (protocol === "anthropic_messages") { + return anthropicFragments(payload); + } + if (protocol === "openai_chat_completions") { + return openAiChatFragments(payload); + } + if (protocol === "openai_responses") { + return openAiResponsesFragments(payload, eventName); + } + if (protocol === "gemini_generate_content" || protocol === "gemini_interactions") { + return geminiFragments(payload); + } + return [ + ...anthropicFragments(payload), + ...openAiChatFragments(payload), + ...openAiResponsesFragments(payload, eventName), + ...geminiFragments(payload) + ]; +} + +function anthropicFragments(payload: Record): SemanticFragment[] { + const delta = isRecord(payload.delta) ? payload.delta : undefined; + const contentBlock = isRecord(payload.content_block) ? payload.content_block : undefined; + return compactFragments([ + fragment("text", readString(delta?.text)), + fragment("reasoning", readString(delta?.thinking)), + fragment("tool", readString(delta?.partial_json)), + fragment("text", readString(contentBlock?.text)), + fragment("reasoning", readString(contentBlock?.thinking)), + fragment("tool", readString(contentBlock?.name)) + ]); +} + +function openAiChatFragments(payload: Record): SemanticFragment[] { + const fragments: Array = []; + const choices = Array.isArray(payload.choices) ? payload.choices : []; + for (const choice of choices) { + if (!isRecord(choice) || !isRecord(choice.delta)) { + continue; + } + const delta = choice.delta; + fragments.push(fragment("text", readString(delta.content))); + fragments.push(fragment("reasoning", readString(delta.reasoning_content))); + fragments.push(fragment("reasoning", readString(delta.reasoning))); + const toolCalls = Array.isArray(delta.tool_calls) ? delta.tool_calls : []; + for (const toolCall of toolCalls) { + const fn = isRecord(toolCall) && isRecord(toolCall.function) ? toolCall.function : undefined; + fragments.push(fragment("tool", readString(fn?.name))); + fragments.push(fragment("tool", readString(fn?.arguments))); + } + } + return compactFragments(fragments); +} + +function openAiResponsesFragments(payload: Record, eventName: string): SemanticFragment[] { + const type = (readString(payload.type) ?? eventName).toLowerCase(); + const delta = readString(payload.delta); + if (!delta || !type.endsWith(".delta")) { + return []; + } + if (type.includes("output_text") || type.includes("refusal")) { + return [{ kind: "text", text: delta }]; + } + if (type.includes("reasoning") || type.includes("summary_text")) { + return [{ kind: "reasoning", text: delta }]; + } + if (type.includes("function_call") || type.includes("tool")) { + return [{ kind: "tool", text: delta }]; + } + return []; +} + +function geminiFragments(payload: Record): SemanticFragment[] { + const fragments: Array = []; + const candidates = Array.isArray(payload.candidates) ? payload.candidates : []; + for (const candidate of candidates) { + const content = isRecord(candidate) && isRecord(candidate.content) ? candidate.content : undefined; + const parts = Array.isArray(content?.parts) ? content.parts : []; + for (const part of parts) { + if (!isRecord(part)) { + continue; + } + fragments.push(fragment(part.thought === true ? "reasoning" : "text", readString(part.text))); + if (isRecord(part.functionCall)) { + fragments.push(fragment("tool", readString(part.functionCall.name))); + if (part.functionCall.args !== undefined) { + fragments.push(fragment("tool", JSON.stringify(part.functionCall.args))); + } + } + } + } + return compactFragments(fragments); +} + +function fragment(kind: SemanticFragmentKind, text: string | undefined): SemanticFragment | undefined { + return text === undefined ? undefined : { kind, text }; +} + +function compactFragments(fragments: Array): SemanticFragment[] { + return fragments.filter((item): item is SemanticFragment => item !== undefined); +} + +function percentile(values: number[], quantile: number): number { + const sorted = [...values].sort((left, right) => left - right); + const index = Math.min(sorted.length - 1, Math.max(0, Math.ceil(sorted.length * quantile) - 1)); + return sorted[index] ?? 0; +} + +function roundedDuration(value: number): number { + return Math.max(0, Math.round(value)); +} + +function readString(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; +} + +function isRecord(value: unknown): value is Record { + return Boolean(value) && typeof value === "object" && !Array.isArray(value); +} + +function emitLiveRateChanged(): void { + liveTokenRateEvents.emit("change", getLiveTokenRateSnapshot()); +} + +function nonNegativeInteger(value: number): number { + return Number.isFinite(value) ? Math.max(0, Math.trunc(value)) : 0; +} + +function nonNegativeNumber(value: number): number { + return Number.isFinite(value) ? Math.max(0, value) : 0; +} diff --git a/packages/core/test/integration/agents/codex-cli-middleware-runtime.test.mjs b/packages/core/test/integration/agents/codex-cli-middleware-runtime.test.mjs index 66b3f73d..311ab127 100644 --- a/packages/core/test/integration/agents/codex-cli-middleware-runtime.test.mjs +++ b/packages/core/test/integration/agents/codex-cli-middleware-runtime.test.mjs @@ -478,9 +478,12 @@ test("Codex app-server merges CCR Fast Mode catalog metadata without spoofing au account: { type: "amazonBedrock", credentialSource: "codexManaged" }, requiresOpenaiAuth: false }); - assert.deepEqual(responses.get(4).result.requirements.featureRequirements, { - fast_mode: true, - other_feature: false + assert.deepEqual(responses.get(4).result.requirements, { + application: null, + featureRequirements: { + fast_mode: true, + other_feature: false + } }); }); diff --git a/packages/core/test/integration/observability/request-log-store.test.mjs b/packages/core/test/integration/observability/request-log-store.test.mjs index 3678a854..56e2859c 100644 --- a/packages/core/test/integration/observability/request-log-store.test.mjs +++ b/packages/core/test/integration/observability/request-log-store.test.mjs @@ -318,6 +318,62 @@ test("RequestLogStore keeps list rows lightweight and detail rows complete", asy } }); +test("RequestLogStore persists stream experience metrics and derives authoritative TPS", async () => { + const dir = mkdtempSync(path.join(tmpdir(), "ccr-request-log-speed-test-")); + let store; + try { + store = new RequestLogStore(path.join(dir, "request-logs.sqlite")); + await store.record({ + completedAt: new Date().toISOString(), + durationMs: 900, + method: "POST", + path: "/v1/messages", + providerName: "test-provider", + requestBody: Buffer.from(JSON.stringify({ messages: [], model: "test-model", stream: true })), + requestHeaders: { "content-type": "application/json" }, + requestId: "request-log-speed-test", + responseBodyText: [ + 'event: message_start\ndata: {"type":"message_start","message":{"usage":{"input_tokens":10,"output_tokens":1}}}', + "", + 'event: message_delta\ndata: {"type":"message_delta","usage":{"output_tokens":21}}', + "", + 'event: message_stop\ndata: {"type":"message_stop"}', + "" + ].join("\n"), + responseHeaders: { "content-type": "text/event-stream" }, + startedAt: new Date().toISOString(), + statusCode: 200, + streamMetrics: { + activeOutputMs: 500, + estimatedOutputTokens: 19, + maxInterEventGapMs: 180, + p95InterEventGapMs: 120, + reasoningObserved: false, + responseHeadersMs: 100, + sampleStatus: "complete", + tailMs: 40, + textObserved: true, + timeToFirstSignalMs: 150, + timeToFirstTextMs: 170, + toolObserved: false, + upstreamTimeToFirstSignalMs: 80 + }, + url: "http://127.0.0.1:3456/v1/messages" + }); + + const entry = (await store.list({ pageSize: 25 })).items[0]; + assert.equal(entry.outputTokens, 21); + assert.equal(entry.outputTokensPerSecond, 40); + assert.equal(entry.timeToFirstSignalMs, 150); + assert.equal(entry.timeToFirstTextMs, 170); + assert.equal(entry.maxInterEventGapMs, 180); + assert.equal(entry.streamSpeedSampleStatus, "complete"); + } finally { + await store?.close(); + rmSync(dir, { force: true, recursive: true }); + } +}); + test("RequestLogStore keeps large request bodies in sidecar storage and reads them by chunk", async () => { const dir = mkdtempSync(path.join(tmpdir(), "ccr-request-log-sidecar-test-")); let store; diff --git a/packages/core/test/unit/gateway/router-plugin.test.mjs b/packages/core/test/unit/gateway/router-plugin.test.mjs index 73af31f8..0cadb935 100644 --- a/packages/core/test/unit/gateway/router-plugin.test.mjs +++ b/packages/core/test/unit/gateway/router-plugin.test.mjs @@ -7,6 +7,9 @@ import { ccrCodexBridgeResponseHookKey, ccrCodexBridgeStreamHookKey, ccrCodexMultiAgentBridgeHeader, + ccrLiveTokenRateConfigMessageType, + ccrLiveTokenRateSnapshotMessageType, + ccrLiveTokenRateStreamHookKey, ccrOpenRouterDiscountFinalizeResponseHookKey, ccrRuntimeConfigReloadMessageType, ccrRouteReasonHeader, @@ -83,6 +86,82 @@ test("CCR router core plugin exposes route endpoint and beforeRouting transform" assert.equal(resolved.requestBody.model, "beta"); }); +test("CCR router core plugin publishes live token rate snapshots from the single runtime", async () => { + const originalSend = process.send; + const messages = []; + process.send = (message) => { + messages.push(message); + return true; + }; + try { + const config = createDefaultAppConfig(); + const plugin = await createGatewayPlugin({ plugin: { config: { appConfig: config } } }); + const streamHook = plugin.streamHooks.find((item) => item.key === ccrLiveTokenRateStreamHookKey); + assert.ok(streamHook); + const disabledResult = await streamHook.transformResponse({ + request: { id: "single-runtime-rate-disabled" }, + targetProvider: "openai", + targetProviderConfig: { type: "openai_chat_completions" }, + upstreamResponse: new Response("disabled", { + headers: { "content-type": "text/event-stream" }, + status: 200 + }) + }); + assert.equal(disabledResult, undefined); + process.emit("message", { + enabled: true, + protocolVersion: 1, + type: ccrLiveTokenRateConfigMessageType + }); + const encoder = new TextEncoder(); + const upstreamResponse = new Response(new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('data: {"choices":[{"delta":{"content":"first token"}}]}\n\n')); + setTimeout(() => { + controller.enqueue(encoder.encode('data: {"choices":[{"delta":{"content":" second token"}}]}\n\n')); + }, 275); + setTimeout(() => controller.close(), 325); + } + }), { + headers: { "content-type": "text/event-stream" }, + status: 200 + }); + const meteredResponse = await streamHook.transformResponse({ + request: { + headers: {}, + id: "single-runtime-rate-test", + method: "POST", + url: "/v1/chat/completions" + }, + targetProvider: "openai", + targetProviderConfig: { + provider: "openai_chat_completions", + type: "openai_chat_completions" + }, + upstreamResponse + }); + + assert.ok(meteredResponse instanceof Response); + await meteredResponse.text(); + assert.ok(messages.some((message) => + message?.type === ccrLiveTokenRateSnapshotMessageType && + message.activeRequests === 1 && + message.tokensPerSecond > 0 + )); + } finally { + process.emit("message", { + enabled: false, + protocolVersion: 1, + type: ccrLiveTokenRateConfigMessageType + }); + if (originalSend) { + process.send = originalSend; + } else { + delete process.send; + } + } +}); + test("CCR router core plugin resolves bare Codex companion models through the authenticated profile provider", async () => { const config = createDefaultAppConfig(); config.APIKEY = "bs-key"; diff --git a/packages/core/test/unit/observability/stream-experience.test.mjs b/packages/core/test/unit/observability/stream-experience.test.mjs new file mode 100644 index 00000000..862e0d10 --- /dev/null +++ b/packages/core/test/unit/observability/stream-experience.test.mjs @@ -0,0 +1,166 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { + combineStreamExperienceMetrics, + createStreamExperienceMeter, + getLiveTokenRateSnapshot, + LiveTokenRateTracker, + setExternalLiveTokenRateSnapshot +} from "@ccr/core/observability/stream-experience.ts"; + +test("stream experience meter parses fragmented Anthropic semantic events", async () => { + const timestamps = [100, 350, 500]; + const meter = createStreamExperienceMeter({ + contentType: "text/event-stream", + now: () => timestamps.shift() ?? 500, + protocol: "anthropic_messages" + }); + const body = [ + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"thinking_delta","thinking":"分析"}}\n\n', + 'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"Hello world"}}\n\n', + 'event: message_stop\ndata: {"type":"message_stop"}\n\n' + ].join(""); + const encoded = Buffer.from(body); + + await writeMeter(meter, [encoded.subarray(0, 17), encoded.subarray(17, 91), encoded.subarray(91)]); + + assert.deepEqual(meter.snapshot(), { + active: true, + estimatedOutputTokens: 4, + firstSignalAtMs: 100, + firstTextAtMs: 350, + lastSignalAtMs: 350, + maxInterEventGapMs: 250, + p95InterEventGapMs: 250, + reasoningObserved: true, + streamEndAtMs: 500, + textObserved: true, + toolObserved: false + }); +}); + +test("stream experience meter supports OpenAI chat, Responses, and Gemini NDJSON", async () => { + const cases = [ + { + body: 'data: {"choices":[{"delta":{"content":"hello"}}]}\n\n', + contentType: "text/event-stream", + protocol: "openai_chat_completions" + }, + { + body: 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"hello"}\n\n', + contentType: "text/event-stream", + protocol: "openai_responses" + }, + { + body: '{"candidates":[{"content":{"parts":[{"text":"你好"}]}}]}\n', + contentType: "application/x-ndjson", + protocol: "gemini_generate_content" + } + ]; + + for (const item of cases) { + const meter = createStreamExperienceMeter({ + contentType: item.contentType, + now: () => 100, + protocol: item.protocol + }); + await writeMeter(meter, [Buffer.from(item.body)]); + const snapshot = meter.snapshot(); + assert.equal(snapshot.active, true, item.protocol); + assert.equal(snapshot.textObserved, true, item.protocol); + assert.equal(snapshot.firstTextAtMs, 100, item.protocol); + assert.ok(snapshot.estimatedOutputTokens >= 1, item.protocol); + } +}); + +test("live token rate sums concurrent active streams over the rolling window", () => { + const tracker = new LiveTokenRateTracker(2_000); + tracker.start("one"); + tracker.start("two"); + tracker.record("one", 10, 1_000); + tracker.record("two", 20, 1_000); + + assert.deepEqual(tracker.snapshot(2_000), { + activeRequests: 2, + tokensPerSecond: 30 + }); + + tracker.finish("one"); + assert.deepEqual(tracker.snapshot(2_000), { + activeRequests: 1, + tokensPerSecond: 20 + }); + + assert.deepEqual(tracker.snapshot(3_100), { + activeRequests: 1, + tokensPerSecond: 0 + }); +}); + +test("live token rate includes and clears child-runtime snapshots", () => { + try { + setExternalLiveTokenRateSnapshot("core-test", { + activeRequests: 2, + tokensPerSecond: 37.5 + }); + assert.deepEqual(getLiveTokenRateSnapshot(), { + activeRequests: 2, + tokensPerSecond: 37.5 + }); + } finally { + setExternalLiveTokenRateSnapshot("core-test"); + } + assert.deepEqual(getLiveTokenRateSnapshot(), { + activeRequests: 0, + tokensPerSecond: 0 + }); +}); + +test("combined metrics keep client latency separate from upstream generation", () => { + const metrics = combineStreamExperienceMetrics({ + client: { + active: true, + estimatedOutputTokens: 8, + firstSignalAtMs: 250, + firstTextAtMs: 300, + lastSignalAtMs: 900, + maxInterEventGapMs: 240, + p95InterEventGapMs: 180, + reasoningObserved: true, + streamEndAtMs: 950, + textObserved: true, + toolObserved: false + }, + requestStartedAtMs: 100, + responseHeadersAtMs: 180, + sampleStatus: "complete", + upstream: { + active: true, + estimatedOutputTokens: 8, + firstSignalAtMs: 220, + lastSignalAtMs: 880, + reasoningObserved: true, + textObserved: true, + toolObserved: false + }, + upstreamAttemptStartedAtMs: 150 + }); + + assert.equal(metrics.timeToFirstSignalMs, 150); + assert.equal(metrics.timeToFirstTextMs, 200); + assert.equal(metrics.upstreamTimeToFirstSignalMs, 70); + assert.equal(metrics.activeOutputMs, 660); + assert.equal(metrics.tailMs, 50); + assert.equal(metrics.maxInterEventGapMs, 240); +}); + +async function writeMeter(meter, chunks) { + meter.stream.resume(); + for (const chunk of chunks) { + meter.stream.write(chunk); + } + await new Promise((resolve, reject) => { + meter.stream.once("error", reject); + meter.stream.end(resolve); + }); +} diff --git a/packages/electron/src/main/tray-controller.ts b/packages/electron/src/main/tray-controller.ts index d35046e1..2f4ecc93 100644 --- a/packages/electron/src/main/tray-controller.ts +++ b/packages/electron/src/main/tray-controller.ts @@ -6,6 +6,7 @@ import { loadAppConfig } from "@ccr/core/config/config"; import { APP_NAME, IPC_CHANNELS } from "@ccr/core/config/constants"; import { getProviderAccountSnapshots } from "@ccr/core/providers/account-service"; import { getTodayUsageTotals, onUsageRecorded } from "@ccr/core/usage/store"; +import { getLiveTokenRateSnapshot, onLiveTokenRateChanged } from "@ccr/core/observability/stream-experience"; import windowsManager from "./windows"; import type { AppConfig, ProviderAccountMeter, TrayBalanceProgressConfig, TrayIconPreference } from "@ccr/core/contracts/app"; @@ -20,6 +21,7 @@ const trayMenuBarIconSize = 20; const trayWindowDarkBackgroundColor = "#1c1c1e"; const trayWindowLightBackgroundColor = "#f2f2f7"; const trayTokenFallbackTitle = "0 tokens"; +const trayTokenRateFallbackTitle = "0 tok/s"; const trayIconFallbackPath = path.join(__dirname, "../assets/tray.png"); const trayMascotIconIds = ["violet", "orange", "cyan"] as const; @@ -40,11 +42,14 @@ class TrayController { private randomTrayIconDateKey?: string; private resolvedRandomTrayIcon?: TrayMascotIconId; private refreshTimer?: NodeJS.Timeout; + private liveRateRefreshTimer?: NodeJS.Timeout; private suppressMainWindowActivationUntil = 0; private tray?: Tray; private trayBalanceProgress?: TrayBalanceProgressConfig; private trayIconPreference: TrayIconPreference = "random"; - private trayTotalTokens = 0; + private trayShowTokenRate = false; + private trayTitle = trayTokenFallbackTitle; + private unsubscribeLiveRateUpdates?: () => void; private unsubscribeUsageUpdates?: () => void; start(): void { @@ -72,10 +77,21 @@ class TrayController { this.unsubscribeUsageUpdates = onUsageRecorded(() => { this.refreshUsageTitle(); }); + this.unsubscribeLiveRateUpdates = onLiveTokenRateChanged((snapshot) => { + if (this.trayShowTokenRate) { + this.applyLiveTokenRateTitle(snapshot); + } + }); this.refreshUsageTitle(); this.refreshTimer = setInterval(() => { this.refreshUsageTitle(); }, 15_000); + this.liveRateRefreshTimer = setInterval(() => { + if (this.trayShowTokenRate) { + this.applyLiveTokenRateTitle(getLiveTokenRateSnapshot()); + } + }, 250); + this.liveRateRefreshTimer.unref?.(); } hidePopover(): void { @@ -93,6 +109,12 @@ class TrayController { clearInterval(this.refreshTimer); this.refreshTimer = undefined; } + if (this.liveRateRefreshTimer) { + clearInterval(this.liveRateRefreshTimer); + this.liveRateRefreshTimer = undefined; + } + this.unsubscribeLiveRateUpdates?.(); + this.unsubscribeLiveRateUpdates = undefined; this.unsubscribeUsageUpdates?.(); this.unsubscribeUsageUpdates = undefined; if (this.detailPopover && !this.detailPopover.isDestroyed()) { @@ -134,12 +156,14 @@ class TrayController { this.resolvedRandomTrayIcon = undefined; } this.trayIconPreference = nextPreference; + this.trayShowTokenRate = nextConfig.trayShowTokenRate === true; this.trayBalanceProgress = normalizeTrayBalanceProgressConfig(nextConfig.trayBalanceProgress); if (nextPreference === "progress" && this.trayBalanceProgress) { await this.refreshBalanceProgressTrayIcon(); - return; + } else { + this.applyTrayIcon(this.resolveTrayIconId(nextPreference)); } - this.applyTrayIcon(this.resolveTrayIconId(nextPreference)); + await this.refreshTrayTitle(); } refreshTheme(theme: AppConfig["theme"]): void { @@ -241,7 +265,7 @@ class TrayController { const menu = Menu.buildFromTemplate([ { enabled: false, - label: formatTokenTitle(this.trayTotalTokens) + label: this.trayTitle }, { type: "separator" }, { @@ -316,9 +340,13 @@ class TrayController { return; } + if (this.trayShowTokenRate) { + this.applyLiveTokenRateTitle(getLiveTokenRateSnapshot()); + return; + } + try { const totals = await getTodayUsageTotals(undefined, { includeProxy: true }); - this.trayTotalTokens = Math.max(0, totals.totalTokens); if (this.trayIconPreference === "progress" && this.trayBalanceProgress) { await this.refreshBalanceProgressTrayIcon(); } @@ -332,12 +360,24 @@ class TrayController { if (!this.tray) { return; } - if (supportsTrayTitle()) { + const titleChanged = this.trayTitle !== title; + this.trayTitle = title; + if (supportsTrayTitle() && titleChanged) { this.tray.setTitle(title); } this.tray.setToolTip(`${APP_NAME} Usage\n${title}`); } + private applyLiveTokenRateTitle(snapshot: { activeRequests: number; tokensPerSecond: number }): void { + const title = snapshot.activeRequests > 0 + ? formatTokenRateTitle(snapshot.tokensPerSecond) + : trayTokenRateFallbackTitle; + this.applyTrayTitle(title); + this.tray?.setToolTip( + `${APP_NAME} Usage\n${title}\n${snapshot.activeRequests} active stream${snapshot.activeRequests === 1 ? "" : "s"} · estimated` + ); + } + private applyTrayIcon(iconId: TrayMascotIconId): void { if (!this.tray) { return; @@ -810,3 +850,11 @@ function formatCompactNumber(value: number): string { function formatTokenTitle(value: number): string { return `${formatCompactNumber(Math.max(0, value))} tokens`; } + +function formatTokenRateTitle(value: number): string { + const rate = Math.max(0, value); + const formatted = new Intl.NumberFormat(undefined, { + maximumFractionDigits: rate < 10 ? 1 : 0 + }).format(rate); + return `${formatted} tok/s`; +} diff --git a/packages/ui/src/pages/home/App.tsx b/packages/ui/src/pages/home/App.tsx index eaa21bfc..b5f809ce 100644 --- a/packages/ui/src/pages/home/App.tsx +++ b/packages/ui/src/pages/home/App.tsx @@ -2519,6 +2519,13 @@ function App() { })); } + function changeTrayShowTokenRate(trayShowTokenRate: boolean) { + updateConfig((config) => ({ + ...config, + trayShowTokenRate + })); + } + function changeTrayBalanceProgress(config: TrayBalanceProgressConfig) { const trayBalanceProgress = normalizeTrayBalanceProgressConfig(config); updateConfig((current) => ({ @@ -3563,6 +3570,7 @@ function App() { onChangeToolHub: changeToolHubConfig, onChangeTrayBalanceProgress: changeTrayBalanceProgress, onChangeTrayIcon: changeTrayIconPreference, + onChangeTrayShowTokenRate: changeTrayShowTokenRate, onChangeTrayWidgets: changeTrayWidgets, onClose: () => setSettingsOpen(false), observability: draftConfig.observability, @@ -3576,6 +3584,7 @@ function App() { providerAccountSnapshots, trayBalanceProgress: normalizeTrayBalanceProgressConfig(draftConfig.trayBalanceProgress), trayIconPreference: draftConfig.trayIcon || "random", + trayShowTokenRate: Boolean(draftConfig.trayShowTokenRate), traySupported, trayWidgets: normalizeTrayWidgets(draftConfig.trayWidgets ?? DEFAULT_TRAY_WIDGETS, draftConfig.trayWindowModules, draftConfig.trayComponentVariants), updateConfig diff --git a/packages/ui/src/pages/home/components/network-logs.tsx b/packages/ui/src/pages/home/components/network-logs.tsx index 1f9b69e4..9218a76f 100644 --- a/packages/ui/src/pages/home/components/network-logs.tsx +++ b/packages/ui/src/pages/home/components/network-logs.tsx @@ -853,6 +853,9 @@ function LogMobileCard({
+ {item.outputTokensPerSecond !== undefined ? ( + + ) : null} {hasCredentialInfo ? : null}
@@ -969,6 +972,16 @@ export function LogExpandedDetails({
+ {entry.outputTokensPerSecond !== undefined ? : null} + {entry.responseHeadersMs !== undefined ? : null} + {entry.timeToFirstSignalMs !== undefined ? : null} + {entry.timeToFirstTextMs !== undefined ? : null} + {entry.upstreamTimeToFirstSignalMs !== undefined ? : null} + {entry.activeOutputMs !== undefined ? : null} + {entry.p95InterEventGapMs !== undefined ? : null} + {entry.maxInterEventGapMs !== undefined ? : null} + {entry.tailMs !== undefined ? : null} + {entry.streamSpeedSampleStatus ? : null} @@ -1483,6 +1496,22 @@ function logCredentialCellLabel(entry: RequestLogEntry): string { return entry.credentialId || entry.credentialChain[0] || (entry.credentialSaturated ? "saturated" : "-"); } +function formatTokenRate(value: number): string { + return `${new Intl.NumberFormat(undefined, { maximumFractionDigits: value < 10 ? 1 : 0 }).format(value)} tok/s`; +} + +function streamSpeedSampleLabel(status: NonNullable): string { + switch (status) { + case "complete": return "Complete speed sample"; + case "partial": return "Partial speed sample"; + case "usage_missing": return "Usage missing speed sample"; + case "insufficient_tokens": return "Insufficient tokens speed sample"; + case "unsupported_protocol": return "Unsupported protocol speed sample"; + case "hidden_reasoning": return "Hidden reasoning speed sample"; + case "batched_output": return "Batched output speed sample"; + } +} + function LogStatusDot({ entry }: { entry: RequestLogEntry }) { return ( diff --git a/packages/ui/src/pages/home/components/settings.tsx b/packages/ui/src/pages/home/components/settings.tsx index c37e5ebe..c0f35e2b 100644 --- a/packages/ui/src/pages/home/components/settings.tsx +++ b/packages/ui/src/pages/home/components/settings.tsx @@ -36,6 +36,7 @@ export function AppSettingsDialog({ onChangeLanguage, onChangeTheme, onChangeTrayIcon, + onChangeTrayShowTokenRate, onChangeTrayWidgets, onClose, observability, @@ -50,6 +51,7 @@ export function AppSettingsDialog({ traySupported, trayBalanceProgress, trayIconPreference, + trayShowTokenRate, trayWidgets, updateConfig }: { @@ -71,6 +73,7 @@ export function AppSettingsDialog({ onChangeLanguage: (value: string) => void; onChangeTheme: (value: string) => void; onChangeTrayIcon: (value: string) => void; + onChangeTrayShowTokenRate: (checked: boolean) => void; onChangeTrayWidgets: (widgets: TrayWidgetConfig[]) => void; onClose: () => void; observability: AppConfig["observability"]; @@ -85,6 +88,7 @@ export function AppSettingsDialog({ traySupported: boolean; trayBalanceProgress?: TrayBalanceProgressConfig; trayIconPreference: AppConfig["trayIcon"]; + trayShowTokenRate: boolean; trayWidgets: TrayWidgetConfig[]; updateConfig: (mutator: (config: AppConfig) => AppConfig) => void; }) { @@ -129,11 +133,14 @@ export function AppSettingsDialog({ copy={copy} onChangeTrayBalanceProgress={onChangeTrayBalanceProgress} onChangeTrayIcon={onChangeTrayIcon} + onChangeTrayShowTokenRate={onChangeTrayShowTokenRate} onChangeTrayWidgets={onChangeTrayWidgets} providerAccountSnapshots={providerAccountSnapshots} trayBalanceProgress={trayBalanceProgress} trayIconPreference={trayIconPreference} + trayShowTokenRate={trayShowTokenRate} trayWidgets={trayWidgets} + tokenRateSupported={appInfo.platform === "darwin"} /> ); } @@ -1902,20 +1909,26 @@ function TraySettingsPage({ copy, onChangeTrayBalanceProgress, onChangeTrayIcon, + onChangeTrayShowTokenRate, onChangeTrayWidgets, providerAccountSnapshots, trayBalanceProgress, trayIconPreference, - trayWidgets + trayShowTokenRate, + trayWidgets, + tokenRateSupported }: { copy: AppCopy; onChangeTrayBalanceProgress: (config: TrayBalanceProgressConfig) => void; onChangeTrayIcon: (value: string) => void; + onChangeTrayShowTokenRate: (checked: boolean) => void; onChangeTrayWidgets: (widgets: TrayWidgetConfig[]) => void; providerAccountSnapshots: ProviderAccountSnapshot[]; trayBalanceProgress?: TrayBalanceProgressConfig; trayIconPreference: AppConfig["trayIcon"]; + trayShowTokenRate: boolean; trayWidgets: TrayWidgetConfig[]; + tokenRateSupported: boolean; }) { const pageRef = useRef(null); const [selectedTrayWidgetId, setSelectedTrayWidgetId] = useState(); @@ -2100,8 +2113,17 @@ function TraySettingsPage({ }, [selectedWidget, selectedWidgetIndex, widgets]); return ( -
+

{copy.settings.tray}

+ {tokenRateSupported ? ( + + ) : null}
diff --git a/packages/ui/src/pages/home/shared/i18n.tsx b/packages/ui/src/pages/home/shared/i18n.tsx index 30bb08b3..28100f55 100644 --- a/packages/ui/src/pages/home/shared/i18n.tsx +++ b/packages/ui/src/pages/home/shared/i18n.tsx @@ -90,6 +90,8 @@ export type AppCopy = { trayPreview: string; trayPreviewEmpty: string; trayProgressTarget: string; + trayTokenRate: string; + trayTokenRateDescription: string; trayWindowModules: string; title: string; }; @@ -204,6 +206,8 @@ export const appCopy: Record = { trayPreview: "Preview", trayPreviewEmpty: "No tray modules enabled", trayProgressTarget: "Progress target", + trayTokenRate: "Live output speed", + trayTokenRateDescription: "Show the estimated combined token rate of all active streams in the macOS menu bar.", trayWindowModules: "Tray window modules", title: "Settings" }, @@ -227,6 +231,16 @@ export const appCopy: Record = { "Stream": "Stream", "Streaming": "Streaming", "Non-streaming": "Non-streaming", + "Output speed": "Output speed", + "Headers ready": "Headers ready", + "First signal": "First signal", + "First text": "First text", + "Upstream first signal": "Upstream first signal", + "Output window": "Output window", + "P95 gap": "P95 gap", + "Max stall": "Max stall", + "Tail wait": "Tail wait", + "Speed sample": "Speed sample", "Token": "Token", "令牌": "Tokens", "成本": "Cost", @@ -820,6 +834,8 @@ export const appCopy: Record = { trayPreview: "预览", trayPreviewEmpty: "未启用 Tray 模块", trayProgressTarget: "进度目标", + trayTokenRate: "实时输出速率", + trayTokenRateDescription: "在 macOS 菜单栏显示所有活跃流叠加后的预估 Token 速率。", trayWindowModules: "Tray 窗口模块", title: "设置" }, @@ -1780,6 +1796,23 @@ export const appCopy: Record = { "Stream": "流式", "Streaming": "流式", "Non-streaming": "非流式", + "Output speed": "输出速率", + "Headers ready": "响应头就绪", + "First signal": "首反馈", + "First text": "首文本", + "Upstream first signal": "上游首反馈", + "Output window": "连续输出窗口", + "P95 gap": "P95 间隔", + "Max stall": "最大卡顿", + "Tail wait": "尾部等待", + "Speed sample": "测速样本", + "Complete speed sample": "有效", + "Partial speed sample": "未完整结束", + "Usage missing speed sample": "缺少 Usage", + "Insufficient tokens speed sample": "Token 不足", + "Unsupported protocol speed sample": "协议不支持", + "Hidden reasoning speed sample": "包含隐藏推理", + "Batched output speed sample": "输出被批量合并", "Started": "开始时间", "Source truncated": "源数据已截断", "User sent": "用户发送", diff --git a/packages/ui/test/unit/config-persistence.test.ts b/packages/ui/test/unit/config-persistence.test.ts index 07842e7a..635e867d 100644 --- a/packages/ui/test/unit/config-persistence.test.ts +++ b/packages/ui/test/unit/config-persistence.test.ts @@ -30,6 +30,15 @@ test("failed provider writes leave the draft unchanged and can be retried once", assert.deepEqual(draft.Providers.map((provider) => provider.name), ["New provider"]); }); +test("live tray token rate is opt-in and survives config reconciliation", () => { + const base = createDefaultAppConfig(); + assert.equal(base.trayShowTokenRate, false); + const draft = { ...base, trayShowTokenRate: true }; + const saved = structuredClone(base); + const next = reconcileSavedConfig(base, draft, saved); + assert.equal(next.trayShowTokenRate, true); +}); + test("API key writes wait for configuration writes and their commit", async () => { const queue = createConfigSaveQueue(); const write = deferred();