feat: refactor rate limit logging and implement provider responses handler

Author: caozhiyuan

src/lib/codex-rate-limit.ts

@@ -56,15 +56,14 @@ export const logCodexRateLimitsEvent = (event: unknown): void => {
     }
     summary.push(
       `used=${window.used_percent}%`,
-      `window=${window.window_minutes}m`,
       `reset_at=${formatCodexRateLimitResetAt(window.reset_at)}`,
     )
 
     const label =
       planType ?
         `Codex ${scope} rate limit (${planType})`
       : `Codex ${scope} rate limit`
-    consola.info(`${label}: ${summary.join(", ")}`)
+    consola.log(`${label}: ${summary.join(", ")}`)
   }
 }
 

src/lib/copilot-rate-limit.ts

@@ -136,7 +136,7 @@ export const logCopilotQuotaSnapshots = (
 const logCopilotRateLimitUsage = (usage: CopilotRateLimitUsage): void => {
   const d = new Date(usage.resetAt)
   const dateStr = Number.isNaN(d.getTime()) ? usage.resetAt : d.toLocaleString()
-  consola.info(
+  consola.log(
     `Copilot ${usage.type} quota remaining: ${usage.remaining}, resets at: ${dateStr}`,
   )
 }

src/routes/provider/responses/handler.ts

@@ -0,0 +1,193 @@
+import type { Context } from "hono"
+
+import { events } from "fetch-event-stream"
+
+import { HTTPError } from "~/lib/error"
+import { createHandlerLogger, debugJson } from "~/lib/logger"
+import { resolveProviderConfig } from "~/lib/provider-resolver"
+import { requestContext } from "~/lib/request-context"
+import {
+  createProviderTokenUsageRecorder,
+  normalizeResponsesUsage,
+  type UsageTokens,
+} from "~/lib/token-usage"
+import { applyResponsesApiContextManagement } from "~/routes/responses/utils"
+import type {
+  ResponsesPayload,
+  ResponsesResult,
+  ResponseStreamEvent,
+} from "~/services/copilot/create-responses"
+import {
+  createStandardizedCodexResponsesEventStream,
+  forwardCodexResponses,
+  normalizeCodexResponsesEvent,
+} from "~/services/codex/create-responses"
+import { getModels as getCodexModels } from "~/services/codex/get-models"
+import {
+  createProviderProxyResponse,
+  forwardProviderResponses,
+} from "~/services/providers/provider-proxy"
+
+const logger = createHandlerLogger("provider-responses-handler")
+
+export async function handleProviderResponsesForProvider(
+  c: Context,
+  options: {
+    payload: ResponsesPayload
+    provider: string
+  },
+): Promise<Response> {
+  const { payload, provider } = options
+  const providerConfig = await resolveProviderConfig(provider)
+  if (providerConfig?.type !== "openai-responses") {
+    return c.json(
+      {
+        error: {
+          message: `Provider '${provider}' does not support the /v1/responses endpoint`,
+          type: "invalid_request_error",
+        },
+      },
+      400,
+    )
+  }
+
+  const selectedModel =
+    providerConfig.name === "codex" ?
+      getCodexModels().data.find((model) => model.id === payload.model)
+    : undefined
+
+  applyResponsesApiContextManagement(
+    payload,
+    selectedModel?.capabilities.limits.max_prompt_tokens,
+  )
+
+  const upstreamResponse =
+    providerConfig.name === "codex" ?
+      await forwardCodexResponses(
+        payload,
+        c.req.raw.headers,
+        providerConfig.baseUrl,
+      )
+    : await forwardProviderResponses(providerConfig, payload, c.req.raw.headers)
+
+  if (!upstreamResponse.ok) {
+    throw new HTTPError(
+      `Failed to create ${provider} responses`,
+      upstreamResponse,
+    )
+  }
+
+  const recordUsage = createProviderResponsesUsageRecorder(payload, provider)
+
+  if (payload.stream) {
+    void recordProviderResponsesStreamUsage(upstreamResponse.clone(), {
+      normalizeCodex: providerConfig.name === "codex",
+      provider,
+      recordUsage,
+    }).catch((error) => {
+      logger.warn("provider.responses.usage_stream_error", {
+        provider,
+        error: getErrorMessage(error),
+      })
+    })
+  } else {
+    const responseBody = (await upstreamResponse
+      .clone()
+      .json()) as ResponsesResult
+    recordUsage(normalizeResponsesUsage(responseBody.usage))
+  }
+
+  if (providerConfig.name === "codex" && payload.stream) {
+    return createProviderProxyResponse(
+      upstreamResponse,
+      createStandardizedCodexResponsesEventStream(
+        getResponsesEvents(upstreamResponse),
+      ),
+    )
+  }
+
+  return createProviderProxyResponse(upstreamResponse)
+}
+
+const getErrorMessage = (error: unknown): string => {
+  if (error instanceof Error && error.message) {
+    return error.message
+  }
+
+  return String(error)
+}
+
+const createProviderResponsesUsageRecorder = (
+  payload: ResponsesPayload,
+  provider: string,
+): ((usage: UsageTokens) => void) => {
+  const sessionAffinity =
+    requestContext.getStore()?.sessionAffinity?.trim() || null
+
+  return createProviderTokenUsageRecorder({
+    endpoint: "responses",
+    model: payload.model,
+    providerName: provider,
+    sessionId: sessionAffinity ?? "",
+  })
+}
+
+const recordProviderResponsesStreamUsage = async (
+  upstreamResponse: unknown,
+  options: {
+    normalizeCodex: boolean
+    provider: string
+    recordUsage: (usage: UsageTokens) => void
+  },
+): Promise<void> => {
+  let usage: UsageTokens = {}
+
+  try {
+    for await (const chunk of getResponsesEvents(upstreamResponse)) {
+      debugJson(logger, "Responses stream chunk:", chunk)
+      if (!chunk.data || chunk.data === "[DONE]") {
+        continue
+      }
+
+      const parsed = parseProviderResponsesStreamEvent(chunk.data, {
+        normalizeCodex: options.normalizeCodex,
+        provider: options.provider,
+      })
+      if (
+        parsed?.type === "response.completed"
+        || parsed?.type === "response.failed"
+        || parsed?.type === "response.incomplete"
+      ) {
+        usage = normalizeResponsesUsage(parsed.response.usage)
+      }
+    }
+  } finally {
+    options.recordUsage(usage)
+  }
+}
+
+const parseProviderResponsesStreamEvent = (
+  data: string,
+  options: {
+    normalizeCodex: boolean
+    provider: string
+  },
+): ResponseStreamEvent | null => {
+  try {
+    const parsed = JSON.parse(data) as ResponseStreamEvent
+    return options.normalizeCodex ?
+        normalizeCodexResponsesEvent(parsed)
+      : parsed
+  } catch (error) {
+    logger.error("provider.responses.parse_chunk_error", {
+      provider: options.provider,
+      data,
+      error,
+    })
+    return null
+  }
+}
+
+const getResponsesEvents = (response: unknown) => {
+  return events(response as Parameters<typeof events>[0])
+}