diff --git a/apps/api/app/routes/admin.py b/apps/api/app/routes/admin.py index cf2eadd..f53c98a 100644 --- a/apps/api/app/routes/admin.py +++ b/apps/api/app/routes/admin.py @@ -5,7 +5,7 @@ from __future__ import annotations import time from datetime import datetime, timezone from decimal import Decimal -from typing import Annotated, Literal, cast +from typing import Annotated, Iterable, Literal, cast from uuid import UUID from fastapi import APIRouter, Depends, HTTPException, Query, status @@ -50,11 +50,14 @@ router = APIRouter(prefix="/admin", tags=["admin"]) AdminPrincipal = Annotated[Principal, Depends(require_admin_access())] HealthStatus = Literal["ok", "degraded", "down"] -UsageBudgetStatus = Literal["disabled", "ok", "warn", "exceeded"] +UsageBudgetStatus = Literal["disabled", "ok", "warn", "exceeded", "indeterminate"] UsageCostBasis = Literal[ "provider_estimate", "provider_reported", "reference_rate", + "reference_upper_bound", + "partial", + "partial_upper_bound", "unavailable", ] TicketCategory = Literal[ @@ -69,8 +72,21 @@ TicketPriority = Literal["low", "normal", "high", "urgent"] TicketStatus = Literal["open", "triaged", "in_progress", "resolved", "closed"] NotificationDeliveryStatus = Literal["queued", "sending", "sent", "failed", "skipped"] -METERED_CLIENT_TURN_FILTER_SQL = """ +SYNTHETIC_USAGE_SIGNATURES = frozenset({("e2e", "fake-client", 1, 1, 0.0)}) + +REPORTABLE_CLIENT_TURN_FILTER_SQL = """ speaker = 'client' + AND NOT ( + LOWER(BTRIM(COALESCE(llm_provider, ''))) = 'e2e' + AND LOWER(BTRIM(COALESCE(model, ''))) = 'fake-client' + AND COALESCE(tokens_in, 0) = 1 + AND COALESCE(tokens_out, 0) = 1 + AND COALESCE(cost_usd, 0) = 0 + ) +""" + +METERED_CLIENT_TURN_FILTER_SQL = f""" + {REPORTABLE_CLIENT_TURN_FILTER_SQL} AND ( llm_provider IS NOT NULL OR model IS NOT NULL OR tokens_in IS NOT NULL OR tokens_out IS NOT NULL @@ -78,28 +94,6 @@ METERED_CLIENT_TURN_FILTER_SQL = """ ) """ -USAGE_AGGREGATE_COLUMNS_SQL = """ - COUNT(*) AS turns, - COUNT(*) FILTER ( - WHERE COALESCE(tokens_in, 0) > 0 OR COALESCE(tokens_out, 0) > 0 - ) AS token_metered_turns, - COUNT(*) FILTER ( - WHERE COALESCE(tokens_in, 0) <= 0 AND COALESCE(tokens_out, 0) <= 0 - ) AS token_unmetered_turns, - COALESCE(SUM(tokens_in), 0)::bigint AS tokens_in, - COALESCE(SUM(tokens_out), 0)::bigint AS tokens_out, - COALESCE(SUM(cost_usd), 0)::numeric AS cost_usd, - COUNT(*) FILTER (WHERE COALESCE(cost_usd, 0) <= 0) AS unpriced_turns, - COALESCE( - SUM(tokens_in) FILTER (WHERE COALESCE(cost_usd, 0) <= 0), - 0 - )::bigint AS unpriced_tokens_in, - COALESCE( - SUM(tokens_out) FILTER (WHERE COALESCE(cost_usd, 0) <= 0), - 0 - )::bigint AS unpriced_tokens_out -""" - SUPPORT_TICKET_DETAIL_FROM_SQL = """ SELECT t.id, @@ -187,6 +181,7 @@ class AdminUsageDailyCost(BaseModel): tokens_in: int tokens_out: int cost_usd: float + cost_basis: UsageCostBasis = "provider_reported" class AdminUsageBudget(BaseModel): @@ -194,6 +189,7 @@ class AdminUsageBudget(BaseModel): used_ratio: float remaining_usd: float | None status: UsageBudgetStatus + cost_basis: UsageCostBasis = "provider_reported" class AdminUsageEvaluatorCache(BaseModel): @@ -221,6 +217,7 @@ class AdminUsageResponse(BaseModel): cost_usd: float recorded_cost_usd: float = 0.0 estimated_cost_usd: float = 0.0 + cost_basis: UsageCostBasis = "provider_reported" budget: AdminUsageBudget evaluator_cache: AdminUsageEvaluatorCache by_provider: list[AdminUsageBreakdown] @@ -458,6 +455,23 @@ def _safe_usage_int(value: object) -> int: return 0 +def _is_reportable_usage( + provider: str, + model: str, + tokens_in: int, + tokens_out: int, + cost_usd: float, +) -> bool: + signature = ( + provider.strip().lower(), + model.strip().lower(), + max(0, tokens_in), + max(0, tokens_out), + max(0.0, cost_usd), + ) + return signature not in SYNTHETIC_USAGE_SIGNATURES + + def _usage_breakdown( *, provider: str, @@ -470,31 +484,41 @@ def _usage_breakdown( stored_cost_usd: float, unpriced_tokens_in: int, unpriced_tokens_out: int, + priced_at: datetime | int | float | str, ) -> AdminUsageBreakdown: """저장된 공급자 비용 추정치와 공식 참조단가를 한 원장 행으로 정규화한다.""" + reference_basis = provider_uses_reference_cost(provider) fallback = estimate_reference_cost( provider=provider, model=model, tokens_in=unpriced_tokens_in, tokens_out=unpriced_tokens_out, + priced_at=priced_at, ) - rate_info = fallback or estimate_reference_cost( - provider=provider, - model=model, - tokens_in=tokens_in, - tokens_out=tokens_out, - ) + rate_info = fallback + if rate_info is None and not (reference_basis and stored_cost_usd > 0): + rate_info = estimate_reference_cost( + provider=provider, + model=model, + tokens_in=tokens_in, + tokens_out=tokens_out, + priced_at=priced_at, + ) fallback_cost = fallback.cost_usd if fallback is not None else 0.0 effective_cost = max(0.0, stored_cost_usd) + fallback_cost - reference_basis = provider_uses_reference_cost(provider) if reference_basis: recorded_cost = 0.0 estimated_cost = effective_cost - basis: UsageCostBasis = ( - "reference_rate" if rate_info is not None or effective_cost > 0 else "unavailable" - ) + if fallback is not None and stored_cost_usd <= 0: + basis: UsageCostBasis = "reference_upper_bound" + else: + basis = ( + "reference_rate" + if rate_info is not None or effective_cost > 0 + else "unavailable" + ) else: recorded_cost = max(0.0, stored_cost_usd) estimated_cost = fallback_cost @@ -505,6 +529,17 @@ def _usage_breakdown( else: basis = "unavailable" + rate_label = rate_info.rate_label if rate_info is not None else None + rate_source_url = rate_info.source_url if rate_info is not None else None + if reference_basis and stored_cost_usd > 0: + if fallback is not None: + rate_label = "호출 시점 저장 추정값 + 미저장분 공식 참조단가 합산" + else: + rate_label = "호출 시점에 저장된 참조단가 추정값" + rate_source_url = None + elif fallback is not None: + rate_label = f"{fallback.rate_label} · 캐시 미보존 과거행은 전체 입력 기준" + return AdminUsageBreakdown( provider=provider, model=model, @@ -513,16 +548,120 @@ def _usage_breakdown( token_unmetered_turns=max(0, token_unmetered_turns), tokens_in=max(0, tokens_in), tokens_out=max(0, tokens_out), - cost_usd=round(effective_cost, 6), - recorded_cost_usd=round(recorded_cost, 6), - estimated_cost_usd=round(estimated_cost, 6), + cost_usd=effective_cost, + recorded_cost_usd=recorded_cost, + estimated_cost_usd=estimated_cost, cost_basis=basis, - rate_label=rate_info.rate_label if rate_info is not None else None, - rate_source_url=rate_info.source_url if rate_info is not None else None, + rate_label=rate_label, + rate_source_url=rate_source_url, ) -def _usage_budget(cost_usd: float) -> AdminUsageBudget: +def _merge_usage_breakdowns( + breakdowns: Iterable[AdminUsageBreakdown], +) -> list[AdminUsageBreakdown]: + grouped: dict[tuple[str, str], list[AdminUsageBreakdown]] = {} + for item in breakdowns: + grouped.setdefault((item.provider, item.model), []).append(item) + + merged: list[AdminUsageBreakdown] = [] + for (provider, model), items in grouped.items(): + basis = _aggregate_cost_basis(item.cost_basis for item in items) + labels = {item.rate_label for item in items if item.rate_label} + source_urls = {item.rate_source_url for item in items if item.rate_source_url} + rate_label = next(iter(labels)) if len(labels) == 1 else None + if basis == "partial": + rate_label = "일부 호출 미산정 · 표시액은 산정 가능분 합계" + elif basis == "partial_upper_bound": + rate_label = "일부 호출 미산정 · 산정된 부분도 상한 추정" + elif len(labels) > 1: + rate_label = ( + "기간별 공식 참조단가 상한 합산" + if basis == "reference_upper_bound" + else "기간별 공식 참조단가 합산" + ) + merged.append( + AdminUsageBreakdown( + provider=provider, + model=model, + turns=sum(item.turns for item in items), + token_metered_turns=sum(item.token_metered_turns for item in items), + token_unmetered_turns=sum(item.token_unmetered_turns for item in items), + tokens_in=sum(item.tokens_in for item in items), + tokens_out=sum(item.tokens_out for item in items), + cost_usd=round(sum(item.cost_usd for item in items), 6), + recorded_cost_usd=round( + sum(item.recorded_cost_usd for item in items), 6 + ), + estimated_cost_usd=round( + sum(item.estimated_cost_usd for item in items), 6 + ), + cost_basis=cast(UsageCostBasis, basis), + rate_label=rate_label, + rate_source_url=( + next(iter(source_urls)) if len(source_urls) == 1 else None + ), + ) + ) + return merged + + +def _aggregate_cost_basis( + bases: Iterable[UsageCostBasis], +) -> UsageCostBasis: + basis_set = set(bases) + if not basis_set: + return "provider_reported" + + has_missing = bool( + basis_set & {"unavailable", "partial", "partial_upper_bound"} + ) + has_upper_bound = bool( + basis_set & {"reference_upper_bound", "partial_upper_bound"} + ) + has_known_amount = basis_set != {"unavailable"} + if has_missing: + if not has_known_amount: + return "unavailable" + return "partial_upper_bound" if has_upper_bound else "partial" + if has_upper_bound: + return "reference_upper_bound" + for candidate in ( + "provider_estimate", + "provider_reported", + "reference_rate", + ): + if candidate in basis_set: + return cast(UsageCostBasis, candidate) + return "unavailable" + + +def _scale_usage_breakdown( + breakdown: AdminUsageBreakdown, + multiplier: int, +) -> AdminUsageBreakdown: + count = max(0, multiplier) + return AdminUsageBreakdown( + provider=breakdown.provider, + model=breakdown.model, + turns=breakdown.turns * count, + token_metered_turns=breakdown.token_metered_turns * count, + token_unmetered_turns=breakdown.token_unmetered_turns * count, + tokens_in=breakdown.tokens_in * count, + tokens_out=breakdown.tokens_out * count, + cost_usd=breakdown.cost_usd * count, + recorded_cost_usd=breakdown.recorded_cost_usd * count, + estimated_cost_usd=breakdown.estimated_cost_usd * count, + cost_basis=breakdown.cost_basis, + rate_label=breakdown.rate_label, + rate_source_url=breakdown.rate_source_url, + ) + + +def _usage_budget( + cost_usd: float, + cost_basis: UsageCostBasis, +) -> AdminUsageBudget: limit = max(0.0, float(settings.admin_usage_budget_usd or 0.0)) if limit <= 0: return AdminUsageBudget( @@ -530,18 +669,35 @@ def _usage_budget(cost_usd: float) -> AdminUsageBudget: used_ratio=0.0, remaining_usd=None, status="disabled", + cost_basis=cost_basis, ) used_ratio = max(0.0, cost_usd / limit) - status_value: UsageBudgetStatus = "ok" - if used_ratio >= 1.0: + remaining_usd: float | None = round(max(0.0, limit - cost_usd), 6) + status_value: UsageBudgetStatus + if cost_basis in {"partial_upper_bound", "unavailable"}: + status_value = "indeterminate" + remaining_usd = None + elif cost_basis == "partial": + if used_ratio >= 1.0: + status_value = "exceeded" + elif used_ratio >= 0.8: + status_value = "warn" + else: + status_value = "indeterminate" + elif cost_basis == "reference_upper_bound": + status_value = "ok" if used_ratio < 0.8 else "indeterminate" + elif used_ratio >= 1.0: status_value = "exceeded" elif used_ratio >= 0.8: status_value = "warn" + else: + status_value = "ok" return AdminUsageBudget( limit_usd=round(limit, 6), used_ratio=round(used_ratio, 4), - remaining_usd=round(max(0.0, limit - cost_usd), 6), + remaining_usd=remaining_usd, status=status_value, + cost_basis=cost_basis, ) @@ -628,7 +784,7 @@ async def _usage_from_database(window_days: int) -> AdminUsageResponse: total_row = await conn.fetchrow( f""" SELECT - COUNT(*) FILTER (WHERE speaker = 'client') AS total_turns, + COUNT(*) FILTER (WHERE {REPORTABLE_CLIENT_TURN_FILTER_SQL}) AS total_turns, COUNT(*) FILTER (WHERE {METERED_CLIENT_TURN_FILTER_SQL}) AS metered_turns, COUNT(*) FILTER ( WHERE {METERED_CLIENT_TURN_FILTER_SQL} @@ -639,58 +795,69 @@ async def _usage_from_database(window_days: int) -> AdminUsageResponse: AND COALESCE(tokens_in, 0) <= 0 AND COALESCE(tokens_out, 0) <= 0 ) AS token_unmetered_turns, - COALESCE(SUM(tokens_in) FILTER (WHERE speaker = 'client'), 0)::bigint AS tokens_in, - COALESCE(SUM(tokens_out) FILTER (WHERE speaker = 'client'), 0)::bigint AS tokens_out, - COALESCE(SUM(cost_usd) FILTER (WHERE speaker = 'client'), 0)::numeric AS cost_usd + COALESCE( + SUM(tokens_in) FILTER (WHERE {REPORTABLE_CLIENT_TURN_FILTER_SQL}), 0 + )::bigint AS tokens_in, + COALESCE( + SUM(tokens_out) FILTER (WHERE {REPORTABLE_CLIENT_TURN_FILTER_SQL}), 0 + )::bigint AS tokens_out, + COALESCE( + SUM(cost_usd) FILTER (WHERE {REPORTABLE_CLIENT_TURN_FILTER_SQL}), 0 + )::numeric AS cost_usd FROM app.turns WHERE created_at >= now() - ($1::int * interval '1 day') + AND speaker = 'client' """, window_days, ) - rows = await conn.fetch( + usage_rows = await conn.fetch( f""" SELECT + to_char( + date_trunc('day', created_at AT TIME ZONE 'UTC'), + 'YYYY-MM-DD' + ) AS day, COALESCE(llm_provider, 'unknown') AS provider, COALESCE(model, 'unknown') AS model, - {USAGE_AGGREGATE_COLUMNS_SQL} + COALESCE(tokens_in, 0)::bigint AS tokens_in, + COALESCE(tokens_out, 0)::bigint AS tokens_out, + COALESCE(cost_usd, 0)::numeric AS cost_usd, + COUNT(*)::bigint AS matching_turns FROM app.turns WHERE created_at >= now() - ($1::int * interval '1 day') AND {METERED_CLIENT_TURN_FILTER_SQL} - GROUP BY 1, 2 - """, - window_days, - ) - daily_rows = await conn.fetch( - f""" - SELECT - to_char(date_trunc('day', created_at), 'YYYY-MM-DD') AS day, - COALESCE(llm_provider, 'unknown') AS provider, - COALESCE(model, 'unknown') AS model, - {USAGE_AGGREGATE_COLUMNS_SQL} - FROM app.turns - WHERE created_at >= now() - ($1::int * interval '1 day') - AND {METERED_CLIENT_TURN_FILTER_SQL} - GROUP BY 1, 2, 3 - ORDER BY 1, 2, 3 + GROUP BY 1, 2, 3, 4, 5, 6 """, window_days, ) - all_breakdowns = [ - _usage_breakdown( - provider=str(row["provider"] or "unknown"), - model=str(row["model"] or "unknown"), - turns=_safe_usage_int(row["turns"]), - token_metered_turns=_safe_usage_int(row["token_metered_turns"]), - token_unmetered_turns=_safe_usage_int(row["token_unmetered_turns"]), - tokens_in=_safe_usage_int(row["tokens_in"]), - tokens_out=_safe_usage_int(row["tokens_out"]), - stored_cost_usd=_decimal_to_float(row["cost_usd"]), - unpriced_tokens_in=_safe_usage_int(row["unpriced_tokens_in"]), - unpriced_tokens_out=_safe_usage_int(row["unpriced_tokens_out"]), + # 요청별 입력 크기로 단가 구간이 갈리는 모델이 있어 동일 계량 signature만 묶는다. + dated_breakdowns: list[AdminUsageBreakdown] = [] + for row in usage_rows: + tokens_in = _safe_usage_int(row["tokens_in"]) + tokens_out = _safe_usage_int(row["tokens_out"]) + stored_cost_usd = _decimal_to_float(row["cost_usd"]) + is_token_metered = tokens_in > 0 or tokens_out > 0 + matching_turns = max(1, _safe_usage_int(row["matching_turns"])) + dated_breakdowns.append( + _scale_usage_breakdown( + _usage_breakdown( + provider=str(row["provider"] or "unknown"), + model=str(row["model"] or "unknown"), + turns=1, + token_metered_turns=1 if is_token_metered else 0, + token_unmetered_turns=0 if is_token_metered else 1, + tokens_in=tokens_in, + tokens_out=tokens_out, + stored_cost_usd=stored_cost_usd, + unpriced_tokens_in=tokens_in if stored_cost_usd <= 0 else 0, + unpriced_tokens_out=tokens_out if stored_cost_usd <= 0 else 0, + priced_at=str(row["day"]), + ), + matching_turns, + ) ) - for row in rows - ] + all_breakdowns = _merge_usage_breakdowns(dated_breakdowns) all_breakdowns.sort( key=lambda item: ( -item.cost_usd, @@ -700,33 +867,32 @@ async def _usage_from_database(window_days: int) -> AdminUsageResponse: item.model, ) ) - recorded_cost = round(sum(item.recorded_cost_usd for item in all_breakdowns), 6) - estimated_cost = round(sum(item.estimated_cost_usd for item in all_breakdowns), 6) + recorded_cost = round(sum(item.recorded_cost_usd for item in dated_breakdowns), 6) + estimated_cost = round(sum(item.estimated_cost_usd for item in dated_breakdowns), 6) total_cost = round(recorded_cost + estimated_cost, 6) + total_cost_basis = _aggregate_cost_basis( + item.cost_basis for item in dated_breakdowns + ) - daily_buckets: dict[str, dict[str, int | float]] = {} - for row in daily_rows: - breakdown = _usage_breakdown( - provider=str(row["provider"] or "unknown"), - model=str(row["model"] or "unknown"), - turns=_safe_usage_int(row["turns"]), - token_metered_turns=_safe_usage_int(row["token_metered_turns"]), - token_unmetered_turns=_safe_usage_int(row["token_unmetered_turns"]), - tokens_in=_safe_usage_int(row["tokens_in"]), - tokens_out=_safe_usage_int(row["tokens_out"]), - stored_cost_usd=_decimal_to_float(row["cost_usd"]), - unpriced_tokens_in=_safe_usage_int(row["unpriced_tokens_in"]), - unpriced_tokens_out=_safe_usage_int(row["unpriced_tokens_out"]), - ) + daily_buckets: dict[str, dict[str, int | float | list[UsageCostBasis]]] = {} + for row, breakdown in zip(usage_rows, dated_breakdowns, strict=True): day = str(row["day"]) bucket = daily_buckets.setdefault( day, - {"turns": 0, "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0}, + { + "turns": 0, + "tokens_in": 0, + "tokens_out": 0, + "cost_usd": 0.0, + "cost_bases": [], + }, ) bucket["turns"] = int(bucket["turns"]) + breakdown.turns bucket["tokens_in"] = int(bucket["tokens_in"]) + breakdown.tokens_in bucket["tokens_out"] = int(bucket["tokens_out"]) + breakdown.tokens_out bucket["cost_usd"] = float(bucket["cost_usd"]) + breakdown.cost_usd + cost_bases = cast(list[UsageCostBasis], bucket["cost_bases"]) + cost_bases.append(breakdown.cost_basis) return AdminUsageResponse( source="database", @@ -746,7 +912,8 @@ async def _usage_from_database(window_days: int) -> AdminUsageResponse: cost_usd=total_cost, recorded_cost_usd=recorded_cost, estimated_cost_usd=estimated_cost, - budget=_usage_budget(total_cost), + cost_basis=total_cost_basis, + budget=_usage_budget(total_cost, total_cost_basis), evaluator_cache=_usage_evaluator_cache(), by_provider=all_breakdowns[:12], daily_cost=[ @@ -756,6 +923,9 @@ async def _usage_from_database(window_days: int) -> AdminUsageResponse: tokens_in=int(values["tokens_in"]), tokens_out=int(values["tokens_out"]), cost_usd=round(float(values["cost_usd"]), 6), + cost_basis=_aggregate_cost_basis( + cast(list[UsageCostBasis], values["cost_bases"]) + ), ) for day, values in sorted(daily_buckets.items()) ], @@ -1038,8 +1208,8 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse: cost_usd = 0.0 recorded_cost_usd = 0.0 estimated_cost_usd = 0.0 - buckets: dict[tuple[str, str], dict[str, int | float]] = {} - daily_buckets: dict[str, dict[str, int | float]] = {} + turn_breakdowns: list[AdminUsageBreakdown] = [] + daily_buckets: dict[str, dict[str, int | float | list[UsageCostBasis]]] = {} for sess in store.list(): for turn in getattr(sess, "turns", []) or []: @@ -1048,12 +1218,20 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse: created_at = float(getattr(turn, "created_at", 0.0) or 0.0) if created_at < window_start: continue - total_turns += 1 provider = str(getattr(turn, "llm_provider", None) or "unknown") model = str(getattr(turn, "model", None) or "unknown") turn_tokens_in = _safe_usage_int(getattr(turn, "tokens_in", 0)) turn_tokens_out = _safe_usage_int(getattr(turn, "tokens_out", 0)) turn_cost = _decimal_to_float(getattr(turn, "cost_usd", 0.0)) + if not _is_reportable_usage( + provider, + model, + turn_tokens_in, + turn_tokens_out, + turn_cost, + ): + continue + total_turns += 1 is_metered = ( provider != "unknown" or model != "unknown" @@ -1082,43 +1260,22 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse: stored_cost_usd=turn_cost, unpriced_tokens_in=turn_tokens_in if turn_cost <= 0 else 0, unpriced_tokens_out=turn_tokens_out if turn_cost <= 0 else 0, + priced_at=created_at, ) + turn_breakdowns.append(turn_breakdown) cost_usd += turn_breakdown.cost_usd recorded_cost_usd += turn_breakdown.recorded_cost_usd estimated_cost_usd += turn_breakdown.estimated_cost_usd - key = (provider, model) - bucket = buckets.setdefault( - key, - { - "turns": 0, - "token_metered_turns": 0, - "token_unmetered_turns": 0, - "tokens_in": 0, - "tokens_out": 0, - "stored_cost_usd": 0.0, - "unpriced_tokens_in": 0, - "unpriced_tokens_out": 0, - }, - ) - bucket["turns"] = int(bucket["turns"]) + 1 - if is_token_metered: - bucket["token_metered_turns"] = int(bucket["token_metered_turns"]) + 1 - else: - bucket["token_unmetered_turns"] = int(bucket["token_unmetered_turns"]) + 1 - bucket["tokens_in"] = int(bucket["tokens_in"]) + turn_tokens_in - bucket["tokens_out"] = int(bucket["tokens_out"]) + turn_tokens_out - bucket["stored_cost_usd"] = float(bucket["stored_cost_usd"]) + turn_cost - if turn_cost <= 0: - bucket["unpriced_tokens_in"] = ( - int(bucket["unpriced_tokens_in"]) + turn_tokens_in - ) - bucket["unpriced_tokens_out"] = ( - int(bucket["unpriced_tokens_out"]) + turn_tokens_out - ) day = datetime.fromtimestamp(created_at, timezone.utc).strftime("%Y-%m-%d") daily_bucket = daily_buckets.setdefault( day, - {"turns": 0, "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0}, + { + "turns": 0, + "tokens_in": 0, + "tokens_out": 0, + "cost_usd": 0.0, + "cost_bases": [], + }, ) daily_bucket["turns"] = int(daily_bucket["turns"]) + 1 daily_bucket["tokens_in"] = int(daily_bucket["tokens_in"]) + turn_tokens_in @@ -1126,22 +1283,12 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse: daily_bucket["cost_usd"] = ( float(daily_bucket["cost_usd"]) + turn_breakdown.cost_usd ) + daily_cost_bases = cast( + list[UsageCostBasis], daily_bucket["cost_bases"] + ) + daily_cost_bases.append(turn_breakdown.cost_basis) - by_provider = [ - _usage_breakdown( - provider=provider, - model=model, - turns=int(values["turns"]), - token_metered_turns=int(values["token_metered_turns"]), - token_unmetered_turns=int(values["token_unmetered_turns"]), - tokens_in=int(values["tokens_in"]), - tokens_out=int(values["tokens_out"]), - stored_cost_usd=float(values["stored_cost_usd"]), - unpriced_tokens_in=int(values["unpriced_tokens_in"]), - unpriced_tokens_out=int(values["unpriced_tokens_out"]), - ) - for (provider, model), values in buckets.items() - ] + by_provider = _merge_usage_breakdowns(turn_breakdowns) by_provider.sort( key=lambda item: ( -item.cost_usd, @@ -1153,6 +1300,9 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse: ) total_cost = round(cost_usd, 6) + total_cost_basis = _aggregate_cost_basis( + item.cost_basis for item in turn_breakdowns + ) return AdminUsageResponse( source="server_session_registry", durable=False, @@ -1167,7 +1317,8 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse: cost_usd=total_cost, recorded_cost_usd=round(recorded_cost_usd, 6), estimated_cost_usd=round(estimated_cost_usd, 6), - budget=_usage_budget(total_cost), + cost_basis=total_cost_basis, + budget=_usage_budget(total_cost, total_cost_basis), evaluator_cache=_usage_evaluator_cache(), by_provider=by_provider[:12], daily_cost=[ @@ -1177,6 +1328,9 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse: tokens_in=int(values["tokens_in"]), tokens_out=int(values["tokens_out"]), cost_usd=round(float(values["cost_usd"]), 6), + cost_basis=_aggregate_cost_basis( + cast(list[UsageCostBasis], values["cost_bases"]) + ), ) for day, values in sorted(daily_buckets.items()) ], diff --git a/apps/api/app/services/llm_pricing.py b/apps/api/app/services/llm_pricing.py index 1ea4ab8..6639984 100644 --- a/apps/api/app/services/llm_pricing.py +++ b/apps/api/app/services/llm_pricing.py @@ -9,10 +9,14 @@ from __future__ import annotations from dataclasses import dataclass +from datetime import date, datetime, timezone MILLION = 1_000_000 RATE_CARD_VERSION = "2026-07-31" +GOOGLE_36_RELEASE_START = date(2026, 7, 21) +GOOGLE_FLASH_INTRO_START = date(2026, 8, 13) +GOOGLE_FLASH_INTRO_END = date(2027, 1, 1) GOOGLE_PRICING_URL = "https://ai.google.dev/gemini-api/docs/pricing" OPENAI_CODEX_RATE_URL = "https://help.openai.com/en/articles/20001106-codex-rate-card" @@ -40,11 +44,65 @@ class CostEstimate: source_url: str +def _pricing_date(value: date | datetime | int | float | str) -> date: + if isinstance(value, datetime): + if value.tzinfo is not None: + value = value.astimezone(timezone.utc) + return value.date() + if isinstance(value, date): + return value + if isinstance(value, (int, float)): + return datetime.fromtimestamp(value, timezone.utc).date() + return date.fromisoformat(value[:10]) + + +def _google_flash_rate(*, model_key: str, priced_on: date) -> ModelRate | None: + if model_key == "gemini-3.6-flash" and priced_on < GOOGLE_36_RELEASE_START: + return None + if model_key == "gemini-3.7-flash" and priced_on < GOOGLE_FLASH_INTRO_START: + return None + if model_key not in {"gemini-3.6-flash", "gemini-3.7-flash"}: + return None + + model_name = "Gemini 3.7 Flash" if model_key == "gemini-3.7-flash" else "Gemini 3.6 Flash" + if GOOGLE_FLASH_INTRO_START <= priced_on < GOOGLE_FLASH_INTRO_END: + return ModelRate( + 0.75, + 3.75, + 0.075, + f"google-{model_key}-standard-intro@2026-08-13", + ( + f"Google {model_name} 프로모션 표준 단가(2026-12-31까지)" + " · 입력 $0.75/M · 캐시 $0.075/M · 출력 $3.75/M" + ), + GOOGLE_PRICING_URL, + ) + + effective_from = "2027-01-01" if priced_on >= GOOGLE_FLASH_INTRO_END else "2026-07-21" + period_label = ( + "2027-01-01부터" + if priced_on >= GOOGLE_FLASH_INTRO_END + else "2026-07-21~2026-08-12" + ) + return ModelRate( + 1.50, + 7.50, + 0.15, + f"google-{model_key}-standard@{effective_from}", + ( + f"Google {model_name} 표준 단가({period_label})" + " · 입력 $1.50/M · 캐시 $0.15/M · 출력 $7.50/M" + ), + GOOGLE_PRICING_URL, + ) + + def _rate( *, provider: str, model: str, tokens_in: int, + priced_on: date, ) -> ModelRate | None: provider_key = provider.strip().lower() model_key = model.strip().lower() @@ -56,15 +114,9 @@ def _rate( break if provider_key == "agy_cli": - if model_key == "gemini-3.6-flash": - return ModelRate( - 1.50, - 7.50, - 0.15, - f"google-gemini-3.6-flash-standard@{RATE_CARD_VERSION}", - "Google Gemini 3.6 Flash 표준 단가 · 입력 $1.50/M · 캐시 $0.15/M · 출력 $7.50/M", - GOOGLE_PRICING_URL, - ) + flash_rate = _google_flash_rate(model_key=model_key, priced_on=priced_on) + if flash_rate is not None: + return flash_rate if model_key == "gemini-3.5-flash": return ModelRate( 1.50, @@ -182,6 +234,7 @@ def estimate_reference_cost( model: str, tokens_in: int, tokens_out: int, + priced_at: date | datetime | int | float | str, cached_input_tokens: int = 0, ) -> CostEstimate | None: """공식 공개 단가로 USD 상당액을 계산한다. @@ -195,7 +248,12 @@ def estimate_reference_cost( if safe_input == 0 and safe_output == 0: return None cached = min(safe_input, max(0, int(cached_input_tokens or 0))) - rate = _rate(provider=provider, model=model, tokens_in=safe_input) + rate = _rate( + provider=provider, + model=model, + tokens_in=safe_input, + priced_on=_pricing_date(priced_at), + ) if rate is None: return None uncached = safe_input - cached diff --git a/apps/api/app/services/usage_report.py b/apps/api/app/services/usage_report.py index 9fd3ffb..c876070 100644 --- a/apps/api/app/services/usage_report.py +++ b/apps/api/app/services/usage_report.py @@ -66,6 +66,8 @@ def build_model_cost_report(usage: Mapping[str, Any]) -> dict[str, Any]: by_provider = list(usage.get("by_provider") or []) budget = dict(usage.get("budget") or {}) evaluator_cache = dict(usage.get("evaluator_cache") or {}) + total_cost_basis = str(usage.get("cost_basis") or "provider_reported") + exact_cost_bases = {"provider_estimate", "provider_reported", "reference_rate"} models: list[dict[str, Any]] = [] for item in by_provider: @@ -87,6 +89,12 @@ def build_model_cost_report(usage: Mapping[str, Any]) -> dict[str, Any]: _number(row.get("recorded_cost_usd"), row_cost), 6 ) row_estimated_cost = round(_number(row.get("estimated_cost_usd")), 6) + row_cost_basis = str(row.get("cost_basis") or "provider_reported") + row_cost_complete = row_cost_basis not in { + "partial", + "partial_upper_bound", + "unavailable", + } models.append( { "provider": str(row.get("provider") or "unknown"), @@ -100,13 +108,24 @@ def build_model_cost_report(usage: Mapping[str, Any]) -> dict[str, Any]: "cost_usd": row_cost, "recorded_cost_usd": row_recorded_cost, "estimated_cost_usd": row_estimated_cost, - "cost_basis": str(row.get("cost_basis") or "provider_reported"), + "cost_basis": row_cost_basis, "rate_label": row.get("rate_label"), "rate_source_url": row.get("rate_source_url"), - "cost_share": _ratio(row_cost, total_cost), + "cost_share": ( + _ratio(row_cost, total_cost) + if total_cost_basis in exact_cost_bases + and row_cost_basis in exact_cost_bases + else None + ), "token_share": _ratio(float(row_tokens), float(total_tokens)), - "cost_per_turn_usd": _cost_per_turn(row_cost, turns), - "cost_per_1k_tokens_usd": _cost_per_1k_tokens(row_cost, row_tokens), + "cost_per_turn_usd": ( + _cost_per_turn(row_cost, turns) if row_cost_complete else None + ), + "cost_per_1k_tokens_usd": ( + _cost_per_1k_tokens(row_cost, row_tokens) + if row_cost_complete + else None + ), } ) models.sort(key=lambda item: (-float(item["cost_usd"]), item["provider"], item["model"])) @@ -120,15 +139,28 @@ def build_model_cost_report(usage: Mapping[str, Any]) -> dict[str, Any]: warnings.append("zero_cost_metered_usage") if estimated_cost > 0: warnings.append("reference_rate_cost") - if any(item["cost_basis"] == "unavailable" for item in models): + if any(item["cost_basis"] == "partial" for item in models): + warnings.append("partial_model_cost") + if any(item["cost_basis"] == "partial_upper_bound" for item in models): + warnings.append("partial_upper_bound_model_cost") + if any(item["cost_basis"] == "reference_upper_bound" for item in models): + warnings.append("reference_upper_bound_cost") + if any( + item["cost_basis"] in {"partial", "partial_upper_bound", "unavailable"} + for item in models + ): warnings.append("unavailable_model_cost") - if str(budget.get("status") or "") in {"warn", "exceeded"}: + if str(budget.get("status") or "") in {"warn", "exceeded", "indeterminate"}: warnings.append(f"budget_{budget.get('status')}") cache_hit_rate = _number(evaluator_cache.get("hit_rate")) if bool(evaluator_cache.get("enabled")) and _integer(evaluator_cache.get("requests")) > 0: if cache_hit_rate < 0.25: warnings.append("low_evaluator_cache_hit_rate") - if models and models[0]["cost_share"] >= 0.8: + if ( + models + and isinstance(models[0]["cost_share"], (float, int)) + and models[0]["cost_share"] >= 0.8 + ): warnings.append("dominant_model_cost") return { @@ -152,14 +184,28 @@ def build_model_cost_report(usage: Mapping[str, Any]) -> dict[str, Any]: "cost_usd": total_cost, "recorded_cost_usd": recorded_cost, "estimated_cost_usd": estimated_cost, - "cost_per_turn_usd": _cost_per_turn(total_cost, metered_turns), - "cost_per_1k_tokens_usd": _cost_per_1k_tokens(total_cost, total_tokens), + "cost_basis": total_cost_basis, + "cost_per_turn_usd": ( + _cost_per_turn(total_cost, metered_turns) + if total_cost_basis not in {"partial", "partial_upper_bound", "unavailable"} + else None + ), + "cost_per_1k_tokens_usd": ( + _cost_per_1k_tokens(total_cost, total_tokens) + if total_cost_basis not in {"partial", "partial_upper_bound", "unavailable"} + else None + ), }, "budget": { "status": str(budget.get("status") or "disabled"), "limit_usd": round(_number(budget.get("limit_usd")), 6), "used_ratio": round(_number(budget.get("used_ratio")), 6), - "remaining_usd": round(_number(budget.get("remaining_usd")), 6), + "remaining_usd": ( + round(_number(budget.get("remaining_usd")), 6) + if budget.get("remaining_usd") is not None + else None + ), + "cost_basis": str(budget.get("cost_basis") or total_cost_basis), }, "evaluator_cache": { "enabled": bool(evaluator_cache.get("enabled")), diff --git a/apps/api/app/test_admin_ops.py b/apps/api/app/test_admin_ops.py index a721a41..5c52633 100644 --- a/apps/api/app/test_admin_ops.py +++ b/apps/api/app/test_admin_ops.py @@ -4,6 +4,7 @@ from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone +from types import SimpleNamespace from unittest.mock import AsyncMock, patch from fastapi import HTTPException @@ -53,6 +54,11 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase): self.fetch_calls = 0 async def fetchrow(self, query, *args, **kwargs): + case.assertIn("LOWER(BTRIM(COALESCE(llm_provider, ''))) = 'e2e'", query) + case.assertIn("LOWER(BTRIM(COALESCE(model, ''))) = 'fake-client'", query) + case.assertIn("COALESCE(tokens_in, 0) = 1", query) + case.assertIn("COALESCE(tokens_out, 0) = 1", query) + case.assertIn("COALESCE(cost_usd, 0) = 0", query) return { "total_turns": 2, "metered_turns": 1, @@ -65,28 +71,15 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase): async def fetch(self, query, *args, **kwargs): self.fetch_calls += 1 - if self.fetch_calls == 2: - case.assertIn("date_trunc('day', created_at)", query) - return [ - { - "day": "2026-06-28", - "turns": 1, - "token_metered_turns": 1, - "token_unmetered_turns": 0, - "tokens_in": 11, - "tokens_out": 13, - "cost_usd": 0.0042, - "unpriced_turns": 0, - "unpriced_tokens_in": 0, - "unpriced_tokens_out": 0, - "provider": "claude_cli", - "model": "gateway-default", - } - ] - case.assertIn("unpriced_tokens_in", query) + case.assertEqual(self.fetch_calls, 1) + case.assertIn("date_trunc('day', created_at AT TIME ZONE 'UTC')", query) + case.assertNotIn("ORDER BY created_at, id", query) + case.assertIn("COUNT(*)::bigint AS matching_turns", query) + case.assertIn("GROUP BY 1, 2, 3, 4, 5, 6", query) case.assertNotIn("LIMIT 12", query) return [ { + "day": "2026-06-28", "provider": "claude_cli", "model": "gateway-default", "turns": 1, @@ -95,6 +88,7 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase): "tokens_in": 11, "tokens_out": 13, "cost_usd": 0.0042, + "matching_turns": 1, "unpriced_turns": 0, "unpriced_tokens_in": 0, "unpriced_tokens_out": 0, @@ -116,6 +110,8 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(usage.token_unmetered_turns, 0) self.assertEqual(usage.by_provider[0].provider, "claude_cli") self.assertEqual(usage.by_provider[0].cost_basis, "provider_estimate") + self.assertEqual(usage.cost_basis, "provider_estimate") + self.assertEqual(usage.daily_cost[0].cost_basis, "provider_estimate") self.assertEqual(usage.recorded_cost_usd, 0.0042) self.assertEqual(usage.estimated_cost_usd, 0) self.assertTrue(usage.evaluator_cache.enabled) @@ -125,6 +121,8 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(usage.daily_cost[0].cost_usd, 0.0042) async def test_usage_from_database_backfills_agy_zero_cost_with_reference_rate(self) -> None: + case = self + class Conn: def __init__(self) -> None: self.fetch_calls = 0 @@ -142,22 +140,20 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase): async def fetch(self, query, *args, **kwargs): self.fetch_calls += 1 - row = { - "provider": "agy_cli", - "model": "gemini-3.6-flash-high", - "turns": 5, - "token_metered_turns": 5, - "token_unmetered_turns": 0, - "tokens_in": 35_703, - "tokens_out": 1_129, - "cost_usd": 0, - "unpriced_turns": 5, - "unpriced_tokens_in": 35_703, - "unpriced_tokens_out": 1_129, - } - if self.fetch_calls == 2: - return [{"day": "2026-07-31", **row}] - return [row] + case.assertEqual(self.fetch_calls, 1) + token_pairs = [(7_141, 226)] * 4 + [(7_139, 225)] + return [ + { + "day": "2026-07-31", + "provider": "agy_cli", + "model": "gemini-3.6-flash-high", + "tokens_in": tokens_in, + "tokens_out": tokens_out, + "cost_usd": 0, + "matching_turns": 1, + } + for tokens_in, tokens_out in token_pairs + ] with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())): usage = await admin_routes._usage_from_database(window_days=30) @@ -168,9 +164,308 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(usage.daily_cost[0].cost_usd, 0.062022) breakdown = usage.by_provider[0] self.assertEqual(breakdown.cost_usd, 0.062022) - self.assertEqual(breakdown.cost_basis, "reference_rate") + self.assertEqual(breakdown.cost_basis, "reference_upper_bound") self.assertIn("Gemini 3.6 Flash", breakdown.rate_label or "") + async def test_usage_from_database_prices_gemini_37_on_its_usage_date(self) -> None: + class Conn: + async def fetchrow(self, query, *args, **kwargs): + return { + "total_turns": 21, + "metered_turns": 21, + "token_metered_turns": 21, + "token_unmetered_turns": 0, + "tokens_in": 115_950, + "tokens_out": 4_407, + "cost_usd": 0, + } + + async def fetch(self, query, *args, **kwargs): + token_pairs = [(5_521, 210)] * 20 + [(5_530, 207)] + return [ + { + "day": "2026-08-28", + "provider": "agy_cli", + "model": "gemini-3.7-flash-high", + "tokens_in": tokens_in, + "tokens_out": tokens_out, + "cost_usd": 0, + "matching_turns": 1, + } + for tokens_in, tokens_out in token_pairs + ] + + with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())): + usage = await admin_routes._usage_from_database(window_days=30) + + self.assertEqual(usage.cost_usd, 0.103489) + self.assertEqual(usage.recorded_cost_usd, 0) + self.assertEqual(usage.estimated_cost_usd, 0.103489) + self.assertEqual(usage.daily_cost[0].cost_usd, 0.103489) + self.assertEqual(usage.cost_basis, "reference_upper_bound") + self.assertEqual(usage.daily_cost[0].cost_basis, "reference_upper_bound") + breakdown = usage.by_provider[0] + self.assertEqual(breakdown.cost_usd, 0.103489) + self.assertEqual(breakdown.cost_basis, "reference_upper_bound") + self.assertIn("Gemini 3.7 Flash", breakdown.rate_label or "") + + async def test_usage_from_database_keeps_effective_rates_across_date_boundary(self) -> None: + class Conn: + async def fetchrow(self, query, *args, **kwargs): + return { + "total_turns": 2, + "metered_turns": 2, + "token_metered_turns": 2, + "token_unmetered_turns": 0, + "tokens_in": 2_000_000, + "tokens_out": 2_000_000, + "cost_usd": 0, + "matching_turns": 1, + } + + async def fetch(self, query, *args, **kwargs): + base = { + "provider": "agy_cli", + "model": "gemini-3.6-flash-high", + "tokens_in": 1_000_000, + "tokens_out": 1_000_000, + "cost_usd": 0, + "matching_turns": 1, + } + return [ + {"day": "2026-08-12", **base}, + {"day": "2026-08-13", **base}, + ] + + with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())): + usage = await admin_routes._usage_from_database(window_days=30) + + self.assertEqual(usage.cost_usd, 13.5) + self.assertEqual(usage.estimated_cost_usd, 13.5) + self.assertEqual([item.cost_usd for item in usage.daily_cost], [9.0, 4.5]) + self.assertEqual(len(usage.by_provider), 1) + self.assertEqual(usage.by_provider[0].cost_usd, 13.5) + self.assertIn("기간별", usage.by_provider[0].rate_label or "") + + async def test_usage_from_database_prices_request_tiers_before_aggregation(self) -> None: + class Conn: + async def fetchrow(self, query, *args, **kwargs): + return { + "total_turns": 2, + "metered_turns": 2, + "token_metered_turns": 2, + "token_unmetered_turns": 0, + "tokens_in": 300_000, + "tokens_out": 20_000, + "cost_usd": 0, + } + + async def fetch(self, query, *args, **kwargs): + return [ + { + "day": "2026-08-28", + "provider": "agy_cli", + "model": "gemini-3.1-pro-high", + "tokens_in": 150_000, + "tokens_out": 10_000, + "cost_usd": 0, + "matching_turns": 2, + }, + ] + + with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())): + database_usage = await admin_routes._usage_from_database(window_days=30) + + observed_at = datetime(2026, 8, 28, 12, tzinfo=timezone.utc).timestamp() + runtime_turns = [ + SimpleNamespace( + speaker="client", + created_at=observed_at, + llm_provider="agy_cli", + model="gemini-3.1-pro-high", + tokens_in=150_000, + tokens_out=10_000, + cost_usd=0, + ) + for _ in range(2) + ] + with ( + patch.object( + admin_routes.store, + "list", + return_value=[SimpleNamespace(turns=runtime_turns)], + ), + patch.object(admin_routes.time, "time", return_value=observed_at + 1), + ): + runtime_usage = admin_routes._usage_from_runtime_store(window_days=30) + + self.assertEqual(database_usage.cost_usd, 0.84) + self.assertEqual(database_usage.by_provider[0].cost_usd, 0.84) + self.assertEqual(database_usage.daily_cost[0].cost_usd, 0.84) + self.assertEqual(runtime_usage.cost_usd, database_usage.cost_usd) + self.assertEqual(runtime_usage.by_provider[0].cost_usd, 0.84) + + async def test_usage_from_database_marks_mixed_pricing_as_partial(self) -> None: + class Conn: + async def fetchrow(self, query, *args, **kwargs): + return { + "total_turns": 2, + "metered_turns": 2, + "token_metered_turns": 2, + "token_unmetered_turns": 0, + "tokens_in": 2_000_000, + "tokens_out": 2_000_000, + "cost_usd": 0, + "matching_turns": 1, + } + + async def fetch(self, query, *args, **kwargs): + base = { + "provider": "agy_cli", + "model": "gemini-3.7-flash-high", + "tokens_in": 1_000_000, + "tokens_out": 1_000_000, + "cost_usd": 0, + "matching_turns": 1, + } + return [ + {"day": "2026-08-12", **base}, + {"day": "2026-08-13", **base}, + ] + + with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())): + usage = await admin_routes._usage_from_database(window_days=30) + + self.assertEqual(usage.cost_usd, 4.5) + self.assertEqual(usage.cost_basis, "partial_upper_bound") + self.assertEqual(usage.budget.cost_basis, "partial_upper_bound") + self.assertEqual(usage.by_provider[0].cost_basis, "partial_upper_bound") + self.assertEqual( + usage.by_provider[0].rate_label, + "일부 호출 미산정 · 산정된 부분도 상한 추정", + ) + self.assertEqual(usage.daily_cost[0].cost_basis, "unavailable") + self.assertEqual(usage.daily_cost[1].cost_basis, "reference_upper_bound") + + def test_cost_basis_preserves_missing_and_upper_bound_dimensions(self) -> None: + self.assertEqual( + admin_routes._aggregate_cost_basis( + ["reference_upper_bound", "unavailable"] + ), + "partial_upper_bound", + ) + self.assertEqual( + admin_routes._aggregate_cost_basis(["reference_rate", "unavailable"]), + "partial", + ) + self.assertEqual( + admin_routes._aggregate_cost_basis( + ["provider_reported", "reference_upper_bound"] + ), + "reference_upper_bound", + ) + + def test_budget_is_indeterminate_when_cost_is_not_bounded_both_ways(self) -> None: + with patch.object(admin_routes.settings, "admin_usage_budget_usd", 10.0): + partial = admin_routes._usage_budget(4.5, "partial") + partial_upper = admin_routes._usage_budget( + 4.5, "partial_upper_bound" + ) + safe_upper = admin_routes._usage_budget(7.5, "reference_upper_bound") + + self.assertEqual(partial.status, "indeterminate") + self.assertEqual(partial.remaining_usd, 5.5) + self.assertEqual(partial_upper.status, "indeterminate") + self.assertIsNone(partial_upper.remaining_usd) + self.assertEqual(safe_upper.status, "ok") + self.assertEqual(safe_upper.remaining_usd, 2.5) + + def test_synthetic_e2e_signature_is_not_reportable_usage(self) -> None: + self.assertFalse(admin_routes._is_reportable_usage("e2e", "fake-client", 1, 1, 0)) + self.assertFalse( + admin_routes._is_reportable_usage(" E2E ", " FAKE-CLIENT ", 1, 1, 0) + ) + self.assertTrue(admin_routes._is_reportable_usage("e2e", "fake-client", 2, 1, 0)) + self.assertTrue(admin_routes._is_reportable_usage("e2e", "real-client", 1, 1, 0)) + self.assertTrue(admin_routes._is_reportable_usage("agy_cli", "fake-client", 1, 1, 0)) + + def test_stored_reference_cost_does_not_claim_a_recomputed_rate(self) -> None: + breakdown = admin_routes._usage_breakdown( + provider="agy_cli", + model="gemini-3.6-flash-high", + turns=1, + token_metered_turns=1, + token_unmetered_turns=0, + tokens_in=1_000_000, + tokens_out=0, + stored_cost_usd=1.5, + unpriced_tokens_in=0, + unpriced_tokens_out=0, + priced_at="2026-08-28", + ) + + self.assertEqual(breakdown.cost_usd, 1.5) + self.assertEqual(breakdown.cost_basis, "reference_rate") + self.assertEqual(breakdown.rate_label, "호출 시점에 저장된 참조단가 추정값") + self.assertIsNone(breakdown.rate_source_url) + + def test_legacy_reference_fallback_is_labeled_as_an_upper_bound(self) -> None: + breakdown = admin_routes._usage_breakdown( + provider="agy_cli", + model="gemini-3.7-flash-high", + turns=1, + token_metered_turns=1, + token_unmetered_turns=0, + tokens_in=115_950, + tokens_out=4_407, + stored_cost_usd=0, + unpriced_tokens_in=115_950, + unpriced_tokens_out=4_407, + priced_at="2026-08-28", + ) + + self.assertEqual(breakdown.cost_basis, "reference_upper_bound") + self.assertIn("캐시 미보존", breakdown.rate_label or "") + + def test_runtime_usage_prices_by_turn_date_and_excludes_synthetic_provider(self) -> None: + observed_at = datetime(2026, 8, 28, 12, tzinfo=timezone.utc).timestamp() + turns = [ + SimpleNamespace( + speaker="client", + created_at=observed_at, + llm_provider="agy_cli", + model="gemini-3.7-flash-high", + tokens_in=115_950, + tokens_out=4_407, + cost_usd=0, + ), + SimpleNamespace( + speaker="client", + created_at=observed_at, + llm_provider="e2e", + model="fake-client", + tokens_in=1, + tokens_out=1, + cost_usd=0, + ), + ] + with ( + patch.object( + admin_routes.store, + "list", + return_value=[SimpleNamespace(turns=turns)], + ), + patch.object(admin_routes.time, "time", return_value=observed_at + 1), + ): + usage = admin_routes._usage_from_runtime_store(window_days=30) + + self.assertEqual(usage.total_turns, 1) + self.assertEqual(usage.metered_turns, 1) + self.assertEqual(usage.tokens_in, 115_950) + self.assertEqual(usage.cost_usd, 0.103489) + self.assertEqual(len(usage.by_provider), 1) + self.assertEqual(usage.by_provider[0].model, "gemini-3.7-flash-high") + async def test_uptime_from_database_aggregates_health_samples(self) -> None: now = datetime.now(timezone.utc) diff --git a/apps/api/app/test_admin_usage_postgres.py b/apps/api/app/test_admin_usage_postgres.py new file mode 100644 index 0000000..f6974a5 --- /dev/null +++ b/apps/api/app/test_admin_usage_postgres.py @@ -0,0 +1,184 @@ +"""격리 PostgreSQL에서 관리자 비용 원장 SQL을 검증하는 선택형 통합 테스트.""" + +from __future__ import annotations + +import os +import unittest +from datetime import datetime, timezone +from pathlib import Path +from urllib.parse import urlparse + +import asyncpg + +from . import db +from .routes import admin as admin_routes +from .services.llm_pricing import estimate_reference_cost + + +TEST_DATABASE_URL = os.getenv("VIGNETTE_USAGE_TEST_DATABASE_URL", "").strip() + + +@unittest.skipUnless( + TEST_DATABASE_URL, + "VIGNETTE_USAGE_TEST_DATABASE_URL이 설정된 새 격리 DB에서만 실행", +) +class AdminUsagePostgresTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + parsed = urlparse(TEST_DATABASE_URL) + if parsed.path != "/vignette_usage_test": + self.fail("격리 DB 이름은 vignette_usage_test여야 한다") + if db._pool is not None: + self.fail("기존 애플리케이션 DB pool이 있는 프로세스에서는 실행할 수 없다") + + bootstrap = await asyncpg.connect(TEST_DATABASE_URL) + try: + # 기존 스키마를 지우지 않는다. 이미 쓰인 DB라면 CREATE가 실패해 닫힌다. + await bootstrap.execute( + """ + CREATE SCHEMA app; + CREATE TABLE app.turns ( + id BIGSERIAL PRIMARY KEY, + speaker TEXT NOT NULL, + llm_provider TEXT, + model TEXT, + tokens_in INT, + tokens_out INT, + cost_usd NUMERIC(12,6), + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + """ + ) + migration = ( + Path(__file__).resolve().parents[3] + / "infra" + / "db" + / "init" + / "18_admin_usage_ledger_index.sql" + ).read_text(encoding="utf-8") + await bootstrap.execute(migration) + finally: + await bootstrap.close() + + db._pool = await asyncpg.create_pool( + dsn=TEST_DATABASE_URL, + min_size=1, + max_size=2, + init=db._init_connection, + ) + + async def asyncTearDown(self) -> None: + if db._pool is not None: + await db._pool.close() + db._pool = None + + async def test_usage_sql_preserves_pricing_boundaries_and_fake_exclusion(self) -> None: + observed_at = datetime.now(timezone.utc) + token_pairs = [(5_521, 210)] * 20 + [(5_530, 207)] + assert db._pool is not None + async with db._pool.acquire() as conn: + await conn.executemany( + """ + INSERT INTO app.turns ( + speaker, llm_provider, model, tokens_in, tokens_out, cost_usd, created_at + ) VALUES ('client', $1, $2, $3, $4, 0, $5) + """, + [ + ( + "agy_cli", + "gemini-3.7-flash-high", + tokens_in, + tokens_out, + observed_at, + ) + for tokens_in, tokens_out in token_pairs + ], + ) + await conn.executemany( + """ + INSERT INTO app.turns ( + speaker, llm_provider, model, tokens_in, tokens_out, cost_usd, created_at + ) VALUES ('client', ' E2E ', ' FAKE-CLIENT ', 1, 1, 0, $1) + """, + [(observed_at,)] * 6, + ) + + expected = estimate_reference_cost( + provider="agy_cli", + model="gemini-3.7-flash-high", + tokens_in=115_950, + tokens_out=4_407, + priced_at=observed_at, + ) + self.assertIsNotNone(expected) + assert expected is not None + + usage = await admin_routes._usage_from_database(window_days=36_500) + + self.assertEqual(usage.total_turns, 21) + self.assertEqual(usage.token_metered_turns, 21) + self.assertEqual(usage.tokens_in, 115_950) + self.assertEqual(usage.tokens_out, 4_407) + self.assertEqual(usage.cost_usd, round(expected.cost_usd, 6)) + self.assertEqual(len(usage.by_provider), 1) + self.assertEqual(usage.by_provider[0].model, "gemini-3.7-flash-high") + self.assertEqual(usage.by_provider[0].cost_basis, "reference_upper_bound") + + async with db._pool.acquire() as conn: + index_definition = await conn.fetchval( + """ + SELECT indexdef + FROM pg_indexes + WHERE schemaname = 'app' + AND tablename = 'turns' + AND indexname = 'idx_turns_admin_usage_created_at' + """ + ) + self.assertIn("created_at DESC", index_definition or "") + self.assertIn("speaker = 'client'", index_definition or "") + + async with db._pool.acquire() as conn: + await conn.executemany( + """ + INSERT INTO app.turns ( + speaker, llm_provider, model, tokens_in, tokens_out, cost_usd, created_at + ) VALUES ('client', 'agy_cli', 'gemini-3.1-pro-high', 150000, 10000, 0, $1) + """, + [(observed_at,), (observed_at,)], + ) + + tier_usage = await admin_routes._usage_from_database(window_days=36_500) + tier_row = next( + item for item in tier_usage.by_provider if item.model == "gemini-3.1-pro-high" + ) + self.assertEqual(tier_row.turns, 2) + self.assertEqual(tier_row.cost_usd, 0.84) + + async with db._pool.acquire() as conn: + await conn.executemany( + """ + INSERT INTO app.turns ( + speaker, llm_provider, model, tokens_in, tokens_out, cost_usd, created_at + ) VALUES ('client', 'agy_cli', 'gemini-3.7-flash-high', 1000000, 1000000, 0, $1) + """, + [ + (datetime(2026, 8, 12, 12, tzinfo=timezone.utc),), + (datetime(2026, 8, 13, 12, tzinfo=timezone.utc),), + ], + ) + + partial_usage = await admin_routes._usage_from_database(window_days=36_500) + partial_row = next( + item + for item in partial_usage.by_provider + if item.model == "gemini-3.7-flash-high" + ) + self.assertEqual(partial_row.turns, 23) + self.assertEqual(partial_row.cost_usd, round(expected.cost_usd + 4.5, 6)) + self.assertEqual(partial_usage.cost_basis, "partial_upper_bound") + self.assertEqual(partial_usage.budget.cost_basis, "partial_upper_bound") + self.assertEqual(partial_row.cost_basis, "partial_upper_bound") + self.assertIn("일부 호출 미산정", partial_row.rate_label or "") + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/api/app/test_llm_pricing.py b/apps/api/app/test_llm_pricing.py index 8ef50e2..7b09182 100644 --- a/apps/api/app/test_llm_pricing.py +++ b/apps/api/app/test_llm_pricing.py @@ -1,17 +1,45 @@ from __future__ import annotations import unittest +from datetime import date from .services.llm_pricing import estimate_reference_cost class LlmPricingTest(unittest.TestCase): + def test_agy_gemini_36_is_unavailable_before_model_release(self) -> None: + estimate = estimate_reference_cost( + provider="agy_cli", + model="gemini-3.6-flash-high", + tokens_in=1_000, + tokens_out=100, + priced_at=date(2026, 7, 20), + ) + + self.assertIsNone(estimate) + + def test_agy_gemini_36_rate_starts_on_model_release_date(self) -> None: + estimate = estimate_reference_cost( + provider="agy_cli", + model="gemini-3.6-flash-high", + tokens_in=1_000, + tokens_out=100, + priced_at=date(2026, 7, 21), + ) + + self.assertIsNotNone(estimate) + assert estimate is not None + self.assertEqual(estimate.cost_usd, 0.00225) + self.assertIn("2026-07-21~2026-08-12", estimate.rate_label) + self.assertTrue(estimate.rate_id.endswith("@2026-07-21")) + def test_agy_gemini_reasoning_suffix_uses_base_model_standard_rate(self) -> None: estimate = estimate_reference_cost( provider="agy_cli", model="gemini-3.6-flash-high", tokens_in=35_703, tokens_out=1_129, + priced_at=date(2026, 7, 31), ) self.assertIsNotNone(estimate) @@ -19,6 +47,59 @@ class LlmPricingTest(unittest.TestCase): self.assertEqual(estimate.cost_usd, 0.062022) self.assertIn("Gemini 3.6 Flash", estimate.rate_label) + def test_agy_gemini_36_uses_introductory_rate_from_august_13(self) -> None: + estimate = estimate_reference_cost( + provider="agy_cli", + model="gemini-3.6-flash-medium", + tokens_in=35_703, + tokens_out=1_129, + priced_at=date(2026, 8, 13), + ) + + self.assertIsNotNone(estimate) + assert estimate is not None + self.assertEqual(estimate.cost_usd, 0.031011) + self.assertIn("2026-12-31까지", estimate.rate_label) + + def test_agy_gemini_37_uses_introductory_rate_and_reasoning_suffix(self) -> None: + estimate = estimate_reference_cost( + provider="agy_cli", + model="gemini-3.7-flash-high", + tokens_in=115_950, + tokens_out=4_407, + priced_at=date(2026, 8, 28), + ) + + self.assertIsNotNone(estimate) + assert estimate is not None + self.assertEqual(estimate.cost_usd, 0.10348875) + self.assertIn("Gemini 3.7 Flash", estimate.rate_label) + + def test_agy_gemini_37_is_unavailable_before_model_release(self) -> None: + estimate = estimate_reference_cost( + provider="agy_cli", + model="gemini-3.7-flash-high", + tokens_in=1_000, + tokens_out=100, + priced_at=date(2026, 8, 12), + ) + + self.assertIsNone(estimate) + + def test_agy_gemini_37_switches_to_standard_rate_in_2027(self) -> None: + estimate = estimate_reference_cost( + provider="agy_cli", + model="gemini-3.7-flash-low", + tokens_in=115_950, + tokens_out=4_407, + priced_at=date(2027, 1, 1), + ) + + self.assertIsNotNone(estimate) + assert estimate is not None + self.assertEqual(estimate.cost_usd, 0.2069775) + self.assertIn("2027-01-01부터", estimate.rate_label) + def test_codex_credit_rate_accounts_for_cached_input(self) -> None: estimate = estimate_reference_cost( provider="codex_cli", @@ -26,6 +107,7 @@ class LlmPricingTest(unittest.TestCase): tokens_in=12, cached_input_tokens=2, tokens_out=3, + priced_at=date(2026, 8, 28), ) self.assertIsNotNone(estimate) @@ -39,6 +121,7 @@ class LlmPricingTest(unittest.TestCase): model="gpt-oss-120b-medium", tokens_in=1000, tokens_out=100, + priced_at=date(2026, 8, 28), ) self.assertIsNone(estimate) diff --git a/apps/api/app/test_usage_report.py b/apps/api/app/test_usage_report.py index 099874a..0e3e17c 100644 --- a/apps/api/app/test_usage_report.py +++ b/apps/api/app/test_usage_report.py @@ -78,6 +78,81 @@ class UsageReportTest(unittest.TestCase): self.assertIsNone(report["top_cost_model"]) self.assertEqual(report["summary"]["metered_coverage"], 0.0) + def test_model_cost_report_warns_when_only_part_of_a_model_is_priced(self) -> None: + report = build_model_cost_report( + { + "cost_usd": 4.5, + "cost_basis": "partial", + "by_provider": [ + { + "provider": "agy_cli", + "model": "gemini-3.7-flash-high", + "turns": 2, + "tokens_in": 2_000_000, + "tokens_out": 2_000_000, + "cost_usd": 4.5, + "cost_basis": "partial", + } + ], + } + ) + + self.assertIn("partial_model_cost", report["warnings"]) + self.assertIn("unavailable_model_cost", report["warnings"]) + self.assertIsNone(report["summary"]["cost_per_turn_usd"]) + self.assertIsNone(report["models"][0]["cost_share"]) + + def test_model_cost_report_warns_for_reference_upper_bound(self) -> None: + report = build_model_cost_report( + { + "cost_usd": 0.1, + "by_provider": [ + { + "provider": "agy_cli", + "model": "gemini-3.7-flash-high", + "turns": 1, + "tokens_in": 100_000, + "tokens_out": 1_000, + "cost_usd": 0.1, + "cost_basis": "reference_upper_bound", + } + ], + } + ) + + self.assertIn("reference_upper_bound_cost", report["warnings"]) + + def test_model_cost_report_preserves_partial_upper_bound_uncertainty(self) -> None: + report = build_model_cost_report( + { + "cost_usd": 4.5, + "cost_basis": "partial_upper_bound", + "budget": { + "status": "indeterminate", + "remaining_usd": None, + "cost_basis": "partial_upper_bound", + }, + "by_provider": [ + { + "provider": "agy_cli", + "model": "gemini-3.7-flash-high", + "turns": 2, + "tokens_in": 2_000_000, + "tokens_out": 2_000_000, + "cost_usd": 4.5, + "cost_basis": "partial_upper_bound", + } + ], + } + ) + + self.assertEqual(report["summary"]["cost_basis"], "partial_upper_bound") + self.assertIsNone(report["summary"]["cost_per_turn_usd"]) + self.assertIsNone(report["budget"]["remaining_usd"]) + self.assertIn("partial_upper_bound_model_cost", report["warnings"]) + self.assertIn("unavailable_model_cost", report["warnings"]) + self.assertIn("budget_indeterminate", report["warnings"]) + if __name__ == "__main__": unittest.main() diff --git a/apps/api/engine_gateway/provider_registry.py b/apps/api/engine_gateway/provider_registry.py index 432e18b..5594829 100644 --- a/apps/api/engine_gateway/provider_registry.py +++ b/apps/api/engine_gateway/provider_registry.py @@ -13,6 +13,7 @@ import shutil import tempfile import time from dataclasses import dataclass +from datetime import datetime, timezone from pathlib import Path from typing import Any, AsyncIterator, Iterable, Literal, cast @@ -80,6 +81,10 @@ def _now() -> float: return time.time() +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + def _efforts(values: Iterable[str]) -> list[ReasoningEffort]: allowed = set(ENGINE_REASONING_EFFORTS) return [cast(ReasoningEffort, value) for value in values if value in allowed] @@ -597,6 +602,7 @@ def _structured_or_none(text: str, req: GenerateRequest) -> dict[str, Any] | Non async def _generate_codex( req: GenerateRequest, system_prompt: str, user_payload: str ) -> ProviderGenerateResult: + pricing_started_at = _utcnow() binary = _binary("CODEX_BIN", "codex") if binary is None: raise ProviderError("Codex CLI를 찾을 수 없습니다.") @@ -652,6 +658,7 @@ async def _generate_codex( model=model, tokens_in=tokens_in, tokens_out=tokens_out, + priced_at=pricing_started_at, cached_input_tokens=cached_input_tokens, ) return ProviderGenerateResult( @@ -685,6 +692,7 @@ async def _stream_agy( --continue/--conversation을 쓰지 않는다. 회기 메모리는 매 요청의 마스킹된 prompt가 소유하고, 프로세스는 응답 뒤 종료한다. """ + pricing_started_at = _utcnow() binary = _binary("AGY_BIN", "agy") if binary is None: raise ProviderError("Agy CLI를 찾을 수 없습니다.") @@ -781,6 +789,7 @@ async def _stream_agy( model=model, tokens_in=tokens_in, tokens_out=tokens_out, + priced_at=pricing_started_at, cached_input_tokens=cached_input_tokens, ) yield ProviderStreamEvent( @@ -800,6 +809,7 @@ async def _stream_agy( async def _generate_claude_api( req: GenerateRequest, system_prompt: str ) -> ProviderGenerateResult: + pricing_started_at = _utcnow() api_key = os.environ.get("ANTHROPIC_API_KEY", "").strip() if not api_key: raise ProviderError("ANTHROPIC_API_KEY가 설정되지 않았습니다.") @@ -849,6 +859,7 @@ async def _generate_claude_api( model=str(body.get("model") or model), tokens_in=tokens_in, tokens_out=tokens_out, + priced_at=pricing_started_at, cached_input_tokens=int(usage.get("cache_read_input_tokens") or 0), ) return ProviderGenerateResult( diff --git a/apps/api/engine_gateway/test_provider_registry.py b/apps/api/engine_gateway/test_provider_registry.py index eeb7df3..e0ae71c 100644 --- a/apps/api/engine_gateway/test_provider_registry.py +++ b/apps/api/engine_gateway/test_provider_registry.py @@ -1,5 +1,6 @@ import json import unittest +from datetime import datetime, timezone from unittest.mock import AsyncMock, patch from app.contracts.engine_gateway import EngineMessage, GenerateRequest @@ -274,19 +275,19 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase): source="live_cli", models=[ provider_registry.EngineModelOption( - id="gemini-3.6-flash-high", - label="Gemini 3.6 Flash (High)", + id="gemini-3.7-flash-high", + label="Gemini 3.7 Flash (High)", reasoning_efforts=["high"], default_reasoning_effort="high", ) ], - default_model="gemini-3.6-flash-high", + default_model="gemini-3.7-flash-high", default_reasoning_effort="high", fetched_at=1, ) request = GenerateRequest( provider="agy_cli", - model="gemini-3.6-flash-high", + model="gemini-3.7-flash-high", reasoning_effort="high", messages=[EngineMessage(role="user", content="hello")], ) @@ -322,6 +323,11 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase): with ( patch.object(provider_registry, "_binary", return_value="agy"), + patch.object( + provider_registry, + "_utcnow", + return_value=datetime(2026, 8, 28, tzinfo=timezone.utc), + ), patch.object( provider_registry, "discover_capabilities", @@ -342,7 +348,7 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(result.text, "OK") self.assertEqual(result.tokens_in, 12) self.assertEqual(result.tokens_out, 2) - self.assertEqual(result.cost_usd, 0.0000303) + self.assertEqual(result.cost_usd, 0.00001515) args = captured[0] print_index = args.index("--print") self.assertEqual(print_index, len(args) - 2) @@ -413,6 +419,11 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase): with ( patch.object(provider_registry, "_binary", return_value="agy.exe"), + patch.object( + provider_registry, + "_utcnow", + return_value=datetime(2026, 8, 28, tzinfo=timezone.utc), + ), patch.object( provider_registry, "discover_capabilities", @@ -438,7 +449,7 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(events[-1].result.text, "안녕") self.assertEqual(events[-1].result.tokens_in, 12) self.assertEqual(events[-1].result.tokens_out, 2) - self.assertEqual(events[-1].result.cost_usd, 0.0000303) + self.assertEqual(events[-1].result.cost_usd, 0.00001515) args = captured[0] self.assertIn("--output-format", args) self.assertEqual(args[args.index("--output-format") + 1], "stream-json") diff --git a/infra/db/init/18_admin_usage_ledger_index.sql b/infra/db/init/18_admin_usage_ledger_index.sql new file mode 100644 index 0000000..c0e613e --- /dev/null +++ b/infra/db/init/18_admin_usage_ledger_index.sql @@ -0,0 +1,10 @@ +-- ============================================================================= +-- Vignette · 18_admin_usage_ledger_index.sql +-- 관리자 AI 사용량 원장의 기간 조회가 전체 turns 테이블을 훑지 않게 한다. +-- 기존 볼륨과 신규 볼륨에 중복 적용되어도 안전해야 한다. +-- 운영 턴 쓰기를 막지 않도록 release agent가 이 파일을 트랜잭션 밖에서 실행한다. +-- ============================================================================= + +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_turns_admin_usage_created_at + ON app.turns (created_at DESC) + WHERE speaker = 'client';