Preserve Copilot quota snapshots for router dashboard

Keep router usage totals populated when Copilot surfaces quota data through streaming terminal events or websocket Responses completions instead of HTTP headers. Preserve last-known-good snapshot fields so later empty headers do not erase valid dashboard data.

Constraint: Dashboard must show practical used/total quota for all routed instances after live SSE traffic.
Rejected: UI-only placeholder handling | It would mask missing upstream quota propagation instead of fixing the data path.
Confidence: high
Scope-risk: moderate
Directive: Keep quota snapshot parsing centralized in router/lib.ts and do not clear good fields with empty upstream responses.
Tested: bun run lint:all --fix; bun run build; bun test; bun run typecheck; live 4140 requests verified 6/6 dashboard instances have u/t values.
Not-tested: none

Author: cuipengfei

router/lib.ts

@@ -206,6 +206,93 @@ export function parseUpstreamHeaderSnapshot(
   }
 }
 
+interface CopilotQuotaSnapshotFields {
+  entitlement: number
+  overage: number
+  remainingPercent: number
+  resetAt: string
+}
+
+const parseCopilotQuotaSnapshot = (
+  value: unknown,
+): CopilotQuotaSnapshotFields | null => {
+  if (!isRecord(value)) {
+    return null
+  }
+
+  const entitlement =
+    typeof value.entitlement === "string" ?
+      Number(value.entitlement)
+    : Number.NaN
+  const remainingPercent =
+    typeof value.percent_remaining === "number" ?
+      value.percent_remaining
+    : Number.NaN
+  const overage =
+    typeof value.overage_count === "number" ? value.overage_count : Number.NaN
+  const resetAt = typeof value.reset_date === "string" ? value.reset_date : null
+
+  if (
+    !Number.isFinite(entitlement)
+    || !Number.isFinite(remainingPercent)
+    || !Number.isFinite(overage)
+    || entitlement < 0
+    || remainingPercent < 0
+    || remainingPercent > 100
+    || overage < 0
+    || !resetAt
+    || Number.isNaN(Date.parse(resetAt))
+  ) {
+    return null
+  }
+
+  return { entitlement, overage, remainingPercent, resetAt }
+}
+
+export function parseUpstreamQuotaSnapshots(
+  value: unknown,
+): UpstreamHeaderSnapshot {
+  const snapshot: UpstreamHeaderSnapshot = {
+    premiumUsage: null,
+    sessionRateLimit: null,
+    weeklyRateLimit: null,
+  }
+
+  if (!isRecord(value)) {
+    return snapshot
+  }
+
+  const premium = parseCopilotQuotaSnapshot(value.premium_interactions)
+  if (premium) {
+    const used =
+      premium.overage > 0 ?
+        premium.entitlement + premium.overage
+      : premium.entitlement
+        - (premium.entitlement * premium.remainingPercent) / 100
+    if (Number.isFinite(used) && used >= 0) {
+      snapshot.premiumUsage = { used, total: premium.entitlement }
+    }
+  }
+
+  const session = parseCopilotQuotaSnapshot(value["5Hour-Session-RateLimits"])
+  if (session) {
+    snapshot.sessionRateLimit = {
+      remaining: session.remainingPercent,
+      resetAt: session.resetAt,
+    }
+  }
+
+  const weekly = parseCopilotQuotaSnapshot(value["Weekly-Session-RateLimits"])
+  if (weekly) {
+    snapshot.weeklyRateLimit = {
+      remaining: weekly.remainingPercent,
+      resetAt: weekly.resetAt,
+    }
+  }
+
+  return snapshot
+}
+
 export function getBindingKey(
   sessionId: string | null,
   agent: string,

router/state.ts

@@ -6,6 +6,7 @@ import {
   getHeaderValue,
   isRecord,
   parseUpstreamHeaderSnapshot,
+  parseUpstreamQuotaSnapshots,
   parseModelFromBody,
   parseModelIds,
   parseModelObjects,
@@ -89,6 +90,7 @@ export interface ProxyToOptions {
   context: ProxyContext
   logger: (line: string) => void
   fetchImpl?: typeof fetch
+  onQuotaSnapshots?: (quotaSnapshots: unknown) => void
 }
 
 export interface DashboardHandlerOptions {
@@ -128,12 +130,52 @@ const EMPTY_UPSTREAM_HEADER_SNAPSHOT: UpstreamHeaderSnapshot = {
   weeklyRateLimit: null,
 }
 
+export function mergeUpstreamHeaderSnapshot(
+  previous: UpstreamHeaderSnapshot | undefined,
+  next: UpstreamHeaderSnapshot,
+): UpstreamHeaderSnapshot {
+  if (!previous) {
+    return next
+  }
+
+  return {
+    premiumUsage: next.premiumUsage ?? previous.premiumUsage,
+    sessionRateLimit: next.sessionRateLimit ?? previous.sessionRateLimit,
+    weeklyRateLimit: next.weeklyRateLimit ?? previous.weeklyRateLimit,
+  }
+}
+
 export function updateUpstreamHeaderSnapshot(
   state: StickyRouterState,
   port: number,
   headers: Headers,
 ) {
-  state.portHeaderSnapshots.set(port, parseUpstreamHeaderSnapshot(headers))
+  const next = parseUpstreamHeaderSnapshot(headers)
+  updateUpstreamSnapshot(state, port, next)
+}
+
+export function updateUpstreamSnapshot(
+  state: StickyRouterState,
+  port: number,
+  next: UpstreamHeaderSnapshot,
+) {
+  const previous = state.portHeaderSnapshots.get(port)
+  state.portHeaderSnapshots.set(
+    port,
+    mergeUpstreamHeaderSnapshot(previous, next),
+  )
+}
+
+export function updateUpstreamQuotaSnapshot(
+  state: StickyRouterState,
+  port: number,
+  quotaSnapshots: unknown,
+) {
+  updateUpstreamSnapshot(
+    state,
+    port,
+    parseUpstreamQuotaSnapshots(quotaSnapshots),
+  )
 }
 
 export function getRemainingCooldownMs(
@@ -494,6 +536,69 @@ export function getInstanceName(
   return state.portToInstance.get(port)?.name || `:${port}`
 }
 
+function observeResponsesSseQuotaSnapshots(
+  body: ReadableStream<Uint8Array> | null,
+  onQuotaSnapshots?: (quotaSnapshots: unknown) => void,
+): ReadableStream<Uint8Array> | null {
+  if (!body || !onQuotaSnapshots) {
+    return body
+  }
+
+  const decoder = new TextDecoder()
+  let buffered = ""
+
+  const inspectEvent = (eventText: string) => {
+    const data = eventText
+      .split(/\r?\n/u)
+      .filter((line) => line.startsWith("data:"))
+      .map((line) => line.slice("data:".length).trimStart())
+      .join("\n")
+
+    if (!data || data === "[DONE]") {
+      return
+    }
+
+    try {
+      const parsed = JSON.parse(data) as { copilot_quota_snapshots?: unknown }
+      if (parsed.copilot_quota_snapshots) {
+        onQuotaSnapshots(parsed.copilot_quota_snapshots)
+      }
+    } catch {
+      return
+    }
+  }
+
+  const inspectText = (text: string, flush = false) => {
+    buffered += text
+
+    for (;;) {
+      const separator = buffered.search(/\r?\n\r?\n/u)
+      if (separator === -1) break
+      const eventText = buffered.slice(0, separator)
+      const separatorLength = buffered[separator] === "\r" ? 4 : 2
+      buffered = buffered.slice(separator + separatorLength)
+      inspectEvent(eventText)
+    }
+
+    if (flush && buffered.trim()) {
+      inspectEvent(buffered)
+      buffered = ""
+    }
+  }
+
+  return body.pipeThrough(
+    new TransformStream<Uint8Array, Uint8Array>({
+      transform(chunk, controller) {
+        inspectText(decoder.decode(chunk, { stream: true }))
+        controller.enqueue(chunk)
+      },
+      flush() {
+        inspectText(decoder.decode(), true)
+      },
+    }),
+  )
+}
+
 export async function proxyTo(options: ProxyToOptions): Promise<Response> {
   const { port, context, logger } = options
   const fetchImpl = options.fetchImpl ?? fetch
@@ -509,7 +614,15 @@ export async function proxyTo(options: ProxyToOptions): Promise<Response> {
       body: req.method !== "GET" && req.method !== "HEAD" ? body : undefined,
     })
 
-    return new Response(upstream.body, {
+    const responseBody =
+      upstream.headers.get("content-type")?.includes("text/event-stream") ?
+        observeResponsesSseQuotaSnapshots(
+          upstream.body,
+          options.onQuotaSnapshots,
+        )
+      : upstream.body
+
+    return new Response(responseBody, {
       status: upstream.status,
       headers: upstream.headers,
     })
@@ -709,6 +822,8 @@ async function handleNoModelRequest(
     context: { body: request.bodyText, req: request.req, url: request.url },
     logger: runtime.logger,
     fetchImpl: runtime.fetchImpl,
+    onQuotaSnapshots: (quotaSnapshots) =>
+      updateUpstreamQuotaSnapshot(runtime.state, port, quotaSnapshots),
   })
   applyCooldownOn429(runtime, proxied, {
     port,
@@ -777,6 +892,8 @@ async function handleModelRequest(
     context: { body: request.bodyText, req: request.req, url: request.url },
     logger: runtime.logger,
     fetchImpl: runtime.fetchImpl,
+    onQuotaSnapshots: (quotaSnapshots) =>
+      updateUpstreamQuotaSnapshot(runtime.state, result.port, quotaSnapshots),
   })
   applyCooldownOn429(runtime, proxied, {
     port: result.port,

src/routes/messages/anthropic-types.ts

@@ -206,6 +206,7 @@ export interface AnthropicMessageDeltaEvent {
     stop_reason?: AnthropicResponse["stop_reason"]
     stop_sequence?: string | null
   }
+  copilot_quota_snapshots?: unknown
   usage?: {
     input_tokens?: number
     output_tokens: number

src/routes/messages/responses-stream-translation.ts

@@ -505,17 +505,20 @@ const handleResponseCompleted = (
     hasToolCall: state.hasToolCall,
     toolSearchName: state.toolSearchName,
   })
-  events.push(
-    {
-      type: "message_delta",
-      delta: {
-        stop_reason: anthropic.stop_reason,
-        stop_sequence: anthropic.stop_sequence,
-      },
-      usage: anthropic.usage,
+  const messageDelta: AnthropicStreamEventData = {
+    type: "message_delta",
+    delta: {
+      stop_reason: anthropic.stop_reason,
+      stop_sequence: anthropic.stop_sequence,
     },
-    { type: "message_stop" },
-  )
+    usage: anthropic.usage,
+  }
+
+  if ("copilot_quota_snapshots" in rawEvent) {
+    messageDelta.copilot_quota_snapshots = rawEvent.copilot_quota_snapshots
+  }
+
+  events.push(messageDelta, { type: "message_stop" })
   state.messageCompleted = true
   return events
 }