feat: 强化用量计费状态机，新增钱包每日消费汇总分类账

- 将 usage.billing_status 默认值从 settled 改为 pending，完善
  pending -> settled/void 的状态转换逻辑，确保终态不可逆
- 新增 WalletDailyUsageLedger 模型和聚合服务，按账单日汇总
  每个钱包的消费金额、请求数和 token 用量
- 前端钱包中心页面集成每日消费流水展示，支持与充值记录混合
  排序和分页
- 新增两个数据库迁移：修复历史数据状态一致性、创建每日汇总表
- 补充计费状态机单元测试

Closes #218

Co-authored-by: LewisPen <LewisPen@nyadoo.com>

alembic/versions/20260311_1900_9e4f1a2b3c4d_tighten_usage_billing_state_machine.py

@@ -0,0 +1,159 @@
+"""tighten usage billing state machine
+
+Revision ID: 9e4f1a2b3c4d
+Revises: a3f1b7c9d2e4
+Create Date: 2026-03-11 19:00:00.000000+00:00
+
+This migration does two things:
+1. Change new `usage.billing_status` default from `settled` to `pending`.
+2. Repair only the clearly-safe inconsistent historical rows for production:
+   - failed/cancelled zero-cost rows that were marked settled are converted to void
+   - terminal rows missing finalized_at are backfilled from created_at
+
+Ambiguous positive-cost settled rows are intentionally left untouched for manual audit.
+
+All data updates are batched (10000 rows per iteration) to avoid long-held locks
+and excessive WAL generation on large usage tables.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy import inspect
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "9e4f1a2b3c4d"
+down_revision: str | None = "a3f1b7c9d2e4"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+BATCH_SIZE = 10000
+
+
+def _table_exists(table_name: str) -> bool:
+    bind = op.get_bind()
+    insp = inspect(bind)
+    insp.clear_cache()
+    return table_name in insp.get_table_names()
+
+
+def _column_exists(table_name: str, column_name: str) -> bool:
+    bind = op.get_bind()
+    insp = inspect(bind)
+    insp.clear_cache()
+    return column_name in [col["name"] for col in insp.get_columns(table_name)]
+
+
+def upgrade() -> None:
+    if not _table_exists("usage"):
+        return
+
+    if _column_exists("usage", "billing_status"):
+        op.alter_column(
+            "usage",
+            "billing_status",
+            existing_type=sa.String(length=20),
+            server_default="pending",
+            existing_nullable=False,
+        )
+
+    required_columns = {
+        "billing_status",
+        "status",
+        "total_cost_usd",
+        "request_cost_usd",
+        "actual_total_cost_usd",
+        "actual_request_cost_usd",
+        "wallet_balance_after",
+        "finalized_at",
+        "created_at",
+    }
+    if not required_columns.issubset(
+        {col for col in required_columns if _column_exists("usage", col)}
+    ):
+        return
+
+    conn = op.get_bind()
+
+    # Step 1: billing_status IS NULL -> 'pending' (batched)
+    while True:
+        result = conn.execute(
+            sa.text("""
+                WITH batch AS (
+                    SELECT id FROM usage
+                    WHERE billing_status IS NULL
+                    LIMIT :batch_size
+                    FOR UPDATE SKIP LOCKED
+                )
+                UPDATE usage
+                SET billing_status = 'pending'
+                FROM batch WHERE usage.id = batch.id
+                """),
+            {"batch_size": BATCH_SIZE},
+        )
+        if result.rowcount < BATCH_SIZE:
+            break
+
+    # Step 2: failed/cancelled zero-cost settled -> void (batched)
+    while True:
+        result = conn.execute(
+            sa.text("""
+                WITH batch AS (
+                    SELECT id FROM usage
+                    WHERE billing_status = 'settled'
+                      AND status IN ('failed', 'cancelled')
+                      AND COALESCE(total_cost_usd, 0) = 0
+                      AND wallet_balance_after IS NULL
+                    LIMIT :batch_size
+                    FOR UPDATE SKIP LOCKED
+                )
+                UPDATE usage
+                SET billing_status = 'void',
+                    finalized_at = COALESCE(usage.finalized_at, usage.created_at),
+                    total_cost_usd = 0,
+                    request_cost_usd = 0,
+                    actual_total_cost_usd = 0,
+                    actual_request_cost_usd = 0
+                FROM batch WHERE usage.id = batch.id
+                """),
+            {"batch_size": BATCH_SIZE},
+        )
+        if result.rowcount < BATCH_SIZE:
+            break
+
+    # Step 3: backfill finalized_at for terminal rows (batched)
+    while True:
+        result = conn.execute(
+            sa.text("""
+                WITH batch AS (
+                    SELECT id FROM usage
+                    WHERE billing_status IN ('settled', 'void')
+                      AND finalized_at IS NULL
+                    LIMIT :batch_size
+                    FOR UPDATE SKIP LOCKED
+                )
+                UPDATE usage
+                SET finalized_at = COALESCE(usage.finalized_at, usage.created_at)
+                FROM batch WHERE usage.id = batch.id
+                """),
+            {"batch_size": BATCH_SIZE},
+        )
+        if result.rowcount < BATCH_SIZE:
+            break
+
+
+def downgrade() -> None:
+    if not _table_exists("usage") or not _column_exists("usage", "billing_status"):
+        return
+
+    op.alter_column(
+        "usage",
+        "billing_status",
+        existing_type=sa.String(length=20),
+        server_default="settled",
+        existing_nullable=False,
+    )

alembic/versions/20260311_2100_d4e5f6a7b8c9_add_wallet_daily_usage_ledgers.py

@@ -0,0 +1,113 @@
+"""add wallet daily usage ledgers
+
+Revision ID: d4e5f6a7b8c9
+Revises: 9e4f1a2b3c4d
+Create Date: 2026-03-11 21:00:00.000000+00:00
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy import inspect
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "d4e5f6a7b8c9"
+down_revision: str | None = "9e4f1a2b3c4d"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _table_exists(table_name: str) -> bool:
+    bind = op.get_bind()
+    insp = inspect(bind)
+    insp.clear_cache()
+    return table_name in insp.get_table_names()
+
+
+def _column_exists(table_name: str, column_name: str) -> bool:
+    bind = op.get_bind()
+    insp = inspect(bind)
+    insp.clear_cache()
+    return column_name in [col["name"] for col in insp.get_columns(table_name)]
+
+
+def _index_exists(table_name: str, index_name: str) -> bool:
+    bind = op.get_bind()
+    insp = inspect(bind)
+    insp.clear_cache()
+    return any(idx["name"] == index_name for idx in insp.get_indexes(table_name))
+
+
+def upgrade() -> None:
+    if not _table_exists("wallet_daily_usage_ledgers"):
+        op.create_table(
+            "wallet_daily_usage_ledgers",
+            sa.Column("id", sa.String(length=36), nullable=False),
+            sa.Column("wallet_id", sa.String(length=36), nullable=False),
+            sa.Column("billing_date", sa.Date(), nullable=False),
+            sa.Column("billing_timezone", sa.String(length=64), nullable=False),
+            sa.Column("total_cost_usd", sa.Numeric(20, 8), nullable=False, server_default="0"),
+            sa.Column("total_requests", sa.Integer(), nullable=False, server_default="0"),
+            sa.Column("input_tokens", sa.BigInteger(), nullable=False, server_default="0"),
+            sa.Column("output_tokens", sa.BigInteger(), nullable=False, server_default="0"),
+            sa.Column("cache_creation_tokens", sa.BigInteger(), nullable=False, server_default="0"),
+            sa.Column("cache_read_tokens", sa.BigInteger(), nullable=False, server_default="0"),
+            sa.Column("first_finalized_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("last_finalized_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("aggregated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.ForeignKeyConstraint(["wallet_id"], ["wallets.id"], ondelete="CASCADE"),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "wallet_id",
+                "billing_date",
+                "billing_timezone",
+                name="uq_wallet_daily_usage_ledgers_wallet_date_tz",
+            ),
+        )
+
+    if not _index_exists("wallet_daily_usage_ledgers", "idx_wallet_daily_usage_wallet_date"):
+        op.create_index(
+            "idx_wallet_daily_usage_wallet_date",
+            "wallet_daily_usage_ledgers",
+            ["wallet_id", "billing_date"],
+        )
+    if not _index_exists("wallet_daily_usage_ledgers", "idx_wallet_daily_usage_date"):
+        op.create_index(
+            "idx_wallet_daily_usage_date",
+            "wallet_daily_usage_ledgers",
+            ["billing_date"],
+        )
+
+    if (
+        _table_exists("usage")
+        and all(
+            _column_exists("usage", col) for col in ["billing_status", "finalized_at", "wallet_id"]
+        )
+        and not _index_exists("usage", "idx_usage_billing_finalized_wallet")
+    ):
+        op.create_index(
+            "idx_usage_billing_finalized_wallet",
+            "usage",
+            ["billing_status", "finalized_at", "wallet_id"],
+        )
+
+
+def downgrade() -> None:
+    if _table_exists("usage") and _index_exists("usage", "idx_usage_billing_finalized_wallet"):
+        op.drop_index("idx_usage_billing_finalized_wallet", table_name="usage")
+
+    if _table_exists("wallet_daily_usage_ledgers"):
+        if _index_exists("wallet_daily_usage_ledgers", "idx_wallet_daily_usage_date"):
+            op.drop_index("idx_wallet_daily_usage_date", table_name="wallet_daily_usage_ledgers")
+        if _index_exists("wallet_daily_usage_ledgers", "idx_wallet_daily_usage_wallet_date"):
+            op.drop_index(
+                "idx_wallet_daily_usage_wallet_date",
+                table_name="wallet_daily_usage_ledgers",
+            )
+        op.drop_table("wallet_daily_usage_ledgers")

frontend/src/api/wallet.ts

@@ -60,6 +60,36 @@ export interface WalletTransactionsResponse extends WalletBalanceResponse {
   offset: number
 }
 
+export interface DailyUsageRecord {
+  id?: string | null
+  date: string | null
+  timezone?: string | null
+  total_cost: number
+  total_requests: number
+  input_tokens: number
+  output_tokens: number
+  cache_creation_tokens: number
+  cache_read_tokens: number
+  first_finalized_at?: string | null
+  last_finalized_at?: string | null
+  aggregated_at?: string | null
+  is_today: boolean
+}
+
+export type FlowItem =
+  | { type: 'transaction'; data: WalletTransaction }
+  | { type: 'daily_usage'; data: DailyUsageRecord }
+
+export interface WalletFlowResponse extends WalletBalanceResponse {
+  today_entry: DailyUsageRecord | null
+  items: FlowItem[]
+  total: number
+  limit: number
+  offset: number
+}
+
+export type TodayCostResponse = DailyUsageRecord
+
 export interface PaymentOrder {
   id: string
   order_no: string
@@ -131,6 +161,16 @@ export const walletApi = {
     return response.data
   },
 
+  async getFlow(params?: { limit?: number; offset?: number }): Promise<WalletFlowResponse> {
+    const response = await apiClient.get<WalletFlowResponse>('/api/wallet/flow', { params })
+    return response.data
+  },
+
+  async getTodayCost(): Promise<TodayCostResponse> {
+    const response = await apiClient.get<TodayCostResponse>('/api/wallet/today-cost')
+    return response.data
+  },
+
   async createRechargeOrder(payload: WalletRechargeCreateRequest): Promise<{
     order: PaymentOrder
     payment_instructions: Record<string, unknown>

frontend/src/utils/walletDisplay.ts

@@ -35,6 +35,21 @@ export function walletTransactionCategoryLabel(category: string | null | undefin
   return labels[category] || category
 }
 
+export function dailyUsageCategoryLabel(isToday = false): string {
+  return isToday ? '今日消费' : '每日消费'
+}
+
+export function formatTokenCount(value: number | null | undefined): string {
+  const amount = Number(value ?? 0)
+  if (amount >= 1_000_000) {
+    return `${(amount / 1_000_000).toFixed(amount >= 10_000_000 ? 0 : 1)}M`
+  }
+  if (amount >= 1_000) {
+    return `${(amount / 1_000).toFixed(amount >= 10_000 ? 0 : 1)}K`
+  }
+  return `${Math.round(amount)}`
+}
+
 export function walletTransactionReasonLabel(reasonCode: string | null | undefined): string {
   const labels: Record<string, string> = {
     topup_admin_manual: '人工充值',

frontend/src/views/user/WalletCenter.vue

@@ -270,46 +270,108 @@
                     </TableRow>
                   </TableHeader>
                   <TableBody>
-                    <TableRow
+                    <TableRow v-if="todayUsage">
-                      v-for="tx in transactions"
-                      :key="tx.id"
-                    >
                       <TableCell class="text-xs text-muted-foreground">
-                        {{ formatDateTime(tx.created_at) }}
+                        {{ todayUsage.date || '-' }}
                       </TableCell>
                       <TableCell>
                         <div class="space-y-1">
-                          <Badge
+                          <div class="flex items-center gap-2">
-                            variant="outline"
+                            <Badge
-                            class="font-mono"
+                              variant="outline"
-                          >
+                              class="font-mono border-amber-500/40 text-amber-700 dark:text-amber-300"
-                            {{ walletTransactionCategoryLabel(tx.category) }}
+                            >
-                          </Badge>
+                              {{ dailyUsageCategoryLabel(true) }}
+                            </Badge>
+                            <span class="inline-flex h-2 w-2 rounded-full bg-emerald-500 animate-pulse" />
+                            <span class="text-[11px] text-muted-foreground">
+                              Live
+                            </span>
+                          </div>
                           <div class="text-[11px] text-muted-foreground">
-                            {{ walletTransactionReasonLabel(tx.reason_code) }}
+                            {{ todayUsage.timezone || 'UTC' }}
                           </div>
                         </div>
                       </TableCell>
-                      <TableCell
+                      <TableCell class="text-rose-600 dark:text-rose-400">
-                        :class="tx.amount >= 0 ? 'text-emerald-600 dark:text-emerald-400' : 'text-rose-600 dark:text-rose-400'"
+                        -{{ todayUsage.total_cost.toFixed(4) }}
-                      >
-                        {{ tx.amount >= 0 ? '+' : '' }}{{ tx.amount.toFixed(4) }}
-                      </TableCell>
-                      <TableCell class="text-xs tabular-nums">
-                        {{ tx.balance_before.toFixed(4) }} → {{ tx.balance_after.toFixed(4) }}
                       </TableCell>
                       <TableCell class="text-xs text-muted-foreground">
-                        {{ tx.description || '-' }}
+                        按日汇总
+                      </TableCell>
+                      <TableCell class="text-xs text-muted-foreground">
+                        {{ todayUsage.total_requests }} 次请求 · {{ formatTokenCount(todayUsage.input_tokens) }} / {{ formatTokenCount(todayUsage.output_tokens) }} tokens
                       </TableCell>
                     </TableRow>
-                    <TableRow v-if="!loadingTransactions && transactions.length === 0">
+                    <template
+                      v-for="item in flowItems"
+                      :key="item.type === 'transaction' ? item.data.id : `daily-${item.data.id || item.data.date}`"
+                    >
+                      <TableRow v-if="item.type === 'transaction'">
+                        <TableCell class="text-xs text-muted-foreground">
+                          {{ formatDateTime(item.data.created_at) }}
+                        </TableCell>
+                        <TableCell>
+                          <div class="space-y-1">
+                            <Badge
+                              variant="outline"
+                              class="font-mono"
+                            >
+                              {{ walletTransactionCategoryLabel(item.data.category) }}
+                            </Badge>
+                            <div class="text-[11px] text-muted-foreground">
+                              {{ walletTransactionReasonLabel(item.data.reason_code) }}
+                            </div>
+                          </div>
+                        </TableCell>
+                        <TableCell
+                          :class="item.data.amount >= 0 ? 'text-emerald-600 dark:text-emerald-400' : 'text-rose-600 dark:text-rose-400'"
+                        >
+                          {{ item.data.amount >= 0 ? '+' : '' }}{{ item.data.amount.toFixed(4) }}
+                        </TableCell>
+                        <TableCell class="text-xs tabular-nums">
+                          {{ item.data.balance_before.toFixed(4) }} → {{ item.data.balance_after.toFixed(4) }}
+                        </TableCell>
+                        <TableCell class="text-xs text-muted-foreground">
+                          {{ item.data.description || '-' }}
+                        </TableCell>
+                      </TableRow>
+                      <TableRow v-else>
+                        <TableCell class="text-xs text-muted-foreground">
+                          {{ item.data.date || '-' }}
+                        </TableCell>
+                        <TableCell>
+                          <div class="space-y-1">
+                            <Badge
+                              variant="outline"
+                              class="font-mono border-amber-500/40 text-amber-700 dark:text-amber-300"
+                            >
+                              {{ dailyUsageCategoryLabel(false) }}
+                            </Badge>
+                            <div class="text-[11px] text-muted-foreground">
+                              {{ item.data.timezone || '-' }}
+                            </div>
+                          </div>
+                        </TableCell>
+                        <TableCell class="text-rose-600 dark:text-rose-400">
+                          -{{ item.data.total_cost.toFixed(4) }}
+                        </TableCell>
+                        <TableCell class="text-xs text-muted-foreground">
+                          按日汇总
+                        </TableCell>
+                        <TableCell class="text-xs text-muted-foreground">
+                          {{ item.data.total_requests }} 次请求 · {{ formatTokenCount(item.data.input_tokens) }} / {{ formatTokenCount(item.data.output_tokens) }} tokens
+                        </TableCell>
+                      </TableRow>
+                    </template>
+                    <TableRow v-if="!loadingTransactions && flowItems.length === 0">
                       <TableCell
                         colspan="5"
                         class="py-10"
                       >
                         <EmptyState
                           title="暂无资金流水"
-                          description="充值或退款后会在这里显示"
+                          description="充值、退款或消费后会在这里显示"
                         />
                       </TableCell>
                     </TableRow>
@@ -476,7 +538,7 @@
 </template>
 
 <script setup lang="ts">
-import { computed, onMounted, reactive, ref } from 'vue'
+import { computed, onBeforeUnmount, onMounted, reactive, ref, watch } from 'vue'
 import {
   Badge,
   Button,
@@ -505,15 +567,18 @@ import {
 import { EmptyState, LoadingState } from '@/components/common'
 import {
   walletApi,
+  type DailyUsageRecord,
+  type FlowItem,
   type PaymentOrder,
   type RefundRequest,
   type WalletBalanceResponse,
-  type WalletTransaction,
 } from '@/api/wallet'
 import { useToast } from '@/composables/useToast'
 import { parseApiError } from '@/utils/errorParser'
 import { log } from '@/utils/logger'
 import {
+  dailyUsageCategoryLabel,
+  formatTokenCount,
   formatWalletCurrency as formatCurrency,
   paymentMethodLabel,
   paymentStatusBadge,
@@ -542,7 +607,8 @@ const submittingRefund = ref(false)
 const walletBalance = ref<WalletBalanceResponse | null>(null)
 const latestRecharge = ref<{ order: PaymentOrder; payment_instructions: Record<string, unknown> } | null>(null)
 
-const transactions = ref<WalletTransaction[]>([])
+const flowItems = ref<FlowItem[]>([])
+const todayUsage = ref<DailyUsageRecord | null>(null)
 const txTotal = ref(0)
 const txPage = ref(1)
 const txPageSize = ref(20)
@@ -558,6 +624,7 @@ const refundPage = ref(1)
 const refundPageSize = ref(20)
 
 const activeTab = ref('transactions')
+let todayCostPollTimer: ReturnType<typeof setInterval> | null = null
 
 const rechargeForm = reactive({
   amount_usd: 10,
@@ -576,18 +643,30 @@ const refundableOrders = computed(() =>
 )
 
 onMounted(async () => {
+  document.addEventListener('visibilitychange', handleVisibilityChange)
   try {
     await Promise.all([
       loadBalance(),
       loadTransactions(),
+      loadTodayCost(),
       loadOrders(),
       loadRefunds(),
     ])
+    syncTodayCostPolling()
   } finally {
     loadingInitial.value = false
   }
 })
 
+onBeforeUnmount(() => {
+  stopTodayCostPolling()
+  document.removeEventListener('visibilitychange', handleVisibilityChange)
+})
+
+watch(activeTab, () => {
+  syncTodayCostPolling()
+})
+
 async function loadBalance() {
   walletBalance.value = await walletApi.getBalance()
 }
@@ -596,9 +675,10 @@ async function loadTransactions() {
   loadingTransactions.value = true
   try {
     const offset = (txPage.value - 1) * txPageSize.value
-    const resp = await walletApi.getTransactions({ limit: txPageSize.value, offset })
+    const resp = await walletApi.getFlow({ limit: txPageSize.value, offset })
-    transactions.value = resp.items
+    flowItems.value = resp.items
     txTotal.value = resp.total
+    todayUsage.value = resp.today_entry
   } catch (error) {
     log.error('加载钱包流水失败:', error)
     showError(parseApiError(error, '加载钱包流水失败'))
@@ -607,6 +687,39 @@ async function loadTransactions() {
   }
 }
 
+async function loadTodayCost() {
+  try {
+    todayUsage.value = await walletApi.getTodayCost()
+  } catch (error) {
+    log.error('加载今日消费失败:', error)
+  }
+}
+
+function syncTodayCostPolling() {
+  if (activeTab.value === 'transactions' && !document.hidden) {
+    startTodayCostPolling()
+  } else {
+    stopTodayCostPolling()
+  }
+}
+
+function startTodayCostPolling() {
+  if (todayCostPollTimer) return
+  todayCostPollTimer = setInterval(() => {
+    void loadTodayCost()
+  }, 20_000)
+}
+
+function stopTodayCostPolling() {
+  if (!todayCostPollTimer) return
+  clearInterval(todayCostPollTimer)
+  todayCostPollTimer = null
+}
+
+function handleVisibilityChange() {
+  syncTodayCostPolling()
+}
+
 async function loadOrders() {
   loadingOrders.value = true
   try {
@@ -688,7 +801,7 @@ async function submitRefund() {
     refundForm.amount_usd = 0
     refundForm.payment_order_id = '__none__'
     refundForm.reason = ''
-    await Promise.all([loadRefunds(), loadBalance(), loadOrders(), loadTransactions()])
+    await Promise.all([loadRefunds(), loadBalance(), loadOrders(), loadTransactions(), loadTodayCost()])
     activeTab.value = 'refunds'
   } catch (error) {
     log.error('提交退款申请失败:', error)

src/api/base/pipeline.py

@@ -37,6 +37,7 @@ QUIET_POLLING_PATHS: set[str] = {
     "/api/admin/usage/stats",
     "/api/admin/usage/aggregation/stats",
     "/api/admin/health/status",
+    "/api/wallet/today-cost",
 }
 
 

src/api/handlers/base/stream_telemetry.py

@@ -492,6 +492,9 @@ class StreamTelemetryRecorder:
 
             usage = db.query(Usage).filter(Usage.request_id == request_id).first()
             if usage:
+                if getattr(usage, "billing_status", None) in {"settled", "void"}:
+                    logger.debug("[{}] Usage 已终态，跳过快速状态更新: {}", self.request_id, status)
+                    return
                 setattr(usage, "status", status)
                 setattr(usage, "status_code", status_code)
                 setattr(usage, "response_time_ms", response_time_ms)

src/api/serializers/__init__.py

@@ -5,6 +5,7 @@ from .wallet_payment import (
     serialize_admin_wallet_transaction,
     serialize_payment_callback,
     serialize_payment_order,
+    serialize_wallet_daily_usage,
     serialize_wallet_payload,
     serialize_wallet_refund,
     serialize_wallet_transaction,
@@ -17,6 +18,7 @@ __all__ = [
     "serialize_admin_wallet_transaction",
     "serialize_payment_callback",
     "serialize_payment_order",
+    "serialize_wallet_daily_usage",
     "serialize_wallet_payload",
     "serialize_wallet_refund",
     "serialize_wallet_transaction",

src/api/serializers/wallet_payment.py

@@ -7,9 +7,10 @@ from src.models.database import (
     PaymentOrder,
     RefundRequest,
     Wallet,
+    WalletDailyUsageLedger,
     WalletTransaction,
 )
-from src.services.wallet import WalletService
+from src.services.wallet import WalletDailyUsageSnapshot, WalletService
 
 
 def safe_gateway_response(raw: dict[str, Any] | None) -> dict[str, Any]:
@@ -124,6 +125,27 @@ def serialize_wallet_transaction(tx: WalletTransaction) -> dict[str, Any]:
     }
 
 
+def serialize_wallet_daily_usage(
+    ledger: WalletDailyUsageLedger | WalletDailyUsageSnapshot,
+) -> dict[str, Any]:
+    billing_date = getattr(ledger, "billing_date", None)
+    return {
+        "id": getattr(ledger, "id", None),
+        "date": billing_date.isoformat() if billing_date is not None else None,
+        "timezone": getattr(ledger, "billing_timezone", None),
+        "total_cost": float(getattr(ledger, "total_cost_usd", 0) or 0),
+        "total_requests": int(getattr(ledger, "total_requests", 0) or 0),
+        "input_tokens": int(getattr(ledger, "input_tokens", 0) or 0),
+        "output_tokens": int(getattr(ledger, "output_tokens", 0) or 0),
+        "cache_creation_tokens": int(getattr(ledger, "cache_creation_tokens", 0) or 0),
+        "cache_read_tokens": int(getattr(ledger, "cache_read_tokens", 0) or 0),
+        "first_finalized_at": getattr(ledger, "first_finalized_at", None),
+        "last_finalized_at": getattr(ledger, "last_finalized_at", None),
+        "aggregated_at": getattr(ledger, "aggregated_at", None),
+        "is_today": bool(getattr(ledger, "is_today", False)),
+    }
+
+
 def serialize_wallet_refund(refund: RefundRequest) -> dict[str, Any]:
     return {
         "id": refund.id,

src/api/wallet/routes.py

@@ -3,9 +3,10 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import date, datetime, timezone
 from typing import Any
 from uuid import uuid4
+from zoneinfo import ZoneInfo
 
 from fastapi import APIRouter, Depends, Query, Request
 from pydantic import BaseModel, Field, ValidationError
@@ -18,15 +19,22 @@ from src.api.base.pipeline import ApiRequestPipeline
 from src.api.serializers import (
     safe_gateway_response,
     serialize_payment_order,
+    serialize_wallet_daily_usage,
     serialize_wallet_payload,
     serialize_wallet_refund,
     serialize_wallet_transaction,
 )
 from src.core.exceptions import InvalidRequestException, NotFoundException, translate_pydantic_error
 from src.database import get_db
-from src.models.database import PaymentOrder, RefundRequest, Wallet, WalletTransaction
+from src.models.database import (
+    PaymentOrder,
+    RefundRequest,
+    Wallet,
+    WalletDailyUsageLedger,
+    WalletTransaction,
+)
 from src.services.payment import PaymentService
-from src.services.wallet import WalletService
+from src.services.wallet import WalletDailyUsageLedgerService, WalletService
 
 router = APIRouter(prefix="/api/wallet", tags=["Wallet"])
 pipeline = ApiRequestPipeline()
@@ -61,6 +69,43 @@ def _build_refund_no() -> str:
     return f"rf_{ts}_{uuid4().hex[:8]}"
 
 
+def _resolve_user_wallet(db: Session, user: Any) -> Wallet | None:
+    existing_wallet = WalletService.get_wallet(db, user_id=user.id)
+    wallet = existing_wallet or WalletService.get_or_create_wallet(db, user=user)
+    if wallet is not None and existing_wallet is None:
+        db.commit()
+        db.refresh(wallet)
+    return wallet
+
+
+def _flow_sort_key(item: dict[str, Any], billing_tz: ZoneInfo) -> tuple[date, int, datetime]:
+    item_type = item.get("type")
+    data = item.get("data") or {}
+    if item_type == "daily_usage":
+        raw_date = data.get("date")
+        billing_date = (
+            date.fromisoformat(raw_date) if isinstance(raw_date, str) and raw_date else date.min
+        )
+        sort_dt = (
+            data.get("last_finalized_at")
+            or data.get("aggregated_at")
+            or datetime.min.replace(tzinfo=timezone.utc)
+        )
+        if isinstance(sort_dt, str):
+            sort_dt = datetime.fromisoformat(sort_dt.replace("Z", "+00:00"))
+        return billing_date, 1, sort_dt
+
+    created_at = data.get("created_at")
+    if isinstance(created_at, str):
+        created_at = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+    if not isinstance(created_at, datetime):
+        created_at = datetime.min.replace(tzinfo=timezone.utc)
+    elif created_at.tzinfo is None:
+        created_at = created_at.replace(tzinfo=timezone.utc)
+    local_date = created_at.astimezone(billing_tz).date()
+    return local_date, 0, created_at
+
+
 @router.get("/balance")
 async def get_wallet_balance(request: Request, db: Session = Depends(get_db)) -> Any:
     adapter = WalletBalanceAdapter()
@@ -78,6 +123,23 @@ async def list_wallet_transactions(
     return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
 
 
+@router.get("/flow")
+async def get_wallet_flow(
+    request: Request,
+    limit: int = Query(50, ge=1, le=200),
+    offset: int = Query(0, ge=0, le=5000),
+    db: Session = Depends(get_db),
+) -> Any:
+    adapter = WalletFlowAdapter(limit=limit, offset=offset)
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
+@router.get("/today-cost")
+async def get_wallet_today_cost(request: Request, db: Session = Depends(get_db)) -> Any:
+    adapter = WalletTodayCostAdapter()
+    return await pipeline.run(adapter=adapter, http_request=request, db=db, mode=adapter.mode)
+
+
 @router.post("/recharge")
 async def create_recharge_order(request: Request, db: Session = Depends(get_db)) -> Any:
     adapter = WalletRechargeCreateAdapter()
@@ -135,8 +197,7 @@ class WalletTransactionsAdapter(AuthenticatedApiAdapter):
         if user is None:
             raise InvalidRequestException("未登录")
 
-        existing_wallet = WalletService.get_wallet(db, user_id=user.id)
+        wallet = _resolve_user_wallet(db, user)
-        wallet = existing_wallet or WalletService.get_or_create_wallet(db, user=user)
         if wallet is None:
             return {
                 "items": [],
@@ -146,10 +207,6 @@ class WalletTransactionsAdapter(AuthenticatedApiAdapter):
                 **serialize_wallet_payload(None),
             }
 
-        if existing_wallet is None:
-            db.commit()
-            db.refresh(wallet)
-
         base_query = db.query(WalletTransaction).filter(WalletTransaction.wallet_id == wallet.id)
         total = base_query.count()
         items = (
@@ -168,6 +225,80 @@ class WalletTransactionsAdapter(AuthenticatedApiAdapter):
         }
 
 
+@dataclass
+class WalletFlowAdapter(AuthenticatedApiAdapter):
+    limit: int
+    offset: int
+
+    async def handle(self, context: ApiRequestContext) -> dict[str, Any]:
+        db = context.db
+        user = context.user
+        if user is None:
+            raise InvalidRequestException("未登录")
+
+        wallet = _resolve_user_wallet(db, user)
+        if wallet is None:
+            return {
+                "today_entry": None,
+                "items": [],
+                "total": 0,
+                "limit": self.limit,
+                "offset": self.offset,
+                **serialize_wallet_payload(None),
+            }
+
+        today_entry = WalletDailyUsageLedgerService.get_today_snapshot(db, wallet.id)
+        billing_tz = WalletDailyUsageLedgerService.get_timezone(today_entry.billing_timezone)
+        fetch_size = min(self.offset + self.limit, 5200)
+
+        tx_query = db.query(WalletTransaction).filter(WalletTransaction.wallet_id == wallet.id)
+        tx_total = tx_query.count()
+        tx_items = tx_query.order_by(WalletTransaction.created_at.desc()).limit(fetch_size).all()
+
+        daily_query = db.query(WalletDailyUsageLedger).filter(
+            WalletDailyUsageLedger.wallet_id == wallet.id,
+            WalletDailyUsageLedger.billing_timezone == today_entry.billing_timezone,
+            WalletDailyUsageLedger.billing_date < today_entry.billing_date,
+        )
+        daily_total = daily_query.count()
+        daily_items = (
+            daily_query.order_by(WalletDailyUsageLedger.billing_date.desc()).limit(fetch_size).all()
+        )
+
+        merged = [
+            {"type": "transaction", "data": serialize_wallet_transaction(item)} for item in tx_items
+        ] + [
+            {"type": "daily_usage", "data": serialize_wallet_daily_usage(item)}
+            for item in daily_items
+        ]
+        merged.sort(key=lambda item: _flow_sort_key(item, billing_tz), reverse=True)
+        paged = merged[self.offset : self.offset + self.limit]
+
+        return {
+            "today_entry": serialize_wallet_daily_usage(today_entry),
+            "items": paged,
+            "total": tx_total + daily_total,
+            "limit": self.limit,
+            "offset": self.offset,
+            **serialize_wallet_payload(wallet),
+        }
+
+
+class WalletTodayCostAdapter(AuthenticatedApiAdapter):
+    async def handle(self, context: ApiRequestContext) -> dict[str, Any]:
+        db = context.db
+        user = context.user
+        if user is None:
+            raise InvalidRequestException("未登录")
+
+        wallet = _resolve_user_wallet(db, user)
+        snapshot = WalletDailyUsageLedgerService.get_today_snapshot(
+            db,
+            wallet.id if wallet is not None else None,
+        )
+        return serialize_wallet_daily_usage(snapshot)
+
+
 class WalletBalanceAdapter(AuthenticatedApiAdapter):
     async def handle(self, context: ApiRequestContext) -> dict[str, Any]:
         db = context.db
@@ -175,15 +306,10 @@ class WalletBalanceAdapter(AuthenticatedApiAdapter):
         if user is None:
             raise InvalidRequestException("未登录")
 
-        existing_wallet = WalletService.get_wallet(db, user_id=user.id)
+        wallet = _resolve_user_wallet(db, user)
-        wallet = existing_wallet or WalletService.get_or_create_wallet(db, user=user)
         if wallet is None:
             return serialize_wallet_payload(None)
 
-        if existing_wallet is None:
-            db.commit()
-            db.refresh(wallet)
-
         pending_refunds = (
             db.query(RefundRequest)
             .filter(

src/models/database.py

@@ -7,7 +7,7 @@ from __future__ import annotations
 import hashlib
 import secrets
 import uuid
-from datetime import datetime, timezone
+from datetime import date, datetime, timezone
 from enum import Enum as PyEnum
 from typing import Any, ClassVar
 
@@ -18,6 +18,7 @@ from sqlalchemy import (
     Boolean,
     CheckConstraint,
     Column,
+    Date,
     DateTime,
     Enum,
     Float,
@@ -413,10 +414,10 @@ class Usage(Base):
     status = Column(String(20), default="completed", nullable=False, index=True)
 
     # 结算状态（与 status 解耦）
-    # - pending: 等待结算（任务未完成 / 流式未结束）
+    # - pending: 等待结算（请求已创建，但账务尚未进入最终状态）
-    # - settled: 已结算（cost 已写入，可能 > 0 或 = 0）
+    # - settled: 已结算（cost 已写入，且钱包侧结算动作已完成）
-    # - void: 作废（不收费，如任务未开始就取消）
+    # - void: 作废（明确不收费）
-    billing_status = Column(String(20), default="settled", nullable=False, index=True)
+    billing_status = Column(String(20), default="pending", nullable=False, index=True)
     finalized_at = Column(DateTime(timezone=True), nullable=True)  # 结算完成时间（可选）
     wallet_balance_before = Column(Numeric(20, 8), nullable=True)  # 结算前可用总余额快照
     wallet_balance_after = Column(Numeric(20, 8), nullable=True)  # 结算后可用总余额快照
@@ -556,6 +557,9 @@ class Wallet(Base):
     transactions = relationship(
         "WalletTransaction", back_populates="wallet", cascade="all, delete-orphan"
     )
+    daily_usage_ledgers = relationship(
+        "WalletDailyUsageLedger", back_populates="wallet", cascade="all, delete-orphan"
+    )
     payment_orders = relationship("PaymentOrder", back_populates="wallet")
     refund_requests = relationship("RefundRequest", back_populates="wallet")
 
@@ -609,6 +613,50 @@ class WalletTransaction(Base):
     operator = relationship("User")
 
 
+class WalletDailyUsageLedger(Base):
+    """钱包按天汇总的消费流水投影。"""
+
+    __tablename__ = "wallet_daily_usage_ledgers"
+    __table_args__ = (
+        UniqueConstraint(
+            "wallet_id",
+            "billing_date",
+            "billing_timezone",
+            name="uq_wallet_daily_usage_ledgers_wallet_date_tz",
+        ),
+        Index("idx_wallet_daily_usage_wallet_date", "wallet_id", "billing_date"),
+        Index("idx_wallet_daily_usage_date", "billing_date"),
+    )
+
+    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
+    wallet_id = Column(String(36), ForeignKey("wallets.id", ondelete="CASCADE"), nullable=False)
+
+    billing_date = Column(Date, nullable=False)
+    billing_timezone = Column(String(64), nullable=False, default="Asia/Shanghai")
+
+    total_cost_usd = Column(Numeric(20, 8), nullable=False, default=0)
+    total_requests = Column(Integer, nullable=False, default=0)
+    input_tokens = Column(BigInteger, nullable=False, default=0)
+    output_tokens = Column(BigInteger, nullable=False, default=0)
+    cache_creation_tokens = Column(BigInteger, nullable=False, default=0)
+    cache_read_tokens = Column(BigInteger, nullable=False, default=0)
+
+    first_finalized_at = Column(DateTime(timezone=True), nullable=True)
+    last_finalized_at = Column(DateTime(timezone=True), nullable=True)
+    aggregated_at = Column(DateTime(timezone=True), nullable=False)
+    created_at = Column(
+        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False
+    )
+    updated_at = Column(
+        DateTime(timezone=True),
+        default=lambda: datetime.now(timezone.utc),
+        onupdate=lambda: datetime.now(timezone.utc),
+        nullable=False,
+    )
+
+    wallet = relationship("Wallet", back_populates="daily_usage_ledgers")
+
+
 class PaymentOrder(Base):
     """充值订单"""
 

src/services/system/maintenance_scheduler.py

@@ -32,6 +32,7 @@ from src.services.system.config import SystemConfigService
 from src.services.system.scheduler import get_scheduler
 from src.services.system.stats_aggregator import StatsAggregatorService
 from src.services.user.apikey import ApiKeyService
+from src.services.wallet import WalletDailyUsageLedgerService
 from src.utils.compression import compress_json
 
 
@@ -45,6 +46,7 @@ class MaintenanceScheduler:
         self.running = False
         self._interval_tasks = []
         self._stats_aggregation_lock = asyncio.Lock()
+        self._wallet_daily_usage_lock = asyncio.Lock()
 
     def _get_checkin_time(self) -> tuple[int, int]:
         """获取签到任务的执行时间
@@ -144,6 +146,13 @@ class MaintenanceScheduler:
             name="统计小时数据聚合",
             timezone="UTC",
         )
+        scheduler.add_cron_job(
+            self._scheduled_wallet_daily_usage_aggregation,
+            hour=0,
+            minute=10,
+            job_id="wallet_daily_usage_aggregation",
+            name="钱包每日消费汇总",
+        )
         # 清理任务 - 凌晨 3 点执行
         scheduler.add_cron_job(
             self._scheduled_cleanup,
@@ -268,6 +277,10 @@ class MaintenanceScheduler:
         """统计聚合任务（定时调用）"""
         await self._perform_stats_aggregation(backfill=backfill)
 
+    async def _scheduled_wallet_daily_usage_aggregation(self) -> None:
+        """钱包每日消费汇总任务（定时调用）"""
+        await self._perform_wallet_daily_usage_aggregation()
+
     async def _scheduled_hourly_stats_aggregation(self) -> None:
         """小时统计聚合任务（定时调用）"""
         await self._perform_hourly_stats_aggregation()
@@ -499,6 +512,32 @@ class MaintenanceScheduler:
             finally:
                 db.close()
 
+    async def _perform_wallet_daily_usage_aggregation(self) -> None:
+        if self._wallet_daily_usage_lock.locked():
+            logger.info("钱包每日消费汇总任务正在运行，跳过本次触发")
+            return
+
+        async with self._wallet_daily_usage_lock:
+            db = create_session()
+            try:
+                logger.info("开始执行钱包每日消费汇总...")
+                billing_today = WalletDailyUsageLedgerService.get_today_billing_date()
+                billing_yesterday = billing_today - timedelta(days=1)
+                affected = WalletDailyUsageLedgerService.aggregate_day(db, billing_yesterday)
+                logger.info(
+                    "钱包每日消费汇总完成: date={}, wallets={}",
+                    billing_yesterday.isoformat(),
+                    affected,
+                )
+            except Exception as e:
+                logger.exception("钱包每日消费汇总任务执行失败: {}", e)
+                try:
+                    db.rollback()
+                except Exception:
+                    pass
+            finally:
+                db.close()
+
     async def _perform_hourly_stats_aggregation(self) -> None:
         """执行小时统计聚合任务"""
         db = create_session()

src/services/task/video/billing.py

@@ -92,6 +92,7 @@ class VideoTaskBillingService:
                 provider_api_key_id=getattr(task, "key_id", None),
                 status="completed" if task.status == "completed" else "failed",
                 target_model=None,
+                finalized_at=getattr(task, "completed_at", None),
             )
             return True
         except Exception as exc:
@@ -122,6 +123,8 @@ class VideoTaskBillingService:
         if not request_id:
             return False
 
+        # Advisory check（无锁）：仅用于快速跳过，实际状态转换由 update_settled_billing 的
+        # with_for_update() 保证原子性。
         existing = self.db.query(Usage).filter(Usage.request_id == request_id).first()
         if not existing:
             logger.warning(
@@ -131,12 +134,12 @@ class VideoTaskBillingService:
             )
             return await self._create_fallback_usage_for_video_task(task, request_id)
 
-        metadata = existing.request_metadata or {}
+        if getattr(existing, "billing_status", None) != "pending":
-        if metadata.get("billing_updated_at"):
             logger.debug(
-                "Video task billing already updated: task_id={} request_id={}",
+                "Skip video task billing finalize because Usage is already terminal: task_id={} request_id={} billing_status={}",
                 getattr(task, "id", None),
                 request_id,
+                getattr(existing, "billing_status", None),
             )
             return False
 
@@ -300,6 +303,7 @@ class VideoTaskBillingService:
                     "field": "video_tasks.request_metadata.poll_raw_response",
                 },
             },
+            finalized_at=getattr(task, "completed_at", None),
         )
 
         if updated:

src/services/task/video/cancel.py

@@ -44,6 +44,19 @@ class VideoTaskCancelService:
         from src.services.provider.auth import get_provider_auth
         from src.services.provider.transport import build_provider_url
 
+        current_status = str(getattr(task, "status", "") or "")
+        non_cancellable_statuses = {
+            VideoStatus.COMPLETED.value,
+            VideoStatus.FAILED.value,
+            VideoStatus.CANCELLED.value,
+            VideoStatus.EXPIRED.value,
+        }
+        if current_status in non_cancellable_statuses:
+            raise HTTPException(
+                status_code=409,
+                detail=f"Task cannot be cancelled in status: {current_status}",
+            )
+
         external_task_id = getattr(task, "external_task_id", None)
         if not external_task_id:
             raise HTTPException(status_code=500, detail="Task missing external_task_id")
@@ -122,8 +135,10 @@ class VideoTaskCancelService:
                 detail=f"Cancel not supported for provider format: {provider_format}",
             )
 
+        now = datetime.now(timezone.utc)
         task.status = VideoStatus.CANCELLED.value
-        task.updated_at = datetime.now(timezone.utc)
+        task.completed_at = getattr(task, "completed_at", None) or now
+        task.updated_at = now
 
         # Void Usage (no charge)
         try:
@@ -131,12 +146,13 @@ class VideoTaskCancelService:
                 self.db,
                 request_id=task.request_id,
                 reason="cancelled_by_user",
+                finalized_at=task.completed_at,
             )
             if not voided:
-                UsageService.void_settled(
+                logger.warning(
-                    self.db,
+                    "Skip voiding video usage because billing is already terminal: task_id={} request_id={}",
-                    request_id=task.request_id,
+                    getattr(task, "id", task_id),
-                    reason="cancelled_by_user",
+                    getattr(task, "request_id", None),
                 )
         except Exception as exc:
             logger.warning(

src/services/task/video/poller_adapter.py

@@ -260,6 +260,19 @@ class VideoTaskPollerAdapter:
                 logger.warning("Task {} disappeared during poll update", task_id)
                 return
 
+            if task.status in {
+                VideoStatus.COMPLETED.value,
+                VideoStatus.FAILED.value,
+                VideoStatus.CANCELLED.value,
+                VideoStatus.EXPIRED.value,
+            }:
+                logger.debug(
+                    "Skip poll update for terminal task {} with status {}",
+                    task_id,
+                    task.status,
+                )
+                return
+
             if error_exception is not None and ctx is not None:
                 # HTTP 请求失败（需要 ctx 来计算 backoff）
                 self._handle_poll_error(task, error_exception, ctx)
@@ -382,6 +395,17 @@ class VideoTaskPollerAdapter:
         """
         兼容入口：复用三阶段轮询流程，避免维护重复逻辑。
         """
+        if task.status in {
+            VideoStatus.COMPLETED.value,
+            VideoStatus.FAILED.value,
+            VideoStatus.CANCELLED.value,
+            VideoStatus.EXPIRED.value,
+        }:
+            logger.debug(
+                "Skip legacy poll for terminal task {} with status {}", task.id, task.status
+            )
+            return
+
         ctx_or_result = await self.prepare_poll_context(db, task)
 
         if isinstance(ctx_or_result, InternalVideoPollResult):

src/services/usage/active_requests.py

@@ -130,7 +130,7 @@ class UsageActiveRequestsMixin:
 
         while True:
             stale_requests = (
-                db.query(Usage.id, Usage.request_id, Usage.status)
+                db.query(Usage.id, Usage.request_id, Usage.status, Usage.billing_status)
                 .filter(
                     Usage.status.in_(["pending", "streaming"]),
                     Usage.created_at < cutoff_time,
@@ -143,13 +143,13 @@ class UsageActiveRequestsMixin:
             if not stale_requests:
                 break
 
-            stale_request_ids = [request_id for _, request_id, _ in stale_requests if request_id]
+            stale_request_ids = [request_id for _, request_id, _, _ in stale_requests if request_id]
             completed_request_ids = cls._find_completed_request_ids(db, stale_request_ids)
 
             usage_updates = []
             failed_request_ids: list[str] = []
 
-            for usage_id, request_id, old_status in stale_requests:
+            for usage_id, request_id, old_status, billing_status in stale_requests:
                 if request_id and request_id in completed_request_ids:
                     usage_updates.append(
                         {
@@ -161,16 +161,22 @@ class UsageActiveRequestsMixin:
                     )
                     recovered_count += 1
                 else:
-                    usage_updates.append(
+                    entry: dict[str, Any] = {
-                        {
+                        "id": usage_id,
-                            "id": usage_id,
+                        "status": "failed",
-                            "status": "failed",
+                        "status_code": 504,
-                            "status_code": 504,
+                        "error_message": (
-                            "error_message": (
+                            f"请求超时: 状态 '{old_status}' 超过 {timeout_minutes} 分钟未完成"
-                                f"请求超时: 状态 '{old_status}' 超过 {timeout_minutes} 分钟未完成"
+                        ),
-                            ),
+                    }
-                        }
+                    if billing_status == "pending":
-                    )
+                        entry["billing_status"] = "void"
+                        entry["finalized_at"] = now
+                        entry["total_cost_usd"] = 0.0
+                        entry["request_cost_usd"] = 0.0
+                        entry["actual_total_cost_usd"] = 0.0
+                        entry["actual_request_cost_usd"] = 0.0
+                    usage_updates.append(entry)
                     failed_count += 1
                     if request_id:
                         failed_request_ids.append(request_id)

src/services/usage/consumer_streams.py

@@ -11,6 +11,7 @@ import json
 import os
 import socket
 import time
+from datetime import datetime, timezone
 from typing import Any
 
 from redis.exceptions import ConnectionError as RedisConnectionError
@@ -58,6 +59,10 @@ def _event_to_record(event: UsageEvent) -> dict[str, Any]:
     elif event.event_type == UsageEventType.CANCELLED:
         status = "cancelled"
 
+    finalized_at = None
+    if event.timestamp_ms > 0:
+        finalized_at = datetime.fromtimestamp(event.timestamp_ms / 1000, tz=timezone.utc)
+
     return {
         "request_id": event.request_id,
         "user_id": data.get("user_id"),
@@ -95,6 +100,7 @@ def _event_to_record(event: UsageEvent) -> dict[str, Any]:
         "provider_api_key_id": data.get("provider_api_key_id"),
         "status": status,
         "target_model": data.get("target_model"),
+        "finalized_at": finalized_at,
     }
 
 
@@ -491,6 +497,11 @@ class UsageQueueConsumer:
                 provider_api_key_id=data.get("provider_api_key_id"),
                 status=status,
                 target_model=data.get("target_model"),
+                finalized_at=(
+                    datetime.fromtimestamp(event.timestamp_ms / 1000, tz=timezone.utc)
+                    if event.timestamp_ms > 0
+                    else None
+                ),
             )
         finally:
             if own_session:

src/services/usage/lifecycle.py

@@ -18,6 +18,12 @@ from src.services.wallet import WalletService
 class UsageLifecycleMixin:
     """使用记录生命周期管理方法"""
 
+    @staticmethod
+    def _is_billing_terminal(usage: Usage | None) -> bool:
+        return bool(
+            usage is not None and getattr(usage, "billing_status", None) in {"settled", "void"}
+        )
+
     @classmethod
     def begin_pending_usage(
         cls,
@@ -152,6 +158,7 @@ class UsageLifecycleMixin:
         response_time_ms: int | None = None,
         billing_snapshot: dict[str, Any] | None = None,
         extra_metadata: dict[str, Any] | None = None,
+        finalized_at: datetime | None = None,
     ) -> bool:
         """
         并发安全的幂等 finalize（settled）。
@@ -160,7 +167,7 @@ class UsageLifecycleMixin:
         - 仅当 billing_status='pending' 时才会生效（rowcount==1）
         - 不在本方法内 commit，由调用方决定事务提交时机
         """
-        now = datetime.now(timezone.utc)
+        now = finalized_at or datetime.now(timezone.utc)
         cost = to_money_decimal(total_cost_usd)
         request_cost = to_money_decimal(request_cost_usd) if request_cost_usd is not None else cost
 
@@ -197,6 +204,7 @@ class UsageLifecycleMixin:
         *,
         reason: str | None = None,
         status_code: int = 499,
+        finalized_at: datetime | None = None,
     ) -> bool:
         """
         并发安全的幂等 finalize（void，不收费）。
@@ -205,7 +213,7 @@ class UsageLifecycleMixin:
         - 仅当 billing_status='pending' 时才会生效（rowcount==1）
         - 不在本方法内 commit，由调用方决定事务提交时机
         """
-        now = datetime.now(timezone.utc)
+        now = finalized_at or datetime.now(timezone.utc)
         usage = db.query(Usage).filter(Usage.request_id == request_id).with_for_update().first()
         if not usage or usage.billing_status != "pending":
             return False
@@ -214,6 +222,8 @@ class UsageLifecycleMixin:
         usage.finalized_at = now
         usage.total_cost_usd = to_money_decimal(0)
         usage.request_cost_usd = to_money_decimal(0)
+        usage.actual_total_cost_usd = to_money_decimal(0)
+        usage.actual_request_cost_usd = to_money_decimal(0)
         usage.status = "cancelled"
         usage.status_code = status_code
         usage.error_message = reason
@@ -313,28 +323,24 @@ class UsageLifecycleMixin:
         response_time_ms: int | None = None,
         billing_snapshot: dict[str, Any] | None = None,
         extra_metadata: dict[str, Any] | None = None,
+        finalized_at: datetime | None = None,
     ) -> bool:
         """
         写入异步任务最终账单（轮询完成后调用）。
 
         语义：
-        - 正常路径：pending -> settled / void（首次最终结算）
+        - 仅允许 pending -> settled / void（首次最终结算）
-        - 补写路径：已写入 0 成本但尚未扣钱包的记录，可补写一次最终值
+        - settled / void 一旦进入即不可再修改
-        - 已 void 的记录不可再结算
-        - 已扣钱包（wallet_balance_after 已存在）的记录不可重复扣费
 
         约定：
         - 不在本方法内 commit，由调用方决定事务提交时机
         """
-        now = datetime.now(timezone.utc)
+        now = finalized_at or datetime.now(timezone.utc)
         cost = to_money_decimal(total_cost_usd)
         request_cost = to_money_decimal(request_cost_usd) if request_cost_usd is not None else cost
 
         usage = db.query(Usage).filter(Usage.request_id == request_id).with_for_update().first()
-        if not usage or usage.billing_status == "void":
+        if not usage or usage.billing_status != "pending":
-            return False
-
-        if usage.billing_status == "settled" and usage.wallet_balance_after is not None:
             return False
 
         usage.total_cost_usd = cost
@@ -345,7 +351,7 @@ class UsageLifecycleMixin:
             usage.error_message = error_message
         if response_time_ms is not None:
             usage.response_time_ms = response_time_ms
-        usage.finalized_at = usage.finalized_at or now
+        usage.finalized_at = now
         if cost > 0:
             usage.billing_status = "settled"
             WalletService.apply_usage_charge(db, usage=usage, amount_usd=cost)
@@ -373,32 +379,15 @@ class UsageLifecycleMixin:
         status_code: int = 499,
     ) -> bool:
         """
-        将已结算的记录作废（用于异步任务取消）。
+        已废弃：settled 为账务终态，不允许再回滚为 void。
-
-        与 finalize_void 不同：
-        - finalize_void: pending -> void（未结算时作废）
-        - void_settled: settled -> void（已结算后取消，费用归零）
-
-        约定：
-        - 仅当 billing_status='settled' 时才会生效
-        - 不在本方法内 commit，由调用方决定事务提交时机
         """
-        now = datetime.now(timezone.utc)
+        logger.warning(
-        usage = db.query(Usage).filter(Usage.request_id == request_id).with_for_update().first()
+            "void_settled is deprecated and ignored: request_id={}, reason={}, status_code={}",
-        if not usage or usage.billing_status != "settled":
+            request_id,
-            return False
+            reason,
-        if usage.wallet_balance_after is not None and to_money_decimal(usage.total_cost_usd) > 0:
+            status_code,
-            # 已实际扣费的记录当前不做自动回滚，避免 silent inconsistency。
+        )
-            return False
+        return False
-
-        usage.billing_status = "void"
-        usage.finalized_at = now
-        usage.total_cost_usd = to_money_decimal(0)
-        usage.request_cost_usd = to_money_decimal(0)
-        usage.status = "cancelled"
-        usage.status_code = status_code
-        usage.error_message = reason
-        return True
 
     @classmethod
     def update_usage_status(
@@ -453,6 +442,15 @@ class UsageLifecycleMixin:
             logger.warning("未找到 request_id={} 的使用记录，无法更新状态", request_id)
             return None
 
+        if cls._is_billing_terminal(usage):
+            logger.debug(
+                "跳过已终态 Usage 状态更新: request_id={}, status={}, billing_status={}",
+                request_id,
+                status,
+                getattr(usage, "billing_status", None),
+            )
+            return usage
+
         # 避免状态回退：streaming 只能从 pending/streaming 进入
         if status == "streaming" and usage.status not in ("pending", "streaming"):
             logger.debug(
@@ -529,6 +527,10 @@ class UsageLifecycleMixin:
             usage.billing_status = "void"
             if getattr(usage, "finalized_at", None) is None:
                 usage.finalized_at = datetime.now(timezone.utc)
+            usage.total_cost_usd = to_money_decimal(0)
+            usage.request_cost_usd = to_money_decimal(0)
+            usage.actual_total_cost_usd = to_money_decimal(0)
+            usage.actual_request_cost_usd = to_money_decimal(0)
 
         db.commit()
 

src/services/usage/recording.py

@@ -26,6 +26,18 @@ from src.services.usage._types import UsageCostInfo, UsageRecordParams
 from src.services.wallet import WalletService
 
 
+def _coerce_finalized_at(value: Any, fallback: datetime | None = None) -> datetime | None:
+    if isinstance(value, datetime):
+        return value if value.tzinfo is not None else value.replace(tzinfo=timezone.utc)
+    if isinstance(value, str):
+        try:
+            parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+            return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=timezone.utc)
+        except ValueError:
+            return fallback
+    return fallback
+
+
 def _extract_manual_proxy_node_id(metadata: dict[str, Any] | None) -> str | None:
     """从 request_metadata 中提取手动代理节点 ID（仅 is_manual 节点）。
 
@@ -131,10 +143,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
 
     @staticmethod
     def _is_usage_finalized(usage: Usage) -> bool:
-        return (
+        # 与 UsageLifecycleMixin._is_billing_terminal 语义一致
-            getattr(usage, "billing_status", None) in {"settled", "void"}
+        return getattr(usage, "billing_status", None) in {"settled", "void"}
-            and getattr(usage, "finalized_at", None) is not None
-        )
 
     @classmethod
     def _finalize_usage_billing(
@@ -157,10 +167,7 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
                 usage.billing_status = "pending"
             return False, False
 
-        if (
+        if getattr(usage, "billing_status", None) in {"settled", "void"}:
-            getattr(usage, "billing_status", None) in {"settled", "void"}
-            and getattr(usage, "finalized_at", None) is not None
-        ):
             return False, False
 
         now = finalized_at or datetime.now(timezone.utc)
@@ -221,6 +228,7 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
         cache_ttl_minutes: int | None = None,
         use_tiered_pricing: bool = True,
         target_model: str | None = None,
+        finalized_at: datetime | None = None,
     ) -> Usage:
         """异步记录使用量（简化版，仅插入新记录）
 
@@ -310,6 +318,7 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
             usage=usage,
             total_cost=total_cost,
             status=status,
+            finalized_at=finalized_at,
         )
 
         dispatch_codex_quota_sync_from_response_headers(
@@ -363,6 +372,7 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
         cache_ttl_minutes: int | None = None,
         use_tiered_pricing: bool = True,
         target_model: str | None = None,
+        finalized_at: datetime | None = None,
     ) -> Usage:
         """记录使用量（完整版，支持更新已存在记录和用户统计）
 
@@ -464,6 +474,7 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
                 usage=usage,
                 total_cost=total_cost,
                 status=status,
+                finalized_at=finalized_at,
             )
 
             if accounted:
@@ -568,6 +579,7 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
         provider_api_key_id: str | None = None,
         status: str = "completed",
         target_model: str | None = None,
+        finalized_at: datetime | None = None,
     ) -> Usage:
         """
         记录"已计算好的"成本（用于 Video/Image/Audio 等异步任务的 FormulaEngine 计费结果）。
@@ -690,6 +702,7 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
             usage=usage,
             total_cost=total_cost,
             status=status,
+            finalized_at=finalized_at,
         )
 
         if accounted:
@@ -961,7 +974,7 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
         update_results = prepared_results[: len(update_params_list)]
         insert_results = prepared_results[len(update_params_list) :]
 
-        finalized_at = datetime.now(timezone.utc)
+        batch_finalized_at = datetime.now(timezone.utc)
 
         # 1. 处理需要更新的记录
         for i, (record, request_id, params) in enumerate(update_params_list):
@@ -982,7 +995,9 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
                     usage=existing_usage,
                     total_cost=total_cost,
                     status=usage_params.get("status"),
-                    finalized_at=finalized_at,
+                    finalized_at=_coerce_finalized_at(
+                        record.get("finalized_at"), batch_finalized_at
+                    ),
                 )
                 usages.append(existing_usage)
                 updated_count += 1
@@ -1043,7 +1058,9 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
                     usage=usage,
                     total_cost=total_cost,
                     status=usage_params.get("status"),
-                    finalized_at=finalized_at,
+                    finalized_at=_coerce_finalized_at(
+                        record.get("finalized_at"), batch_finalized_at
+                    ),
                 )
                 usages.append(usage)
                 inserted_count += 1

src/services/wallet/__init__.py

@@ -1,3 +1,12 @@
+from src.services.wallet.daily_usage_ledger import (
+    WalletDailyUsageLedgerService,
+    WalletDailyUsageSnapshot,
+)
 from src.services.wallet.service import WalletAccessResult, WalletService
 
-__all__ = ["WalletAccessResult", "WalletService"]
+__all__ = [
+    "WalletAccessResult",
+    "WalletDailyUsageLedgerService",
+    "WalletDailyUsageSnapshot",
+    "WalletService",
+]

src/services/wallet/daily_usage_ledger.py

@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, datetime, time, timedelta, timezone
+from decimal import Decimal
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from src.core.logger import logger
+from src.models.database import Usage, WalletDailyUsageLedger
+from src.services.billing.precision import to_money_decimal
+from src.services.system.scheduler import APP_TIMEZONE
+
+
+@dataclass(slots=True)
+class WalletDailyUsageSnapshot:
+    wallet_id: str | None
+    billing_date: date
+    billing_timezone: str
+    total_cost_usd: Decimal
+    total_requests: int
+    input_tokens: int
+    output_tokens: int
+    cache_creation_tokens: int
+    cache_read_tokens: int
+    first_finalized_at: datetime | None
+    last_finalized_at: datetime | None
+    aggregated_at: datetime
+    is_today: bool = False
+    id: str | None = None
+
+
+class WalletDailyUsageLedgerService:
+    """钱包每日消费汇总服务。"""
+
+    @staticmethod
+    def get_timezone(timezone_name: str | None = None) -> ZoneInfo:
+        tz_name = timezone_name or APP_TIMEZONE
+        try:
+            return ZoneInfo(tz_name)
+        except Exception:
+            logger.warning("Invalid billing timezone {}, fallback to UTC", tz_name)
+            return ZoneInfo("UTC")
+
+    @classmethod
+    def get_today_billing_date(cls, timezone_name: str | None = None) -> date:
+        tz = cls.get_timezone(timezone_name)
+        return datetime.now(tz).date()
+
+    @classmethod
+    def get_day_window_utc(
+        cls,
+        billing_date: date,
+        timezone_name: str | None = None,
+    ) -> tuple[datetime, datetime]:
+        tz = cls.get_timezone(timezone_name)
+        local_start = datetime.combine(billing_date, time.min, tzinfo=tz)
+        local_end = local_start + timedelta(days=1)
+        return local_start.astimezone(timezone.utc), local_end.astimezone(timezone.utc)
+
+    @classmethod
+    def aggregate_day(
+        cls,
+        db: Session,
+        billing_date: date,
+        *,
+        timezone_name: str | None = None,
+        commit: bool = True,
+    ) -> int:
+        tz_name = timezone_name or APP_TIMEZONE
+        start_utc, end_utc = cls.get_day_window_utc(billing_date, tz_name)
+        now_utc = datetime.now(timezone.utc)
+
+        rows = (
+            db.query(
+                Usage.wallet_id.label("wallet_id"),
+                func.count(Usage.id).label("total_requests"),
+                func.coalesce(func.sum(Usage.total_cost_usd), 0).label("total_cost_usd"),
+                func.coalesce(func.sum(Usage.input_tokens), 0).label("input_tokens"),
+                func.coalesce(func.sum(Usage.output_tokens), 0).label("output_tokens"),
+                func.coalesce(func.sum(Usage.cache_creation_input_tokens), 0).label(
+                    "cache_creation_tokens"
+                ),
+                func.coalesce(func.sum(Usage.cache_read_input_tokens), 0).label(
+                    "cache_read_tokens"
+                ),
+                func.min(Usage.finalized_at).label("first_finalized_at"),
+                func.max(Usage.finalized_at).label("last_finalized_at"),
+            )
+            .filter(
+                Usage.wallet_id.isnot(None),
+                Usage.billing_status == "settled",
+                Usage.total_cost_usd > 0,
+                Usage.finalized_at >= start_utc,
+                Usage.finalized_at < end_utc,
+            )
+            .group_by(Usage.wallet_id)
+            .all()
+        )
+
+        existing_ledgers = (
+            db.query(WalletDailyUsageLedger)
+            .filter(
+                WalletDailyUsageLedger.billing_date == billing_date,
+                WalletDailyUsageLedger.billing_timezone == tz_name,
+            )
+            .all()
+        )
+        existing_map = {ledger.wallet_id: ledger for ledger in existing_ledgers}
+        seen_wallet_ids: set[str] = set()
+
+        for row in rows:
+            wallet_id = getattr(row, "wallet_id", None)
+            if not wallet_id:
+                continue
+            seen_wallet_ids.add(str(wallet_id))
+            ledger = existing_map.get(str(wallet_id))
+            if ledger is None:
+                ledger = WalletDailyUsageLedger(
+                    wallet_id=str(wallet_id),
+                    billing_date=billing_date,
+                    billing_timezone=tz_name,
+                    aggregated_at=now_utc,
+                )
+                db.add(ledger)
+
+            ledger.total_cost_usd = to_money_decimal(getattr(row, "total_cost_usd", 0) or 0)
+            ledger.total_requests = int(getattr(row, "total_requests", 0) or 0)
+            ledger.input_tokens = int(getattr(row, "input_tokens", 0) or 0)
+            ledger.output_tokens = int(getattr(row, "output_tokens", 0) or 0)
+            ledger.cache_creation_tokens = int(getattr(row, "cache_creation_tokens", 0) or 0)
+            ledger.cache_read_tokens = int(getattr(row, "cache_read_tokens", 0) or 0)
+            ledger.first_finalized_at = getattr(row, "first_finalized_at", None)
+            ledger.last_finalized_at = getattr(row, "last_finalized_at", None)
+            ledger.aggregated_at = now_utc
+
+        stale_ledgers = [
+            ledger for ledger in existing_ledgers if ledger.wallet_id not in seen_wallet_ids
+        ]
+        for ledger in stale_ledgers:
+            db.delete(ledger)
+
+        if commit:
+            db.commit()
+        return len(seen_wallet_ids)
+
+    @classmethod
+    def get_today_snapshot(
+        cls,
+        db: Session,
+        wallet_id: str | None,
+        *,
+        timezone_name: str | None = None,
+    ) -> WalletDailyUsageSnapshot:
+        tz_name = timezone_name or APP_TIMEZONE
+        billing_date = cls.get_today_billing_date(tz_name)
+        start_utc, end_utc = cls.get_day_window_utc(billing_date, tz_name)
+        now_utc = datetime.now(timezone.utc)
+
+        if wallet_id:
+            row = (
+                db.query(
+                    func.count(Usage.id).label("total_requests"),
+                    func.coalesce(func.sum(Usage.total_cost_usd), 0).label("total_cost_usd"),
+                    func.coalesce(func.sum(Usage.input_tokens), 0).label("input_tokens"),
+                    func.coalesce(func.sum(Usage.output_tokens), 0).label("output_tokens"),
+                    func.coalesce(func.sum(Usage.cache_creation_input_tokens), 0).label(
+                        "cache_creation_tokens"
+                    ),
+                    func.coalesce(func.sum(Usage.cache_read_input_tokens), 0).label(
+                        "cache_read_tokens"
+                    ),
+                    func.min(Usage.finalized_at).label("first_finalized_at"),
+                    func.max(Usage.finalized_at).label("last_finalized_at"),
+                )
+                .filter(
+                    Usage.wallet_id == wallet_id,
+                    Usage.billing_status == "settled",
+                    Usage.total_cost_usd > 0,
+                    Usage.finalized_at >= start_utc,
+                    Usage.finalized_at < end_utc,
+                )
+                .first()
+            )
+        else:
+            row = None
+
+        return WalletDailyUsageSnapshot(
+            wallet_id=wallet_id,
+            billing_date=billing_date,
+            billing_timezone=tz_name,
+            total_cost_usd=to_money_decimal(getattr(row, "total_cost_usd", 0) or 0),
+            total_requests=int(getattr(row, "total_requests", 0) or 0),
+            input_tokens=int(getattr(row, "input_tokens", 0) or 0),
+            output_tokens=int(getattr(row, "output_tokens", 0) or 0),
+            cache_creation_tokens=int(getattr(row, "cache_creation_tokens", 0) or 0),
+            cache_read_tokens=int(getattr(row, "cache_read_tokens", 0) or 0),
+            first_finalized_at=getattr(row, "first_finalized_at", None),
+            last_finalized_at=getattr(row, "last_finalized_at", None),
+            aggregated_at=now_utc,
+            is_today=True,
+        )

tests/services/test_usage_queue_events.py

@@ -378,6 +378,7 @@ async def test_event_to_record_body_deserialization(monkeypatch: Any) -> None:
     event = build_usage_event(
         event_type=UsageEventType.COMPLETED,
         request_id="req-body-test",
+        timestamp_ms=1_700_000_000_123,
         data={
             "user_id": "user-1",
             "api_key_id": "key-1",
@@ -396,6 +397,7 @@ async def test_event_to_record_body_deserialization(monkeypatch: Any) -> None:
     assert record["request_body"]["messages"][0]["content"] == "hello"
     assert isinstance(record["response_body"], dict)
     assert record["response_body"]["choices"][0]["message"]["content"] == "hi"
+    assert record["finalized_at"] is not None
 
 
 @pytest.mark.asyncio

tests/services/test_usage_state_machine.py

@@ -0,0 +1,85 @@
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException
+
+from src.services.task.video.cancel import VideoTaskCancelService
+from src.services.usage.service import UsageService
+
+
+class _DummyQuery:
+    def __init__(self, obj: Any) -> None:
+        self._obj = obj
+
+    def filter(self, *args: Any, **kwargs: Any) -> "_DummyQuery":
+        return self
+
+    def with_for_update(self) -> "_DummyQuery":
+        return self
+
+    def first(self) -> Any:
+        return self._obj
+
+
+def test_update_settled_billing_rejects_terminal_usage() -> None:
+    usage = SimpleNamespace(
+        request_id="req-1",
+        billing_status="settled",
+        finalized_at="2026-03-06T00:00:00+00:00",
+        total_cost_usd=1.23,
+    )
+    db = MagicMock()
+    db.query.return_value = _DummyQuery(usage)
+
+    updated = UsageService.update_settled_billing(
+        db,
+        request_id="req-1",
+        total_cost_usd=2.34,
+        status="completed",
+    )
+
+    assert updated is False
+    assert usage.total_cost_usd == 1.23
+
+
+def test_update_usage_status_skips_terminal_usage() -> None:
+    usage = SimpleNamespace(
+        request_id="req-1",
+        billing_status="settled",
+        finalized_at="2026-03-06T00:00:00+00:00",
+        status="completed",
+        error_message=None,
+        provider_name="demo",
+    )
+    db = MagicMock()
+    db.query.return_value = _DummyQuery(usage)
+
+    result = UsageService.update_usage_status(
+        db=db,
+        request_id="req-1",
+        status="failed",
+        error_message="boom",
+    )
+
+    assert result is usage
+    assert usage.status == "completed"
+    assert usage.error_message is None
+    db.commit.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_task_cancel_rejects_terminal_task() -> None:
+    task = SimpleNamespace(
+        id="t1",
+        user_id="u1",
+        status="completed",
+        request_id="req-1",
+    )
+    svc = VideoTaskCancelService(MagicMock())
+
+    with pytest.raises(HTTPException) as exc:
+        await svc.cancel_task(task=task, task_id="t1")
+
+    assert exc.value.status_code == 409

tests/services/test_video_finalize.py

@@ -54,8 +54,8 @@ def _make_db() -> MagicMock:
     usage_obj = SimpleNamespace(
         id="usage-1",
         request_id="req-1",
-        billing_status="settled",  # finalize_submitted already settled
+        billing_status="pending",
-        request_metadata=None,  # no billing_updated_at yet
+        request_metadata=None,
     )
 
     q_usage = MagicMock()

tests/services/test_video_task_poller.py

@@ -48,3 +48,33 @@ async def test_poll_task_status_routes_gemini_video_to_gemini(
     assert result.status == VideoStatus.PROCESSING
     assert poll_gemini.await_count == 1
     assert poll_openai.await_count == 0
+
+
+@pytest.mark.asyncio
+async def test_update_task_after_poll_skips_terminal_cancelled_task(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    finalize = AsyncMock()
+    adapter = VideoTaskPollerAdapter(finalize_video_task_fn=finalize)
+
+    cancelled_task = SimpleNamespace(
+        id="t1",
+        status=VideoStatus.CANCELLED.value,
+    )
+
+    session = MagicMock()
+    session.__enter__.return_value = session
+    session.__exit__.return_value = None
+    session.get.return_value = cancelled_task
+
+    monkeypatch.setattr("src.services.task.video.poller_adapter.create_session", lambda: session)
+
+    await adapter.update_task_after_poll(
+        task_id="t1",
+        result=InternalVideoPollResult(status=VideoStatus.COMPLETED),
+        ctx=None,
+        redis_client=None,
+    )
+
+    finalize.assert_not_awaited()
+    session.commit.assert_not_called()