Merge pull request #2122 from pikasTech/feat/sub2api-credit-user-tokens

feat: 按用户聚合补偿窗口 Token

.agents/skills/unidesk-sub2api/references/codex-pool.md

@@ -156,7 +156,7 @@ bun scripts/cli.ts platform-infra sub2api codex-pool cleanup-probes --target D60
   - `--active-start <ISO> --active-end <ISO>` 使用精确半开区间 `[start,end)`；未携带 UTC 偏移的输入按 `config/platform-infra/sub2api-codex-pool.yaml#credits.timezone` 解释，并同时披露本地与 UTC 边界；
   - `--users <id-or-email,...>` 使用精确 selector；相对窗口、固定窗口和精确 selector 三者必须且只能选择一个；
   - `config/platform-infra/sub2api-codex-pool.yaml#credits.excludedUsers` 声明不得获得客户补偿的精确用户 ID 或邮箱；活跃窗口和显式用户模式都必须排除，并在默认表格中披露匹配 selector 与原因；
-  - 默认 dry-run，逐用户披露完整邮箱、用户 ID、请求数、当前余额、增加额和目标余额；
+  - 默认 dry-run，逐用户披露完整邮箱、用户 ID、请求数、原生 `input_tokens` / `output_tokens` / 总 Token、Token 字段覆盖情况、当前余额、增加额和目标余额；Token 字段缺失时标记 `partial`，不把缺失值伪装成完整零消耗；
   - 活跃窗口只允许生成计划；实际写入必须把 dry-run 返回的用户 ID 固定为精确 `--users` selectors，再显式 `--confirm`，防止移动窗口在确认时扩大或缩小名单；
   - 确认写入为每个用户生成由精确名单、金额和原因确定的 `Idempotency-Key`，调用原生 `POST /api/v1/admin/users/:id/balance` 的 `add` 操作，随后逐用户回读余额并对账；
   - 实际批量充值属于外部财务影响操作，执行前必须取得用户对本次具体名单、金额和原因的明确授权；仅提出设计目标或要求 dry-run 不构成授权。

scripts/src/platform-infra-sub2api-codex/credits.ts

