Add live token rate tracking and stream metrics

2026-11-02 23:21:47 +08:00 · 2026-09-14 10:45:53 +08:00
parent afba89fa37
commit 774bfb85b7
23 changed files with 1458 additions and 32 deletions
@@ -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;
@@ -875,6 +875,10 @@ function pickConfig(value: Partial<AppConfig>): LoadedAppConfig {
  if (trayIcon) {
    config.trayIcon = trayIcon;
  }
+  const trayShowTokenRate = (value as Record<string, unknown>).trayShowTokenRate;
+  if (typeof trayShowTokenRate === "boolean") {
+    config.trayShowTokenRate = trayShowTokenRate;
+  }
  const trayBalanceProgress = parseTrayBalanceProgress((value as Record<string, unknown>).trayBalanceProgress);
  if (trayBalanceProgress) {
    config.trayBalanceProgress = trayBalanceProgress;
@@ -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,
@@ -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<string, string | string[]>;
@@ -2226,7 +2257,12 @@ export type RequestLogEntry = {
  responseModel?: string;
  responseHeaders: Record<string, string | string[]>;
  statusCode: number;
+  streamSpeedSampleStatus?: StreamSpeedSampleStatus;
+  tailMs?: number;
+  timeToFirstSignalMs?: number;
+  timeToFirstTextMs?: number;
  totalTokens: number;
+  upstreamTimeToFirstSignalMs?: number;
  url: string;
 };

@@ -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<string, string | string[] | undefined>;
 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<RouteScriptValidationResult> {
@@ -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<void> {
@@ -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<string, unknown>;
+  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,
@@ -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";
@@ -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<GatewayProviderConfig, "provider" | "type">;
  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<typeof Readable.fromWeb>[0]
+  );
+  const metered = source.pipe(meter.stream);
+  return new Response(Readable.toWeb(metered) as ReadableStream<Uint8Array>, {
+    headers: new Headers(streamInput.upstreamResponse.headers),
+    status: streamInput.upstreamResponse.status,
+    statusText: streamInput.upstreamResponse.statusText
+  });
+}
+
 async function handleRuntimeConfigControlRoute(
  config: AppConfig,
  request: GatewayPluginHttpRequest,
@@ -206,6 +206,9 @@ export type UpstreamFetchResult = {
  attempt: UpstreamAttempt;
  failedAttempts: UpstreamFailedAttempt[];
  response: Response;
+  timing: {
+    attemptStartedAtMonoMs: number;
+  };
 };


@@ -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<ContextArchiveReplayResult> {
@@ -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);
@@ -1086,7 +1086,8 @@ function plainHeaderRecord(value: Headers | Record<string, string | string[] | u

 function estimateRecordBytes(input: RequestLogRecordInput): number {
  return input.requestBody.byteLength + Buffer.byteLength(input.responseBodyText ?? "") +
-    jsonBytes(input.requestHeaders) + jsonBytes(input.responseHeaders) + jsonBytes(input.routeTrace) + 1_024;
+    jsonBytes(input.requestHeaders) + jsonBytes(input.responseHeaders) + jsonBytes(input.routeTrace) +
+    jsonBytes(input.streamMetrics) + 1_024;
 }

 function estimateRawTraceBytes(
@@ -64,6 +64,8 @@ import type {
  RequestRouteTrace,
  RequestRouteTraceHop,
  RequestRouteTraceSnapshot,
+  RequestStreamMetrics,
+  StreamSpeedSampleStatus,
  UsageStatsRange
 } from "@ccr/core/contracts/app";

@@ -136,6 +138,7 @@ export type RequestLogRecordInput = {
  requestId?: string;
  resolvedModel?: string;
  routeTrace?: RequestRouteTrace;
+  streamMetrics?: RequestStreamMetrics;
  responseBodyText?: string;
  responseBodySizeBytes?: number;
  responseBodyTruncated?: boolean;
@@ -217,6 +220,7 @@ export type RequestLogStoreWriteResult = {
 };

 type StoredRequestLogEntry = {
+  activeOutputMs?: number;
  cacheReadTokens: number;
  cacheWriteTokens: number;
  client: string;
@@ -231,13 +235,17 @@ type StoredRequestLogEntry = {
  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<string, string | string[]>;
@@ -252,7 +260,12 @@ type StoredRequestLogEntry = {
  responseModel: string;
  responseHeaders: Record<string, string | string[]>;
  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<string, SqlValue>): 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<string, SqlValue>): 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<string, SqlValue>): 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<string, SqlValue>): 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 extends keyof RequestStreamMetrics>(
+  key: Key,
+  value: unknown
+): Partial<Pick<RequestStreamMetrics, Key>> {
+  const number = asNumber(value);
+  return number === undefined ? {} : { [key]: number } as Partial<Pick<RequestStreamMetrics, Key>>;
+}
+
+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<string, SqlValue>, prefix: "request" | "response"): RequestLogBody | undefined {
  const text = String(row[`${prefix}_body_text`] ?? "");
  const sizeBytes = normalizeCount(row[`${prefix}_body_size_bytes`]);
@@ -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<string, ActiveTokenStream>();
+
+  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<string, LiveTokenRateSnapshot>();
+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<string, unknown>): 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<string, unknown>): SemanticFragment[] {
+  const fragments: Array<SemanticFragment | undefined> = [];
+  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<string, unknown>, 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<string, unknown>): SemanticFragment[] {
+  const fragments: Array<SemanticFragment | undefined> = [];
+  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 | undefined>): 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<string, unknown> {
+  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;
+}
@@ -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
+    }
  });
 });

