feat(db): per-request cost ledger + per-key tpm/rpm/monthly quota (RIC-741)

open-sse/handlers/chatCore.ts

@@ -275,7 +275,7 @@ import { ensureEngineBreakdown } from "../services/compression/engineBreakdown.t
 import { handleBypassRequest } from "../utils/bypassHandler.ts";
 import { saveRequestUsage, trackPendingRequest, appendRequestLog } from "@/lib/usageDb";
 import { finalizePendingScope, updatePendingScope } from "@/lib/usage/pendingRequestScope";
-import { recordCost } from "@/domain/costRules";
+import { recordCost, recordChatCallCost, buildCostCtx } from "@/domain/costRules";
 import { calculateCost } from "@/lib/usage/costCalculator";
 import {
   buildClaudePassthroughToolNameMap,
@@ -309,7 +309,7 @@ import {
 } from "./chatCore/pluginOnResponse.ts";
 import { scheduleStreamingQuotaShareConsumption } from "./chatCore/streamingQuotaShare.ts";
 import { recordStreamingUsageStats } from "./chatCore/streamingUsageStats.ts";
-import { recordStreamingCost } from "./chatCore/streamingCost.ts";
+import { recordStreamingCost, buildStreamLedgerDetails } from "./chatCore/streamingCost.ts";
 import { isJsonRecord } from "./chatCore/nonStreamingResponseParse.ts";
 import { recordNonStreamingUsageStats } from "./chatCore/nonStreamingUsageStats.ts";
 import {
@@ -5447,6 +5447,7 @@ export async function handleChatCore({
       const estimatedCost = costUsage
         ? await calculateCost(provider, model, costUsage, { serviceTier: effectiveServiceTier })
         : 0;
+      const chatCostCtx = buildCostCtx(provider, model, usage, effectiveServiceTier, traceId);
 
       if (postCallGuardrails.blocked) {
         const guardrailMessage = postCallGuardrails.message || "Response blocked by guardrail";
@@ -5467,9 +5468,7 @@ export async function handleChatCore({
           claudeCacheUsageMeta: cacheUsageLogMeta,
           cacheSource: "upstream",
         });
-        if (apiKeyInfo?.id && estimatedCost > 0) {
-          recordCost(apiKeyInfo.id, estimatedCost);
-        }
+        recordChatCallCost(apiKeyInfo, estimatedCost, chatCostCtx, false);
         log?.warn?.(
           "GUARDRAIL",
           `Response blocked by ${postCallGuardrails.guardrail || "guardrail"}: ${guardrailMessage}`
@@ -5607,9 +5606,7 @@ export async function handleChatCore({
         claudeCacheUsageMeta: cacheUsageLogMeta,
         cacheSource: "upstream",
       });
-      if (apiKeyInfo?.id && estimatedCost > 0) {
-        recordCost(apiKeyInfo.id, estimatedCost);
-      }
+      recordChatCallCost(apiKeyInfo, estimatedCost, chatCostCtx, true);
 
       // === Quota Share POST-hook (B/F7) — fire-and-forget, fail-open ===
       await scheduleQuotaShareConsumption({
@@ -6041,6 +6038,7 @@ export async function handleChatCore({
       serviceTier: effectiveServiceTier,
       calculateCost,
       recordCost,
+      ledger: buildStreamLedgerDetails(effectiveServiceTier, normalizedStreamStatus < 400, traceId),
     });
 
     // === Quota Share POST-hook streaming (B/F7) — fire-and-forget, fail-open ===

open-sse/handlers/chatCore/nonStreamingUsageStats.ts

@@ -11,6 +11,7 @@
  */
 
 import { saveRequestUsage } from "@/lib/usageDb";
+import { recordKeyQuotaUsage } from "@/domain/keyQuota";
 import { formatUsageLog } from "@/lib/usage/tokenAccounting";
 import { COLORS } from "../../utils/stream.ts";
 import { recordTokenUsage } from "../../services/tokenLimitCounter.ts";
@@ -71,6 +72,10 @@ function recordBillableTokens(
   if (!apiKeyInfo?.id) return;
   try {
     const billable = computeBillableTokens(usage);
+    // Key-quota tpm/rpm counters advance for every successful call regardless
+    // of token count (rpm always +1), so the gate in apiKeyPolicy sees current
+    // usage. Mirrors recordTokenUsage's swallow-and-continue.
+    recordKeyQuotaUsage(apiKeyInfo.id, billable);
     if (billable > 0)
       recordTokenUsage(apiKeyInfo.id, provider || "unknown", model || "unknown", billable);
   } catch {

open-sse/handlers/chatCore/streamingCost.ts

@@ -16,22 +16,62 @@ type CostResolver = (
   options: { serviceTier?: string }
 ) => Promise<number>;
 
+/** Extra per-request ledger context threaded from the handler. */
+type LedgerDetails = {
+  serviceTier?: string | null;
+  success?: boolean;
+  timestamp?: string;
+  requestId?: string | null;
+};
+
+/** Build a {@link LedgerDetails} for a completed streaming response. */
+export function buildStreamLedgerDetails(
+  serviceTier: string | null | undefined,
+  success: boolean,
+  requestId: string | null | undefined
+): LedgerDetails {
+  return { serviceTier, success, timestamp: new Date().toISOString(), requestId };
+}
+
 export function recordStreamingCost(args: {
   apiKeyId: string | null | undefined;
   provider: string | null | undefined;
   model: string | null | undefined;
   streamUsage: Record<string, number | undefined> | null | undefined;
   serviceTier?: string;
   calculateCost: CostResolver;
-  recordCost: (apiKeyId: string, cost: number) => void;
+  recordCost: (
+    apiKeyId: string,
+    cost: number,
+    details?: {
+      provider?: string | null;
+      model?: string | null;
+      tokens?: unknown;
+      serviceTier?: string | null;
+      success?: boolean;
+      timestamp?: string;
+      requestId?: string | null;
+    }
+  ) => void;
+  ledger?: LedgerDetails;
 }): void {
   if (!args.apiKeyId || !args.streamUsage) return;
 
   const apiKeyId = args.apiKeyId;
   args
     .calculateCost(args.provider, args.model, args.streamUsage, { serviceTier: args.serviceTier })
     .then((estimatedCost) => {
-      if (estimatedCost > 0) args.recordCost(apiKeyId, estimatedCost);
+      if (estimatedCost > 0) {
+        args.recordCost(apiKeyId, estimatedCost, {
+          provider: args.provider,
+          model: args.model,
+          tokens: args.streamUsage,
+          serviceTier: args.serviceTier ?? args.ledger?.serviceTier,
+          success: args.ledger?.success ?? true,
+          timestamp: args.ledger?.timestamp,
+          requestId: args.ledger?.requestId,
+        });
+      }
     })
     .catch(() => {});
 }

open-sse/handlers/chatCore/streamingUsageStats.ts

@@ -11,6 +11,7 @@
  */
 
 import { saveRequestUsage } from "@/lib/usageDb";
+import { recordKeyQuotaUsage } from "@/domain/keyQuota";
 import { recordTokenUsage } from "../../services/tokenLimitCounter.ts";
 import { computeBillableTokens } from "./upstreamTimeouts.ts";
 import { type EffectiveServiceTier } from "./serviceTier.ts";
@@ -57,6 +58,8 @@ function recordStreamingBillableTokens(usage: object, ctx: RecordStreamingUsageS
   if (!ctx.apiKeyInfo?.id || ctx.streamStatus !== 200) return;
   try {
     const billable = computeBillableTokens(usage);
+    // Key-quota tpm/rpm counters advance on every completed stream.
+    recordKeyQuotaUsage(ctx.apiKeyInfo.id, billable);
     if (billable > 0)
       recordTokenUsage(
         ctx.apiKeyInfo.id,

src/app/api/usage/key-quota/route.ts

@@ -0,0 +1,88 @@
+import { NextResponse } from "next/server";
+import { requireManagementAuth } from "@/lib/api/requireManagementAuth";
+import {
+  getKeyQuotaStatus,
+  upsertKeyQuotaLimits,
+  clearKeyQuotaLimits,
+} from "@/lib/db/keyQuota";
+import { setKeyQuotaSchema } from "@/shared/validation/schemas";
+import { isValidationFailure, validateBody } from "@/shared/validation/helpers";
+
+/**
+ * GET /api/usage/key-quota?apiKeyId=... — per-key tpm/rpm/monthly quota status
+ * (limits + current usage + exceeded flags).
+ *
+ * POST /api/usage/key-quota — upsert quota limits for a key. Omitted fields
+ * keep their existing value; 0/null clears a dimension (unlimited).
+ */
+export async function GET(request: Request) {
+  const authError = await requireManagementAuth(request);
+  if (authError) return authError;
+
+  try {
+    const { searchParams } = new URL(request.url);
+    const apiKeyId = searchParams.get("apiKeyId");
+    if (!apiKeyId) {
+      return NextResponse.json({ error: "apiKeyId query param is required" }, { status: 400 });
+    }
+    const status = getKeyQuotaStatus(apiKeyId);
+    return NextResponse.json(status);
+  } catch (error) {
+    console.error("Error fetching key quota:", error);
+    return NextResponse.json({ error: "Failed to fetch key quota" }, { status: 500 });
+  }
+}
+
+export async function POST(request: Request) {
+  const authError = await requireManagementAuth(request);
+  if (authError) return authError;
+
+  let rawBody: unknown;
+  try {
+    rawBody = await request.json();
+  } catch {
+    return NextResponse.json(
+      {
+        error: {
+          message: "Invalid request",
+          details: [{ field: "body", message: "Invalid JSON body" }],
+        },
+      },
+      { status: 400 }
+    );
+  }
+
+  try {
+    const validation = validateBody(setKeyQuotaSchema, rawBody);
+    if (isValidationFailure(validation)) {
+      return NextResponse.json({ error: validation.error }, { status: 400 });
+    }
+    const { apiKeyId, tpmLimit, rpmLimit, monthlyAmountUsd } = validation.data;
+    const limits = upsertKeyQuotaLimits(apiKeyId, { tpmLimit, rpmLimit, monthlyAmountUsd });
+    return NextResponse.json({ success: true, apiKeyId, limits });
+  } catch (error) {
+    console.error("Error setting key quota:", error);
+    return NextResponse.json({ error: "Failed to set key quota" }, { status: 500 });
+  }
+}
+
+/**
+ * DELETE /api/usage/key-quota?apiKeyId=... — remove quota limits + counters.
+ */
+export async function DELETE(request: Request) {
+  const authError = await requireManagementAuth(request);
+  if (authError) return authError;
+
+  try {
+    const { searchParams } = new URL(request.url);
+    const apiKeyId = searchParams.get("apiKeyId");
+    if (!apiKeyId) {
+      return NextResponse.json({ error: "apiKeyId query param is required" }, { status: 400 });
+    }
+    clearKeyQuotaLimits(apiKeyId);
+    return NextResponse.json({ success: true, apiKeyId });
+  } catch (error) {
+    console.error("Error clearing key quota:", error);
+    return NextResponse.json({ error: "Failed to clear key quota" }, { status: 500 });
+  }
+}

src/domain/costRules.ts

@@ -26,6 +26,7 @@ import {
   resetSpendBatchWriterForTests,
   spendBatchWriter,
 } from "@/lib/spend/batchWriter";
+import { recordLedgerFromCost } from "@/lib/usage/costLedgerRecorder";
 
 export type BudgetResetInterval = "daily" | "weekly" | "monthly";
 
@@ -393,17 +394,84 @@ export function deleteBudget(apiKeyId: string) {
 /**
  * Record a cost for an API key.
  *
+ * When `details` is provided, the same call is appended to the per-request cost
+ * ledger (`request_cost_ledger`) with the full breakdown so the amount is
+ * traceable to provider/model/tokens — used by the chat paths. Bare calls
+ * (web-search cost, etc.) write an amount-only ledger row with unknown
+ * provider/model.
+ *
  * @param {string} apiKeyId
  * @param {number} cost - Cost in USD
+ * @param {RecordCostDetails} [details] - Optional per-request breakdown.
  */
-export function recordCost(apiKeyId: string, cost: number): void {
+export function recordCost(apiKeyId: string, cost: number, details?: RecordCostDetails): void {
   try {
     spendBatchWriter.increment(apiKeyId, cost, Date.now());
+    if (details) {
+      // Fire-and-forget — never block the response on ledger I/O.
+      void recordLedgerFromCost({
+        apiKeyId,
+        provider: details.provider,
+        model: details.model,
+        tokens: details.tokens,
+        amountUsd: cost,
+        serviceTier: details.serviceTier,
+        success: details.success,
+        timestamp: details.timestamp,
+        requestId: details.requestId,
+      });
+    }
   } catch {
     // Non-critical.
   }
 }
 
+/**
+ * Optional per-request breakdown passed to {@link recordCost} for ledger
+ * traceability. `tokens` is the raw provider/normalized usage object — the
+ * same shape `saveRequestUsage` accepts.
+ */
+export interface RecordCostDetails {
+  provider?: string | null;
+  model?: string | null;
+  tokens?: unknown;
+  serviceTier?: string | null;
+  success?: boolean;
+  timestamp?: string;
+  requestId?: string | null;
+}
+
+/**
+ * Build the provider/model/tokens/serviceTier/requestId breakdown shared by
+ * both {@link recordChatCallCost} call sites in chatCore.ts (built once,
+ * kept out of that file to stay under its frozen file-size ratchet).
+ */
+export function buildCostCtx(
+  provider: string | null | undefined,
+  model: string | null | undefined,
+  tokens: unknown,
+  serviceTier: string | null | undefined,
+  requestId: string | null | undefined
+): RecordCostDetails {
+  return { provider, model, tokens, serviceTier, requestId };
+}
+
+/**
+ * Record a chat-path cost when there is a real amount to log — the shared
+ * `apiKeyInfo?.id && estimatedCost > 0` guard used at both chatCore.ts
+ * `recordCost` call sites. `context` (see {@link buildCostCtx}) is
+ * built once by the caller; only `success` varies per call site.
+ */
+export function recordChatCallCost(
+  apiKeyInfo: { id?: string | null } | null | undefined,
+  estimatedCost: number,
+  context: RecordCostDetails,
+  success: boolean
+): void {
+  if (!apiKeyInfo?.id || estimatedCost <= 0) return;
+  recordCost(apiKeyInfo.id, estimatedCost, { ...context, success });
+}
+
 /**
  * Sync all budgets against the current clock so overdue resets get persisted.
  */