@@ -209,18 +209,19 @@ function renderCredits(response: Record<string, unknown>): RenderedCliResult {
   if (typeof report.error === "string") lines.push(`ERROR ${text(report.error)}`);
   if (rows.length > 0) {
     lines.push("", renderTable([
-      ["EMAIL", "USER_ID", "ROLE", "STATUS", "REQUESTS", "BALANCE_BEFORE", "ADD_USD", "BALANCE_AFTER", "WRITE", "RECONCILED"],
+      ["EMAIL", "USER_ID", "ROLE", "STATUS", "REQUESTS", "INPUT_TOKENS", "OUTPUT_TOKENS", "TOTAL_TOKENS", "TOKEN_DATA", "BALANCE_BEFORE", "ADD_USD", "BALANCE_AFTER", "WRITE", "RECONCILED"],
       ...rows.map((row) => [
-        text(row.email), text(row.userId), text(row.role), text(row.status), text(row.requestCount), usd(row.balanceBefore), usd(row.amountUsd),
-        usd(row.balanceAfter), text(record(row.write).status), text(row.reconciled),
+        text(row.email), text(row.userId), text(row.role), text(row.status), integer(row.requestCount), integer(row.inputTokens), integer(row.outputTokens),
+        integer(row.totalTokens), tokenData(row), usd(row.balanceBefore), usd(row.amountUsd), usd(row.balanceAfter), text(record(row.write).status), text(row.reconciled),
       ]),
     ]));
   }
   if (excludedRows.length > 0) {
     lines.push("", "EXCLUDED", renderTable([
-      ["EMAIL", "USER_ID", "REQUESTS", "REASON", "MATCHED_SELECTOR"],
+      ["EMAIL", "USER_ID", "REQUESTS", "INPUT_TOKENS", "OUTPUT_TOKENS", "TOTAL_TOKENS", "TOKEN_DATA", "REASON", "MATCHED_SELECTOR"],
       ...excludedRows.map((row) => [
-        text(row.email), text(row.userId), text(row.requestCount), text(row.reason), text(row.matchedSelector),
+        text(row.email), text(row.userId), integer(row.requestCount), integer(row.inputTokens), integer(row.outputTokens), integer(row.totalTokens),
+        tokenData(row), text(row.reason), text(row.matchedSelector),
       ]),
     ]));
   }
@@ -243,3 +244,12 @@ function record(value: unknown): Record<string, unknown> { return typeof value =
 function arrayOfRecords(value: unknown): Record<string, unknown>[] { return Array.isArray(value) ? value.filter((item): item is Record<string, unknown> => typeof item === "object" && item !== null && !Array.isArray(item)) : []; }
 function text(value: unknown): string { return value === null || value === undefined || value === "" ? "-" : String(value).replace(/[\r\n\t]+/gu, " "); }
 function usd(value: unknown): string { const numeric = Number(value); return Number.isFinite(numeric) ? `$${numeric.toFixed(8)}` : text(value); }
+function integer(value: unknown): string {
+  if (value === null || value === undefined || value === "") return "-";
+  const numeric = Number(value);
+  return Number.isSafeInteger(numeric) ? numeric.toLocaleString("en-US") : text(value);
+}
+function tokenData(row: Record<string, unknown>): string {
+  if (row.tokenCompleteness !== "complete" && row.tokenCompleteness !== "partial") return "unavailable";
+  return `${text(row.tokenCompleteness)}:${integer(row.tokenRows)}/${integer(row.requestCount)} rows`;
+}

scripts/src/platform-infra-sub2api-codex/remote-python-credits.ts

@@ -39,7 +39,7 @@ def credits_active_usage(token, since, active_start, active_end, active_start_ut
         selection_kind = "active-window"
     if start_at is None or end_at is None or end_at <= start_at:
         raise RuntimeError("credits active window end must be after start")
-    user_requests = {}
+    user_usage = {}
     page = 1
     while True:
         path = f"/api/v1/admin/usage?page={page}&page_size=100&start_date={start_at.date().isoformat()}&end_date={end_at.date().isoformat()}&timezone=UTC&sort_by=created_at&sort_order=desc"
@@ -55,10 +55,21 @@ def credits_active_usage(token, since, active_start, active_end, active_start_ut
                 continue
             user_id = row.get("user_id") if row.get("user_id") is not None else row.get("userId")
             if user_id is not None and created_at is not None and start_at <= created_at < end_at:
-                user_requests[str(user_id)] = user_requests.get(str(user_id), 0) + 1
+                usage = user_usage.setdefault(str(user_id), {
+                    "requestCount": 0, "inputTokens": 0, "outputTokens": 0,
+                    "tokenRows": 0, "missingTokenRows": 0,
+                })
+                usage["requestCount"] += 1
+                has_tokens = row.get("input_tokens") is not None and row.get("output_tokens") is not None
+                if has_tokens:
+                    usage["tokenRows"] += 1
+                else:
+                    usage["missingTokenRows"] += 1
+                usage["inputTokens"] += int(row.get("input_tokens") or 0)
+                usage["outputTokens"] += int(row.get("output_tokens") or 0)
         total = int(data.get("total", 0)) if isinstance(data, dict) else 0
         if not batch or len(batch) < 100 or reached_start or page * 100 >= total:
-            return user_requests, start_at, end_at, {
+            return user_usage, start_at, end_at, {
                 "kind": selection_kind,
                 "since": since,
                 "timezone": timezone_name if selection_kind == "active-window" else "UTC",
@@ -78,6 +89,25 @@ def credits_excluded_selector(user, payload):
             return str(selector)
     return None
 
+def credits_usage_fields(usage):
+    if not isinstance(usage, dict):
+        return {
+            "requestCount": None, "inputTokens": None, "outputTokens": None, "totalTokens": None,
+            "tokenRows": None, "missingTokenRows": None, "tokenCompleteness": "unavailable",
+        }
+    input_tokens = int(usage.get("inputTokens") or 0)
+    output_tokens = int(usage.get("outputTokens") or 0)
+    missing_rows = int(usage.get("missingTokenRows") or 0)
+    return {
+        "requestCount": int(usage.get("requestCount") or 0),
+        "inputTokens": input_tokens,
+        "outputTokens": output_tokens,
+        "totalTokens": input_tokens + output_tokens,
+        "tokenRows": int(usage.get("tokenRows") or 0),
+        "missingTokenRows": missing_rows,
+        "tokenCompleteness": "complete" if missing_rows == 0 else "partial",
+    }
+
 def credits_selected_users(token, payload):
     users = credits_all_users(token)
     users_by_id = {str(item.get("id")): item for item in users if item.get("id") is not None}
@@ -86,28 +116,29 @@ def credits_selected_users(token, payload):
     active_start = payload.get("activeStart")
     active_end = payload.get("activeEnd")
     if active_since or (active_start and active_end):
-        requests, _, _, selection = credits_active_usage(
+        usage_by_user, _, _, selection = credits_active_usage(
             token, active_since, active_start, active_end,
             payload.get("activeStartUtc"), payload.get("activeEndUtc"), payload.get("timezone"),
         )
         selected = []
         excluded = []
-        for user_id, request_count in requests.items():
+        for user_id, usage in usage_by_user.items():
             user = users_by_id.get(user_id)
+            usage_fields = credits_usage_fields(usage)
             excluded_selector = credits_excluded_selector(user, payload) if isinstance(user, dict) else None
             if not isinstance(user, dict):
-                excluded.append({"userId": user_id, "reason": "user-not-found"})
+                excluded.append({"userId": user_id, "reason": "user-not-found", **usage_fields})
             elif excluded_selector is not None:
                 excluded.append({
-                    "userId": user.get("id"), "email": user.get("email"), "requestCount": request_count,
-                    "reason": "yaml-excluded-user", "matchedSelector": excluded_selector,
+                    "userId": user.get("id"), "email": user.get("email"),
+                    "reason": "yaml-excluded-user", "matchedSelector": excluded_selector, **usage_fields,
                 })
             elif user.get("role") == "admin":
-                excluded.append({"userId": user.get("id"), "email": user.get("email"), "reason": "admin-excluded"})
+                excluded.append({"userId": user.get("id"), "email": user.get("email"), "reason": "admin-excluded", **usage_fields})
             elif user.get("status") != "active" or user.get("deleted_at") is not None:
-                excluded.append({"userId": user.get("id"), "email": user.get("email"), "reason": "inactive-or-deleted"})
+                excluded.append({"userId": user.get("id"), "email": user.get("email"), "reason": "inactive-or-deleted", **usage_fields})
             else:
-                selected.append((user, request_count))
+                selected.append((user, usage_fields))
         return selected, excluded, selection
     selected = []
     excluded = []
@@ -123,11 +154,11 @@ def credits_selected_users(token, payload):
         excluded_selector = credits_excluded_selector(user, payload)
         if excluded_selector is not None:
             excluded.append({
-                "userId": user.get("id"), "email": user.get("email"), "requestCount": None,
-                "reason": "yaml-excluded-user", "matchedSelector": excluded_selector,
+                "userId": user.get("id"), "email": user.get("email"),
+                "reason": "yaml-excluded-user", "matchedSelector": excluded_selector, **credits_usage_fields(None),
             })
         else:
-            selected.append((user, None))
+            selected.append((user, credits_usage_fields(None)))
     return selected, excluded, {"kind": "exact-users", "selectors": payload.get("users") or []}
 
 def run_credits_grant():
@@ -146,12 +177,12 @@ def run_credits_grant():
     write_succeeded = 0
     write_failed = 0
     reconciled = 0
-    for user, request_count in sorted(selected, key=lambda item: (str(item[0].get("email") or ""), int(item[0].get("id") or 0))):
+    for user, usage_fields in sorted(selected, key=lambda item: (str(item[0].get("email") or ""), int(item[0].get("id") or 0))):
         before = profit_decimal(user.get("balance"))
         after = before + amount
         item = {
             "userId": user.get("id"), "email": user.get("email"), "role": user.get("role"), "status": user.get("status"),
-            "requestCount": request_count, "balanceBefore": profit_decimal_text(before), "amountUsd": profit_decimal_text(amount),
+            **usage_fields, "balanceBefore": profit_decimal_text(before), "amountUsd": profit_decimal_text(amount),
             "balanceAfter": profit_decimal_text(after), "write": {"attempted": False, "status": "dry-run", "error": None}, "reconciled": None,
         }
         if confirm: