Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 35 additions & 27 deletions api/src/gateway/router.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@ import { getSettings, defaultQuotas } from '../settings/service.ts'
import { streamedToolCallsBroken, contextBudget, OPENAI_COMPATIBLE_PROVIDER_NAME } from '../models/operations.ts'
import { resolveRoleModel, resolveRoleEntry, type ResolvedRoleModel } from '../models/service.ts'
import { recordUsage } from '../usage/service.ts'
import { computeCreditBreakdown } from '../usage/operations.ts'
import { resolveUsageIdentity, enforceQuotas } from '../usage/enforce.ts'
import { computeCreditBreakdown, quotaErrorBody } from '../usage/operations.ts'
import { resolveUsageIdentity, enforceQuotas, getSelfUsage } from '../usage/enforce.ts'
import { convertOpenAITools, convertOpenAIMessages, convertToolChoice, mapFinishReason, supportsMediaToolResults, injectMediaAsUserMessages } from './operations.ts'
import type { OpenAIMessage, OpenAIToolDefinition, OpenAIToolChoice, FinishReason } from './operations.ts'
import { recordTraceRequest } from '../traces/service.ts'
Expand All @@ -21,7 +21,8 @@ import createDebug from 'debug'
// provider reports them (Anthropic cache read/write, OpenAI cached_tokens, etc.).
// The AI SDK normalizes these into usage.inputTokenDetails regardless of provider,
// so the gateway can forward them uniformly for the debug trace.
function buildUsage (usage: LanguageModelUsage | undefined) {
// `cost` is the credits billed for the call (moderation included), always present.
function buildUsage (usage: LanguageModelUsage | undefined, cost: number) {
if (!usage) return undefined
const promptTokens = usage.inputTokens ?? 0
const completionTokens = usage.outputTokens ?? 0
Expand All @@ -31,11 +32,14 @@ function buildUsage (usage: LanguageModelUsage | undefined) {
prompt_tokens: number
completion_tokens: number
total_tokens: number
// credits billed for this call (OpenRouter's `usage.cost` convention), moderation included
cost: number
prompt_tokens_details?: { cached_tokens: number, cache_creation_tokens: number }
} = {
prompt_tokens: promptTokens,
completion_tokens: completionTokens,
total_tokens: promptTokens + completionTokens
total_tokens: promptTokens + completionTokens,
cost
}
if (cacheRead != null || cacheWrite != null) {
result.prompt_tokens_details = { cached_tokens: cacheRead ?? 0, cache_creation_tokens: cacheWrite ?? 0 }
Expand All @@ -53,6 +57,21 @@ function isValidModelId (id: string): id is ModelId {
return MODEL_IDS.includes(id as ModelId)
}

// The caller's own consumption: same identity resolution (and 401/403) as a
// completion, so it reports exactly the usage that the gateway enforces.
router.get('/:type/:id/usage', async (req, res, next) => {
try {
const sessionState = reqSession(req)
const owner = req.params as unknown as AccountKeys
const settings = await getSettings(owner)
const quotas = settings.quotas ?? defaultQuotas
const identity = await resolveUsageIdentity(req, owner, quotas, sessionState, isAuthenticated(sessionState))
res.json(await getSelfUsage(owner, quotas, identity))
} catch (err) {
next(err)
}
})

// OpenAI-compatible chat completions endpoint
router.post('/:type/:id/v1/chat/completions', async (req, res, next) => {
try {
Expand Down Expand Up @@ -161,16 +180,7 @@ router.post('/:type/:id/v1/chat/completions', async (req, res, next) => {

const quotaCheck = await enforceQuotas(owner, quotas, identity)
if (quotaCheck) {
res.status(429).json({
error: {
message: quotaCheck.reason,
type: 'rate_limit_error',
scope: quotaCheck.scope,
usage: quotaCheck.usage,
limit: quotaCheck.limit,
resets_at: quotaCheck.resetsAt
}
})
res.status(429).json({ error: quotaErrorBody(quotaCheck, identity.role === 'admin') })
return
}

Expand Down Expand Up @@ -286,8 +296,6 @@ router.post('/:type/:id/v1/chat/completions', async (req, res, next) => {
// avoids TypeScript's (here incorrect) control-flow narrowing of the let.
const gateBlocked = () => gateState === 'blocked'
const buffered: string[] = []
// a finish trace captured while the gate was still pending: only stored if the gate opens
let deferredFinishTrace: (() => void) | null = null
const sseWrite = (payload: string) => {
if (gateState === 'blocked' || res.writableEnded) return
if (gateState === 'pending') buffered.push(payload)
Expand Down Expand Up @@ -317,8 +325,6 @@ router.post('/:type/:id/v1/chat/completions', async (req, res, next) => {
gateState = 'open'
for (const payload of buffered) res.write(payload)
buffered.length = 0
deferredFinishTrace?.()
deferredFinishTrace = null
}
})
moderation.onLateBlock(() => {
Expand Down Expand Up @@ -391,14 +397,16 @@ router.post('/:type/:id/v1/chat/completions', async (req, res, next) => {
}
})
}
sseWrite(`data: ${JSON.stringify({ id: completionId, object: 'chat.completion.chunk', created, model: modelId, choices: [{ index: 0, delta: {}, finish_reason: mapFinishReason(gen.finishReason as FinishReason) }], usage: buildUsage(gen.usage) })}\n\n`)
if (moderation) await moderation.gate
sseWrite(`data: ${JSON.stringify({ id: completionId, object: 'chat.completion.chunk', created, model: modelId, choices: [{ index: 0, delta: {}, finish_reason: mapFinishReason(gen.finishReason as FinishReason) }], usage: buildUsage(gen.usage, credits.total + (moderation?.cost() ?? 0)) })}\n\n`)
const recordFinishTrace = () => recordTrace(
{ content: streamedText, toolCalls: [...streamedToolCalls.values()], finishReason: mapFinishReason(gen.finishReason as FinishReason) },
{ inputTokens, outputTokens, cacheReadTokens: details?.cacheReadTokens, cacheWriteTokens: details?.cacheWriteTokens },
ttfc
)
if (gateState === 'pending') deferredFinishTrace = recordFinishTrace
else if (gateState === 'open') recordFinishTrace()
// the gate is already settled here (awaited above): a block verdict has
// recorded its own content-free content_filter trace instead
if (gateState === 'open') recordFinishTrace()
} else {
for await (const part of result!.fullStream) {
if (part.type === 'error') {
Expand Down Expand Up @@ -493,24 +501,24 @@ router.post('/:type/:id/v1/chat/completions', async (req, res, next) => {
})
}

if (moderation) await moderation.gate
sseWrite(`data: ${JSON.stringify({
id: completionId,
object: 'chat.completion.chunk',
created,
model: modelId,
choices: [{ index: 0, delta: {}, finish_reason: mapFinishReason(part.finishReason as FinishReason) }],
usage: buildUsage(part.totalUsage)
usage: buildUsage(part.totalUsage, credits.total + (moderation?.cost() ?? 0))
})}\n\n`)

const recordFinishTrace = () => recordTrace(
{ content: streamedText, toolCalls: [...streamedToolCalls.values()], finishReason: mapFinishReason(part.finishReason as FinishReason) },
{ inputTokens, outputTokens, cacheReadTokens: details?.cacheReadTokens, cacheWriteTokens: details?.cacheWriteTokens },
ttfc
)
// While the gate is pending the content must not reach trace storage:
// a block verdict records its own content-free content_filter trace.
if (gateState === 'pending') deferredFinishTrace = recordFinishTrace
else if (gateState === 'open') recordFinishTrace()
// the gate is already settled here (awaited above): a block verdict has
// recorded its own content-free content_filter trace instead
if (gateState === 'open') recordFinishTrace()
}
}
}
Expand Down Expand Up @@ -648,7 +656,7 @@ router.post('/:type/:id/v1/chat/completions', async (req, res, next) => {
message: responseMessage,
finish_reason: mapFinishReason(result.finishReason as FinishReason)
}],
usage: buildUsage(result.usage)
usage: buildUsage(result.usage, credits.total + (moderation?.cost() ?? 0))
})

recordTrace(
Expand Down
6 changes: 5 additions & 1 deletion api/src/moderation/service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,8 @@ export interface ModerationRun {
onLateBlock: (cb: () => void) => void
// best-known verdict info for trace embedding (undefined until the check settles)
traceInfo: () => TraceModeration | undefined
// credits of the classifier call once it settled (0 before, or when it failed)
cost: () => number
}

export function startModeration (params: {
Expand All @@ -133,6 +135,7 @@ export function startModeration (params: {
let lateBlockCb: (() => void) | undefined
let timedOut = false
let trace: TraceModeration | undefined
let cost = 0

// Exactly one event per check, written when the check settles.
const finalize = (action: ModerationEventAction, verdict?: ModerationVerdict, opts?: { failOpen?: 'timeout' | 'error' }) => {
Expand Down Expand Up @@ -185,6 +188,7 @@ export function startModeration (params: {
entry,
config.eurosPerCredit
)
cost = credits.total
if (credits.total > 0) {
await recordUsage(owner, {
cost: credits.total,
Expand Down Expand Up @@ -236,7 +240,7 @@ export function startModeration (params: {
})
])

return { gate, onLateBlock: (cb) => { lateBlockCb = cb }, traceInfo: () => trace }
return { gate, onLateBlock: (cb) => { lateBlockCb = cb }, traceInfo: () => trace, cost: () => cost }
}

// ---- admin probe ----
Expand Down
53 changes: 38 additions & 15 deletions api/src/usage/enforce.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,9 @@ import { assertCanUseModel, assertRoleQuota, getEffectiveRole, type EffectiveRol
import { assertAnonymousActionToken } from '../anonymous-token/service.ts'
import { getUsage, getMonthlyResetsAt } from './service.ts'
import { getCreditInfo } from '../limits/service.ts'
import { firstQuotaViolation, isUntrustedRole, type QuotaCheckInput, type QuotaExceeded } from './operations.ts'
import { firstQuotaViolation, isUntrustedRole, quotaWindows, type QuotaExceeded, type SelfUsage } from './operations.ts'

export type { SelfUsage } from './operations.ts'

type Quotas = NonNullable<Settings['quotas']>

Expand Down Expand Up @@ -82,16 +84,11 @@ export async function resolveUsageIdentity (req: Request, owner: AccountKeys, qu
}

/**
* Enforce the org-wide credit cap, then the untrusted-pool and per-user
* quotas, in that order. Returns the first violation (for a 429 response) or
* null when within all limits.
*
* The account-wide cap is no longer a quota: it is enforced from the
* account's credits limit (customers-pushed, or the configured default) and
* checked first, short-circuiting even when the caller's own profile quota
* is unlimited.
* The shared-budget checks: the org-wide credit cap, then (untrusted callers
* only) the anonymous+external pool. First of enforceQuotas()'s checks, and
* what the self-usage view reports as the account status.
*/
export async function enforceQuotas (owner: AccountKeys, quotas: Quotas, identity: UsageIdentity): Promise<QuotaExceeded | null> {
export async function accountViolation (owner: AccountKeys, quotas: Quotas, identity: UsageIdentity): Promise<QuotaExceeded | null> {
// org-wide cap from the limits API (customers-pushed, or the configured default)
const { limit, consumption } = await getCreditInfo(owner)
if (limit >= 0 && consumption >= limit) {
Expand All @@ -106,25 +103,51 @@ export async function enforceQuotas (owner: AccountKeys, quotas: Quotas, identit
}
}

const checks: (QuotaCheckInput | null)[] = []

// combined anonymous + external pool — caps untrusted traffic as a group
if (identity.isUntrusted) {
const poolLimits = quotas.untrusted
if (poolLimits && !poolLimits.unlimited && poolLimits.monthlyLimit) {
const usage = await getUsage(owner, UNTRUSTED_POOL_ID)
checks.push({ usage, limits: poolLimits, scope: 'untrusted' })
return firstQuotaViolation([{ usage, limits: poolLimits, scope: 'untrusted' }])
}
}
return null
}

/**
* Enforce the shared budgets (see accountViolation), then the per-user (or
* per-IP) quota. Returns the first violation (for a 429 response) or null.
*/
export async function enforceQuotas (owner: AccountKeys, quotas: Quotas, identity: UsageIdentity): Promise<QuotaExceeded | null> {
const shared = await accountViolation(owner, quotas, identity)
if (shared) return shared

// per-user (or per-IP) role cap
if (identity.trackPerUser) {
const roleLimits = quotas[identity.role]
if (roleLimits && !roleLimits.unlimited && roleLimits.monthlyLimit) {
const usage = await getUsage(owner, identity.usageUserId)
checks.push({ usage, limits: roleLimits, scope: 'user' })
return firstQuotaViolation([{ usage, limits: roleLimits, scope: 'user' }])
}
}
return null
}

return firstQuotaViolation(checks)
/**
* What a caller may know about their own consumption: their own quota windows,
* and the shared budgets as a status — with numbers only for an admin of the owner.
*/
export async function getSelfUsage (owner: AccountKeys, quotas: Quotas, identity: UsageIdentity): Promise<SelfUsage> {
const isAdmin = identity.role === 'admin'
const [usage, violation, credits] = await Promise.all([
// untracked callers (owner of a user account) consume the account aggregate
getUsage(owner, identity.trackPerUser ? identity.usageUserId : undefined),
accountViolation(owner, quotas, identity),
isAdmin ? getCreditInfo(owner) : Promise.resolve(null)
])
// a per-profile quota is only enforced for per-user tracked callers
const quota = quotaWindows(usage, identity.trackPerUser ? quotas[identity.role] : undefined)
const account: SelfUsage['account'] = violation ? { status: 'exhausted', resetsAt: violation.resetsAt } : { status: 'ok' }
if (credits && credits.limit >= 0) Object.assign(account, { used: credits.consumption, limit: credits.limit })
return { role: identity.role, quota, account }
}
105 changes: 68 additions & 37 deletions api/src/usage/operations.ts
Original file line number Diff line number Diff line change
Expand Up @@ -29,45 +29,76 @@ export interface QuotaExceeded {
resetsAt: string
}

export function checkQuota (usage: UsageInfo, limits: UsageLimits, scope: string): QuotaExceeded | null {
if (limits.unlimited) return null
const monthlyLimit = limits.monthlyLimit
if (!monthlyLimit || monthlyLimit <= 0) return null

const weeklyLimit = monthlyLimit / 2
const dailyLimit = monthlyLimit / 4

if (usage.daily.cost >= dailyLimit) {
return {
allowed: false,
reason: 'Daily cost quota exceeded',
scope,
period: 'daily',
usage: usage.daily.cost,
limit: dailyLimit,
resetsAt: usage.daily.resetsAt
}
export interface QuotaWindow {
used: number
// absent when the quota is not enforced (unlimited, or monthlyLimit <= 0)
limit?: number
resetsAt: string
}

export interface SelfQuota {
unlimited: boolean
daily: QuotaWindow
weekly: QuotaWindow
monthly: QuotaWindow
}

export interface SelfUsage {
role: 'admin' | 'contrib' | 'user' | 'external' | 'anonymous'
quota: SelfQuota
account: { status: 'ok' | 'exhausted', resetsAt?: string, used?: number, limit?: number }
}

/**
* The three enforced windows of a quota, derived from its monthly limit
* (weekly = monthly / 2, daily = monthly / 4). The single source of these
* ratios: checkQuota() enforces exactly what this reports.
*/
export function quotaWindows (usage: UsageInfo, limits: UsageLimits | undefined): SelfQuota {
const monthlyLimit = limits?.monthlyLimit ?? 0
const unlimited = !!limits?.unlimited || monthlyLimit <= 0
const window = (period: UsagePeriodInfo, limit: number): QuotaWindow =>
unlimited ? { used: period.cost, resetsAt: period.resetsAt } : { used: period.cost, limit, resetsAt: period.resetsAt }
return {
unlimited,
daily: window(usage.daily, monthlyLimit / 4),
weekly: window(usage.weekly, monthlyLimit / 2),
monthly: window(usage.monthly, monthlyLimit)
}
if (usage.weekly.cost >= weeklyLimit) {
return {
allowed: false,
reason: 'Weekly cost quota exceeded',
scope,
period: 'weekly',
usage: usage.weekly.cost,
limit: weeklyLimit,
resetsAt: usage.weekly.resetsAt
}
}

const PERIOD_REASONS = {
daily: 'Daily cost quota exceeded',
weekly: 'Weekly cost quota exceeded',
monthly: 'Monthly cost quota exceeded'
} as const

// Scopes whose numbers are the account's, not the caller's own.
const SHARED_SCOPES = new Set(['account', 'untrusted'])

/**
* The `error` object of a 429. `usage`/`limit` of a shared budget (the org credit
* cap, the anonymous+external pool) are only disclosed to an admin of the owner.
*/
export function quotaErrorBody (violation: QuotaExceeded, isAdmin: boolean) {
const disclose = isAdmin || !SHARED_SCOPES.has(violation.scope)
return {
message: violation.reason,
type: 'rate_limit_error' as const,
scope: violation.scope,
period: violation.period,
resets_at: violation.resetsAt,
...(disclose ? { usage: violation.usage, limit: violation.limit } : {})
}
if (usage.monthly.cost >= monthlyLimit) {
return {
allowed: false,
reason: 'Monthly cost quota exceeded',
scope,
period: 'monthly',
usage: usage.monthly.cost,
limit: monthlyLimit,
resetsAt: usage.monthly.resetsAt
}

export function checkQuota (usage: UsageInfo, limits: UsageLimits, scope: string): QuotaExceeded | null {
const windows = quotaWindows(usage, limits)
if (windows.unlimited) return null
for (const period of ['daily', 'weekly', 'monthly'] as const) {
const w = windows[period]
if (w.used >= w.limit!) {
return { allowed: false, reason: PERIOD_REASONS[period], scope, period, usage: w.used, limit: w.limit!, resetsAt: w.resetsAt }
}
}
return null
Expand Down
Loading
Loading