@@ -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;
@@ -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";
@@ -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);
+  });
+}
@@ -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`;
+}
@@ -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
@@ -853,6 +853,9 @@ function LogMobileCard({
            <div className="mt-2 grid grid-cols-2 gap-2 text-[11px]">
              <LogCompactMetric label={t("Token")} value={tokenSummary} />
              <LogCompactMetric label={t("持续时间")} value={formatDuration(item.durationMs)} />
+              {item.outputTokensPerSecond !== undefined ? (
+                <LogCompactMetric label={t("Output speed")} value={formatTokenRate(item.outputTokensPerSecond)} />
+              ) : null}
              {hasCredentialInfo ? <LogCompactMetric label={t("Credential")} value={logCredentialCellLabel(item)} /> : null}
              <LogCompactMetric label={t("Provider")} value={item.provider || "-"} />
            </div>
@@ -969,6 +972,16 @@ export function LogExpandedDetails({
      <div className={cn("network-body-meta grid grid-cols-2 gap-y-2 border-b px-3 py-2 text-[12px] sm:grid-cols-4", hasCredentialInfo ? "lg:grid-cols-12" : "lg:grid-cols-9")}>
        <LogMetric label={t("持续时间")} value={formatDuration(entry.durationMs)} />
        <LogMetric label={t("Stream")} value={entry.isStream ? t("Streaming") : t("Non-streaming")} />
+        {entry.outputTokensPerSecond !== undefined ? <LogMetric label={t("Output speed")} value={formatTokenRate(entry.outputTokensPerSecond)} /> : null}
+        {entry.responseHeadersMs !== undefined ? <LogMetric label={t("Headers ready")} value={formatDuration(entry.responseHeadersMs)} /> : null}
+        {entry.timeToFirstSignalMs !== undefined ? <LogMetric label={t("First signal")} value={formatDuration(entry.timeToFirstSignalMs)} /> : null}
+        {entry.timeToFirstTextMs !== undefined ? <LogMetric label={t("First text")} value={formatDuration(entry.timeToFirstTextMs)} /> : null}
+        {entry.upstreamTimeToFirstSignalMs !== undefined ? <LogMetric label={t("Upstream first signal")} value={formatDuration(entry.upstreamTimeToFirstSignalMs)} /> : null}
+        {entry.activeOutputMs !== undefined ? <LogMetric label={t("Output window")} value={formatDuration(entry.activeOutputMs)} /> : null}
+        {entry.p95InterEventGapMs !== undefined ? <LogMetric label={t("P95 gap")} value={formatDuration(entry.p95InterEventGapMs)} /> : null}
+        {entry.maxInterEventGapMs !== undefined ? <LogMetric label={t("Max stall")} value={formatDuration(entry.maxInterEventGapMs)} /> : null}
+        {entry.tailMs !== undefined ? <LogMetric label={t("Tail wait")} value={formatDuration(entry.tailMs)} /> : null}
+        {entry.streamSpeedSampleStatus ? <LogMetric label={t("Speed sample")} value={t(streamSpeedSampleLabel(entry.streamSpeedSampleStatus))} /> : null}
        <LogMetric label={t("Request ID")} value={entry.requestId || "-"} />
        <LogMetric label={t("Client")} value={entry.client || "-"} />
        <LogMetric label={t("Provider")} value={entry.provider || "-"} />
@@ -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<RequestLogEntry["streamSpeedSampleStatus"]>): 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 (
    <span className={cn("h-3 w-3 shrink-0 rounded-full", entry.ok ? "network-dot-completed" : "network-dot-error")} />
@@ -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<HTMLDivElement>(null);
  const [selectedTrayWidgetId, setSelectedTrayWidgetId] = useState<string>();
@@ -2100,8 +2113,17 @@ function TraySettingsPage({
  }, [selectedWidget, selectedWidgetIndex, widgets]);

  return (
-    <div className={cn(settingsPageContentWidthClassName, "grid min-h-[520px] grid-rows-[auto_auto_auto] gap-4")} ref={pageRef}>
+    <div className={cn(settingsPageContentWidthClassName, "grid min-h-[520px] grid-rows-[auto_auto_auto_auto] gap-4")} ref={pageRef}>
      <h3 className="text-[15px] font-semibold text-foreground">{copy.settings.tray}</h3>
+      {tokenRateSupported ? (
+        <SettingsSwitchRow
+          checked={trayShowTokenRate}
+          description={copy.settings.trayTokenRateDescription}
+          icon={Gauge}
+          label={copy.settings.trayTokenRate}
+          onChange={onChangeTrayShowTokenRate}
+        />
+      ) : null}
      <div className="flex flex-wrap items-end gap-3 rounded-md border border-border bg-background p-3">
        <Field className="min-w-[220px] flex-1" label={copy.settings.trayIcon}>
          <TrayIconSelect onChange={changeTrayIcon} options={trayIconOptions} progress={progressPreviewValue} value={effectiveTrayIconPreference} />
@@ -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<ResolvedLanguage, AppCopy> = {
      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<ResolvedLanguage, AppCopy> = {
      "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<ResolvedLanguage, AppCopy> = {
      trayPreview: "预览",
      trayPreviewEmpty: "未启用 Tray 模块",
      trayProgressTarget: "进度目标",
+      trayTokenRate: "实时输出速率",
+      trayTokenRateDescription: "在 macOS 菜单栏显示所有活跃流叠加后的预估 Token 速率。",
      trayWindowModules: "Tray 窗口模块",
      title: "设置"
    },
@@ -1780,6 +1796,23 @@ export const appCopy: Record<ResolvedLanguage, AppCopy> = {
      "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": "用户发送",
@@ -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<number>();