From 46cb394ba1a40de35aded4e1a70998e430ad8060 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 19:19:06 +0800 Subject: [PATCH 1/7] Reject invalid success payloads from the billing APIs An HTTP 200 with a failing business code or a missing result structure was treated as a valid empty/zero response: fetch_request_usage never checked the business code (and a missing total silently stopped pagination after the first page), and fetch_credits could not tell a missing Accounts structure from a legitimately empty balance. Both now raise on a non-zero business code or missing data.data/total/Accounts structure, while a present-but-empty result remains a valid zero. Verified by new failure-semantics tests; existing paging and HTTP tests unchanged and passing. --- app/credits.py | 24 +++++++++++++--- tests/test_credits.py | 65 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 4 deletions(-) diff --git a/app/credits.py b/app/credits.py index 4fbba28..a70138f 100644 --- a/app/credits.py +++ b/app/credits.py @@ -312,7 +312,16 @@ def fetch_credits(access_token: str, uid: str = "", domain: str = "") -> dict: raise RuntimeError(str(payload.get("msg") or f"积分接口 code={code}")) data = payload.get("data") or {} resp = (data.get("Response") or {}).get("Data") or (data.get("data") or {}).get("Response", {}).get("Data") or data - accounts = resp.get("Accounts") or payload.get("data", {}).get("accounts") or [] + # 区分「合法的零余额」与「结构缺失的未知失败」：只有 Accounts/accounts 键存在才算有效响应 + accounts = None + if isinstance(resp, dict) and isinstance(resp.get("Accounts"), list): + accounts = resp["Accounts"] + elif isinstance(payload.get("data"), dict) and isinstance(payload["data"].get("accounts"), list): + accounts = payload["data"]["accounts"] + if accounts is None: + last_err = RuntimeError("积分接口返回缺少 Accounts 结构") + time.sleep(0.3 * (attempt + 1)) + continue if not accounts and attempt < 2: # 偶发空 Accounts，重试一次 time.sleep(0.3 * (attempt + 1)) continue @@ -554,8 +563,15 @@ def fetch_request_usage(access_token: str, days: int = USAGE_MAX_DAYS, dict(body_base, pageNum=page, pageSize=USAGE_PAGE_SIZE)) if status != 200: raise RuntimeError(f"用量明细接口 HTTP {status}") - data = payload.get("data") or {} - rows = data.get("data") or [] + code = payload.get("code") + if code not in (0, None): + raise RuntimeError(f"用量明细接口 code={code}: {str(payload.get('msg'))[:120]}") + data = payload.get("data") + # HTTP 200 但缺少业务结构不是「零用量」：total 缺失还会让分页提前中断 + if not isinstance(data, dict) or not isinstance(data.get("data"), list) \ + or not isinstance(data.get("total"), (int, float)): + raise RuntimeError("用量明细接口返回缺少 data.data/total 结构") + rows = data["data"] for row in rows: date = str(row.get("requestTime") or "")[:10] model = str(row.get("model") or "unknown") @@ -566,7 +582,7 @@ def fetch_request_usage(access_token: str, days: int = USAGE_MAX_DAYS, by_day[date][model] = round(by_day[date].get(model, 0.0) + credit, 6) total_credits += credit requests += 1 - if requests >= int(data.get("total") or 0) or not rows: + if requests >= int(data["total"]) or not rows: break return {"by_day": by_day, "total_credits": round(total_credits, 2), "requests": requests} diff --git a/tests/test_credits.py b/tests/test_credits.py index df77d6f..24d3674 100644 --- a/tests/test_credits.py +++ b/tests/test_credits.py @@ -502,6 +502,69 @@ def test_aggregate_credits(): print("✅ test_aggregate_credits") +def _fake_client(pages, seen=None): + """按 pageNum 返回预置响应的 httpx.Client 替身。""" + class FakeResp: + status_code = 200 + def __init__(self, payload): + self._p = payload + def json(self): + return self._p + + class FakeClient: + def __enter__(self): + return self + def __exit__(self, *a): + return False + def post(self, url, headers=None, json=None, timeout=None): + if seen is not None: + seen.append(json.get("pageNum", json.get("PageNumber"))) + return FakeResp(pages[min((json.get("pageNum") or json.get("PageNumber")) - 1, len(pages) - 1)]) + return FakeClient + + +def _with_client(fake, fn): + orig = credits.httpx.Client + credits.httpx.Client = fake + try: + return fn() + finally: + credits.httpx.Client = orig + + +def test_fetch_request_usage_rejects_invalid_success_payloads(): + """HTTP 200 但业务码失败或结构缺失：必须报错，不能当作零用量。""" + token = _jwt("https://www.codebuddy.cn/x") + for pages in ([{"code": 1059, "msg": "rate limited", "data": {"total": 0, "data": []}}], + [{"code": 0, "data": {}}], # 缺 data.data/total + [{"data": {"data": [], "total": 0}}], # code 缺失但结构完整 → 合法空 + ): + try: + result = _with_client(_fake_client(pages), + lambda: credits.fetch_request_usage(token)) + except RuntimeError: + assert pages[0].get("code") not in (0, None) or "data" not in pages[0].get("data", {}) \ + or "data" not in pages[0]["data"] + else: + assert pages[0].get("code") is None and result["requests"] == 0 # 合法空结果照旧可用 + print("✅ test_fetch_request_usage_rejects_invalid_success_payloads") + + +def test_fetch_credits_distinguishes_empty_from_missing_structure(): + """Accounts 键存在但为空 = 合法零余额；结构整体缺失 = 报错，不得覆盖缓存为零。""" + token = _jwt("https://www.codebuddy.cn/x") + empty = _with_client(_fake_client([{"code": 0, "data": {"Response": {"Data": {"Accounts": []}}}}]), + lambda: credits.fetch_credits(token)) + assert empty["credits"] == 0.0 and empty["count"] == 0 + for bad in ({"code": 0, "data": {}}, {"code": 0, "data": {"Response": {"Data": {}}}}): + try: + _with_client(_fake_client([bad]), lambda: credits.fetch_credits(token)) + raise AssertionError("missing Accounts structure must raise") + except RuntimeError as error: + assert "Accounts" in str(error) + print("✅ test_fetch_credits_distinguishes_empty_from_missing_structure") + + def test_fetch_request_usage_paging(): """mock 分页明细：跨页聚合 credit，按 日期×模型 归并；请求天数夹到 30 天。""" pages = [ @@ -728,6 +791,8 @@ def test_model_catalog_cache(): test_current_models_merge() test_credits_to_usd() test_aggregate_credits() + test_fetch_request_usage_rejects_invalid_success_payloads() + test_fetch_credits_distinguishes_empty_from_missing_structure() test_fetch_request_usage_paging() test_billing_balance_identity() test_billing_intl_split() From 9df6754d661d9e24777a96cc07f69e519b26cb99 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 19:27:02 +0800 Subject: [PATCH 2/7] Paginate billing APIs fully and mark truncated results as partial fetch_credits read only the first page of 100 packages and fetch_request_usage stopped silently at the 6000-row cap, so balances and usage could be understated while looking complete. Both now page until a short page (with a hard cap) and return partial=True when the cap is hit; the flag is stored in the credit ledger, surfaced by aggregate_credits, and exposed on the billing totals so an incomplete snapshot is visible instead of pretending to be exact. Verified by new pagination and page-cap tests (full suite passed). --- app/credits.py | 116 ++++++++++++++++++++++++++---------------- converter.py | 2 + tests/test_credits.py | 36 ++++++++++++- 3 files changed, 109 insertions(+), 45 deletions(-) diff --git a/app/credits.py b/app/credits.py index a70138f..dfba871 100644 --- a/app/credits.py +++ b/app/credits.py @@ -37,6 +37,8 @@ RESOURCE_PATH = "/v2/billing/meter/get-user-resource" CONFIG_PATH = "/v3/config" # cbc CLI CloudProductProvider 同源：云端模型表 RESOURCE_PRODUCT_CODE = "p_tcaca" +CREDITS_PAGE_SIZE = 100 +CREDITS_MAX_PAGES = 20 # 2000 个积分包封顶；到顶必须 partial 标记，不得装作完整 _INACTIVE_RE = re.compile(r"未开启|未开始|未开放|已过期|无.*活动|活动.*(?:结束|关闭|暂停)", re.I) _ALREADY_RE = re.compile(r"已签到|已领取|已经.*(?:签到|领取)|重复签到|already", re.I) @@ -274,12 +276,12 @@ def soonest_expiry(segments: list, now: float | None = None) -> float | None: return min(exps) if exps else None -def _resource_body() -> dict: +def _resource_body(page: int) -> dict: """与官方 Web 端一致：有效状态 [0,3]，结束时间范围 现在 ~ +101 年（只取未过期包）。""" fmt = "%Y-%m-%d %H:%M:%S" return { - "PageNumber": 1, - "PageSize": 100, + "PageNumber": page, + "PageSize": CREDITS_PAGE_SIZE, "ProductCode": RESOURCE_PRODUCT_CODE, "Status": [0, 3], "PackageEndTimeRangeBegin": time.strftime(fmt), @@ -287,50 +289,69 @@ def _resource_body() -> dict: } +def _fetch_accounts_page(client, url: str, headers: dict, page: int, *, retry_empty: bool) -> list: + """拉一页积分包（限次重试）；返回 Accounts 列表，失败抛 RuntimeError，401 抛 AuthExpiredError。""" + last_err: Exception | None = None + for attempt in range(3): + try: + status, payload = _post_json(client, url, headers, _resource_body(page)) + except AuthExpiredError: + raise + except httpx.HTTPError as e: + last_err = e + time.sleep(0.3 * (attempt + 1)) + continue + if status != 200: + last_err = RuntimeError(f"积分接口 HTTP {status}") + time.sleep(0.3 * (attempt + 1)) + continue + code = payload.get("code") + if code not in (0, None): + raise RuntimeError(str(payload.get("msg") or f"积分接口 code={code}")) + data = payload.get("data") or {} + resp = (data.get("Response") or {}).get("Data") or (data.get("data") or {}).get("Response", {}).get("Data") or data + # 区分「合法的零余额」与「结构缺失的未知失败」：只有 Accounts/accounts 键存在才算有效响应 + accounts = None + if isinstance(resp, dict) and isinstance(resp.get("Accounts"), list): + accounts = resp["Accounts"] + elif isinstance(payload.get("data"), dict) and isinstance(payload["data"].get("accounts"), list): + accounts = payload["data"]["accounts"] + if accounts is None: + last_err = RuntimeError("积分接口返回缺少 Accounts 结构") + time.sleep(0.3 * (attempt + 1)) + continue + if not accounts and retry_empty and attempt < 2: # 偶发空 Accounts，重试一次 + time.sleep(0.3 * (attempt + 1)) + continue + return accounts + raise RuntimeError(f"积分查询失败: {last_err}") + + def fetch_credits(access_token: str, uid: str = "", domain: str = "") -> dict: - """查询剩余积分：{credits, count, segments, soonest_expiry}。空结果重试，401 抛 AuthExpiredError。""" + """查询剩余积分：{credits, count, segments, soonest_expiry, partial}。401 抛 AuthExpiredError。 + + 分页遍历到不足一页为止；达到 CREDITS_MAX_PAGES 上限时 partial=True，调用方不得把结果当完整值。""" host = hosts_for_token(access_token, domain)[0] url = host + RESOURCE_PATH headers = _web_headers(host, access_token, uid, domain) - last_err: Exception | None = None + accounts: list = [] + partial = False with httpx.Client() as client: - for attempt in range(3): - try: - status, payload = _post_json(client, url, headers, _resource_body()) - except AuthExpiredError: - raise - except httpx.HTTPError as e: - last_err = e - time.sleep(0.3 * (attempt + 1)) - continue - if status != 200: - last_err = RuntimeError(f"积分接口 HTTP {status}") - time.sleep(0.3 * (attempt + 1)) - continue - code = payload.get("code") - if code not in (0, None): - raise RuntimeError(str(payload.get("msg") or f"积分接口 code={code}")) - data = payload.get("data") or {} - resp = (data.get("Response") or {}).get("Data") or (data.get("data") or {}).get("Response", {}).get("Data") or data - # 区分「合法的零余额」与「结构缺失的未知失败」：只有 Accounts/accounts 键存在才算有效响应 - accounts = None - if isinstance(resp, dict) and isinstance(resp.get("Accounts"), list): - accounts = resp["Accounts"] - elif isinstance(payload.get("data"), dict) and isinstance(payload["data"].get("accounts"), list): - accounts = payload["data"]["accounts"] - if accounts is None: - last_err = RuntimeError("积分接口返回缺少 Accounts 结构") - time.sleep(0.3 * (attempt + 1)) - continue - if not accounts and attempt < 2: # 偶发空 Accounts，重试一次 - time.sleep(0.3 * (attempt + 1)) - continue - segments = merge_segments(extract_segments(accounts)) - credits = round(sum(s["remaining"] for s in segments), 2) - return {"credits": credits, "count": len(accounts), "segments": segments, - "soonest_expiry": soonest_expiry(segments), - "intl": is_international_host(host)} - raise RuntimeError(f"积分查询失败: {last_err}") + page = 0 + while True: + page += 1 + if page > CREDITS_MAX_PAGES: + partial = True + break + rows = _fetch_accounts_page(client, url, headers, page, retry_empty=(page == 1)) + accounts.extend(rows) + if len(rows) < CREDITS_PAGE_SIZE: + break + segments = merge_segments(extract_segments(accounts)) + credits = round(sum(s["remaining"] for s in segments), 2) + return {"credits": credits, "count": len(accounts), "segments": segments, + "soonest_expiry": soonest_expiry(segments), + "intl": is_international_host(host), "partial": partial} @@ -536,6 +557,8 @@ def aggregate_credits(creds_snapshot: dict) -> dict: return {"remaining": round(sum(g["remaining"] for g in groups.values()), 2), "used_by_quota": round(sum(g["used_by_quota"] for g in groups.values()), 2), "soonest_expiry": min(exps) if exps else None, + "partial": any(bool((e.get("credits") or {}).get("partial")) + for e in dedupe_by_identity(creds_snapshot).values()), "groups": groups} @@ -544,7 +567,7 @@ def fetch_request_usage(access_token: str, days: int = USAGE_MAX_DAYS, uid: str = "", domain: str = "") -> dict: """拉官方用量明细，按 日期×模型 聚合实际扣减的 credits。 - 返回 {by_day: {'YYYY-MM-DD': {model: credits}}, total_credits, requests}。 + 返回 {by_day: {'YYYY-MM-DD': {model: credits}}, total_credits, requests, partial}。 跨度超 31 天官方会静默返回空，故 days 强制夹到 USAGE_MAX_DAYS。""" days = max(1, min(int(days or USAGE_MAX_DAYS), USAGE_MAX_DAYS)) host = hosts_for_token(access_token, domain)[0] @@ -557,6 +580,7 @@ def fetch_request_usage(access_token: str, days: int = USAGE_MAX_DAYS, by_day: dict = {} total_credits = 0.0 requests = 0 + partial = False with httpx.Client() as client: for page in range(1, USAGE_MAX_PAGES + 1): status, payload = _post_json(client, url, headers, @@ -584,7 +608,10 @@ def fetch_request_usage(access_token: str, days: int = USAGE_MAX_DAYS, requests += 1 if requests >= int(data["total"]) or not rows: break - return {"by_day": by_day, "total_credits": round(total_credits, 2), "requests": requests} + else: + partial = True # 达到页数上限仍可能有剩余：标记不完整，不装作全量 + return {"by_day": by_day, "total_credits": round(total_credits, 2), "requests": requests, + "partial": partial} # --------------------------------------------------------------------------- @@ -669,6 +696,7 @@ def update_credits(self, cred_id: str, result: dict): "soonest_expiry": result.get("soonest_expiry"), "fetched_at": time.time(), "intl": bool(result.get("intl")), # 站点归属：国内/国际积分与单价均独立 + "partial": bool(result.get("partial")), # 分页到顶：余额被低估，下游必须可见 } e["error"] = None self._save() diff --git a/converter.py b/converter.py index 5dd3bea..4736acc 100644 --- a/converter.py +++ b/converter.py @@ -1773,6 +1773,8 @@ def _billing_totals() -> dict: "soonest_expiry": agg.get("soonest_expiry"), "price_cny": price_cny, "price_usd": price_usd, "rate": rate, "used_source": "official_usage_detail" if detail else "quota_delta", + # 任一端数据不完整（积分分页到顶 / 用量到顶 / 账号同步失败）时对外可见 + "partial": bool(agg.get("partial") or cache.get("partial")), "groups": groups_out, "by_day": cache.get("by_day") or {}} diff --git a/tests/test_credits.py b/tests/test_credits.py index 24d3674..cca44d4 100644 --- a/tests/test_credits.py +++ b/tests/test_credits.py @@ -565,6 +565,38 @@ def test_fetch_credits_distinguishes_empty_from_missing_structure(): print("✅ test_fetch_credits_distinguishes_empty_from_missing_structure") +def test_fetch_credits_paginates_until_short_page(): + """积分包超过一页时翻页累加；不足一页停止；达到页数上限标记 partial。""" + token = _jwt("https://www.codebuddy.cn/x") + account = lambda i: {"PackageName": f"p{i}", "PackageCode": f"c{i}", + "SlicePeriodUsageDetails": [{"SlicePeriodCapacityRemainPrecise": "1", + "DeductionEndTime": None}]} + full_page = {"code": 0, "data": {"Response": {"Data": {"Accounts": [account(i) for i in range(100)]}}}} + short_page = {"code": 0, "data": {"Response": {"Data": {"Accounts": [account(1000)]}}}} + seen = [] + result = _with_client(_fake_client([full_page, short_page], seen), lambda: credits.fetch_credits(token)) + assert seen == [1, 2] and result["count"] == 101 and result["credits"] == 101.0 + assert result["partial"] is False + + seen = [] + result = _with_client(_fake_client([full_page] * credits.CREDITS_MAX_PAGES, seen), + lambda: credits.fetch_credits(token)) + assert len(seen) == credits.CREDITS_MAX_PAGES + assert result["partial"] is True + print("✅ test_fetch_credits_paginates_until_short_page") + + +def test_fetch_request_usage_marks_partial_at_page_cap(): + """用量明细达到页数上限且 total 更大时必须标记 partial。""" + token = _jwt("https://www.codebuddy.cn/x") + big_total = credits.USAGE_MAX_PAGES * credits.USAGE_PAGE_SIZE + 1 + row = {"requestTime": "2026-09-01 10:00:00", "model": "m", "credit": 0.01} + page = {"code": 0, "data": {"total": big_total, "data": [row]}} + result = _with_client(_fake_client([page]), lambda: credits.fetch_request_usage(token)) + assert result["partial"] is True and result["requests"] == credits.USAGE_MAX_PAGES + print("✅ test_fetch_request_usage_marks_partial_at_page_cap") + + def test_fetch_request_usage_paging(): """mock 分页明细：跨页聚合 credit，按 日期×模型 归并；请求天数夹到 30 天。""" pages = [ @@ -600,7 +632,7 @@ def post(self, url, headers=None, json=None, timeout=None): finally: credits.httpx.Client = orig assert seen["pages"] == [1, 2], seen # 按 total 停止分页 - assert u["requests"] == 3 and abs(u["total_credits"] - 0.75) < 1e-9 + assert u["requests"] == 3 and abs(u["total_credits"] - 0.75) < 1e-9 and u["partial"] is False assert u["by_day"]["2026-09-01"]["glm-5.3"] == 0.75 assert "hy4-preview" in u["by_day"]["2026-09-02"] # 免费模型 0 credit 也计入请求数 import time as _t @@ -793,6 +825,8 @@ def test_model_catalog_cache(): test_aggregate_credits() test_fetch_request_usage_rejects_invalid_success_payloads() test_fetch_credits_distinguishes_empty_from_missing_structure() + test_fetch_credits_paginates_until_short_page() + test_fetch_request_usage_marks_partial_at_page_cap() test_fetch_request_usage_paging() test_billing_balance_identity() test_billing_intl_split() From 6c71125330f80f54a397c92b5d834a6eba53f8c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 19:54:12 +0800 Subject: [PATCH 3/7] Keep per-account usage snapshots so one failing account loses nothing Usage sync rebuilt the aggregate from scratch each round and published it whenever at least one account succeeded, so an account whose fetch failed saw its previously synced usage vanish while fetched_at claimed a fresh sync. Each account now keeps its own snapshot in usage_daily_accounts; a successful sync replaces only that account's data, and a failed one keeps its history while the published view is marked partial with stale_accounts listed. Snapshots of removed or disabled credentials are excluded from the aggregate. Verified by a new sync regression test covering failure, recovery, and credential removal (test_credits.py passes). --- converter.py | 92 +++++++++++++++++++++++++++++++------------ tests/test_credits.py | 63 +++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+), 26 deletions(-) diff --git a/converter.py b/converter.py index 4736acc..5fbc2d3 100644 --- a/converter.py +++ b/converter.py @@ -1198,9 +1198,11 @@ def publish(): def _sync_usage(pool): - """历史用量仅在定时/手动维护时同步，入库唤醒不额外拉取历史。""" - by_day, groups = {}, {} - used, count, any_success = 0.0, 0, False + """历史用量仅在定时/手动维护时同步；每账号独立快照，单账号失败只替换自身数据。""" + accounts = CONFIG.get("usage_daily_accounts") + if not isinstance(accounts, dict): + accounts = CONFIG["usage_daily_accounts"] = {} + stale = set() for entry in pool.entries(): if not model_policy.credential_enabled(CONFIG, entry): continue @@ -1212,29 +1214,66 @@ def _sync_usage(pool): site = site_for_headers(headers) usage = credits_mod.fetch_request_usage(_bearer_token(headers), uid=headers.get("X-User-Id", ""), domain=headers.get("X-Domain", "")) - def merge(): - nonlocal used, count, any_success - any_success = True - group = groups.setdefault(site, {"by_day": {}, "total_credits": 0.0, "requests": 0}) - for day, models in usage["by_day"].items(): - total_day = by_day.setdefault(day, {}) - site_day = group["by_day"].setdefault(day, {}) - for model, credit in models.items(): - total_day[model] = round(total_day.get(model, 0.0) + credit, 6) - site_day[model] = round(site_day.get(model, 0.0) + credit, 6) - group["total_credits"] += usage["total_credits"] - group["requests"] += usage["requests"] - used += usage["total_credits"] - count += usage["requests"] - pool.apply_if_current(cm, generation, merge) + def store(): + accounts[entry["id"]] = {"site": site, "by_day": usage["by_day"], + "total_credits": round(usage["total_credits"], 2), + "requests": usage["requests"], + "partial": bool(usage.get("partial")), + "fetched_at": time.time()} + pool.apply_if_current(cm, generation, store) except Exception as error: - _log(f"[usage] {Path(entry['id']).name} 明细拉取失败: {_network_error_text(error)}") - if any_success: - for group in groups.values(): - group["total_credits"] = round(group["total_credits"], 2) - CONFIG["usage_daily"] = {"by_day": by_day, "groups": groups, "total_credits": round(used, 2), - "requests": count, "fetched_at": time.time()} - _log(f"[usage] 明细已同步: {count} 请求 / {used:.2f} credits") + stale.add(entry["id"]) + _log(f"[usage] {Path(entry['id']).name} 明细拉取失败（保留其上次成功快照）: {_network_error_text(error)}") + _publish_usage_daily(pool, stale) + + +def _publish_usage_daily(pool, stale=()): + """按当前启用账号的快照重建聚合视图；本轮失败的账号保留历史并列入 stale_accounts。 + + 窗口说明：凭据身份更换后，旧快照最多残留一个同步周期，随后被新账号的快照替换。""" + accounts = CONFIG.get("usage_daily_accounts") + if not isinstance(accounts, dict): + accounts = {} + enabled = {e["id"] for e in pool.entries() if model_policy.credential_enabled(CONFIG, e)} + by_day, groups = {}, {} + used, count = 0.0, 0 + partial = False + newest = 0.0 + included = False + stale_out = [] + for cred_id, snap in accounts.items(): + if cred_id not in enabled: + continue + included = True + site = snap.get("site") or "domestic" + group = groups.setdefault(site, {"by_day": {}, "total_credits": 0.0, "requests": 0}) + for day, models in (snap.get("by_day") or {}).items(): + total_day = by_day.setdefault(day, {}) + site_day = group["by_day"].setdefault(day, {}) + for model, credit in models.items(): + total_day[model] = round(total_day.get(model, 0.0) + credit, 6) + site_day[model] = round(site_day.get(model, 0.0) + credit, 6) + group["total_credits"] += float(snap.get("total_credits") or 0) + group["requests"] += int(snap.get("requests") or 0) + used += float(snap.get("total_credits") or 0) + count += int(snap.get("requests") or 0) + newest = max(newest, float(snap.get("fetched_at") or 0)) + if snap.get("partial"): + partial = True + if cred_id in stale: + partial = True + stale_out.append(Path(cred_id).name) + if not included: + return # 还没有任何成功快照：不覆盖已有视图 + for group in groups.values(): + group["total_credits"] = round(group["total_credits"], 2) + out = {"by_day": by_day, "groups": groups, "total_credits": round(used, 2), + "requests": count, "fetched_at": newest, "partial": partial} + if stale_out: + out["stale_accounts"] = sorted(stale_out) + CONFIG["usage_daily"] = out + _log(f"[usage] 明细已同步: {count} 请求 / {used:.2f} credits" + + (f" | {len(stale_out)} 账号保留历史快照" if stale_out else "")) def _housekeep_once(pool: CredentialPool, ledger, *, pending_only=False): @@ -1353,7 +1392,8 @@ async def _protocol_http_exception(request: Request, exc: HTTPException): "model_guard": True, # 表外模型本地拦截，不转发上游 "max_images": 16, "image_policy": "truncate", "max_request_bytes": 32 * 1024 * 1024, "log_body_limit": 65536, - "usage_daily": None, # 官方用量明细（日期×模型 credit），供 billing/usage 出 daily_costs + "usage_daily": None, # 官方用量聚合视图（日期×模型 credit），供 billing/usage 出 daily_costs + "usage_daily_accounts": None, # 按账号的用量快照；单账号失败不丢历史 "credit_price_cny": None, "credit_price_usd": None, "usd_rate": None, "desensitize": False, "no_compact": False, "keep_tool_metadata": False} # 单价 None=取 credits 模块默认 diff --git a/tests/test_credits.py b/tests/test_credits.py index cca44d4..333258a 100644 --- a/tests/test_credits.py +++ b/tests/test_credits.py @@ -597,6 +597,68 @@ def test_fetch_request_usage_marks_partial_at_page_cap(): print("✅ test_fetch_request_usage_marks_partial_at_page_cap") +def test_sync_usage_keeps_per_account_snapshots_on_failure(): + """单账号同步失败：聚合保留其上次成功快照并标记 stale/partial，不再整体覆盖丢失。""" + import converter + with tempfile.TemporaryDirectory() as td: + paths = [] + for uid in ("u1", "u2"): + p = Path(td) / f"{uid}.info" + p.write_text(json.dumps({ + "auth": {"accessToken": f"token-{uid}", "refreshToken": "r", + "domain": "https://www.codebuddy.cn", + "expiresAt": int(time.time() * 1000) + 86400000, + "lastRefreshTime": time.time() * 1000}, + "account": {"uid": uid, "enterpriseId": "e"}}), encoding="utf-8") + paths.append(p) + pool = converter.CredentialPool(paths) + snapshots = { + "token-u1": {"by_day": {"2026-09-01": {"m": 10.0}}, "total_credits": 10.0, "requests": 1, "partial": False}, + "token-u2": {"by_day": {"2026-09-02": {"m": 20.0}}, "total_credits": 20.0, "requests": 2, "partial": False}, + } + failing = set() + + def fake_fetch(token, uid="", domain=""): + if token in failing: + raise RuntimeError("synthetic sync failure") + return snapshots[token] + + saved = (converter.CONFIG.get("usage_daily"), converter.CONFIG.get("usage_daily_accounts"), + converter.CONFIG.get("control_store")) + orig_fetch = credits.fetch_request_usage + credits.fetch_request_usage = fake_fetch + converter.CONFIG.update(usage_daily=None, usage_daily_accounts=None, control_store=None) + try: + converter._sync_usage(pool) + view = converter.CONFIG["usage_daily"] + assert view["total_credits"] == 30.0 and view["requests"] == 3 + assert view["partial"] is False and "stale_accounts" not in view + + failing.add("token-u2") + converter._sync_usage(pool) + view = converter.CONFIG["usage_daily"] + assert view["total_credits"] == 30.0 and view["requests"] == 3 # u2 历史保留 + assert view["partial"] is True and view["stale_accounts"] == ["u2.info"] + + failing.clear() + snapshots["token-u2"] = {"by_day": {"2026-09-02": {"m": 25.0}}, + "total_credits": 25.0, "requests": 4, "partial": False} + converter._sync_usage(pool) + view = converter.CONFIG["usage_daily"] + assert view["total_credits"] == 35.0 and view["partial"] is False # 成功后自愈 + + paths[1].unlink() + pool.prune() + converter._sync_usage(pool) + view = converter.CONFIG["usage_daily"] + assert view["total_credits"] == 10.0 # 凭证删除后其快照不再计入 + finally: + credits.fetch_request_usage = orig_fetch + converter.CONFIG["usage_daily"], converter.CONFIG["usage_daily_accounts"], \ + converter.CONFIG["control_store"] = saved + print("✅ test_sync_usage_keeps_per_account_snapshots_on_failure") + + def test_fetch_request_usage_paging(): """mock 分页明细：跨页聚合 credit，按 日期×模型 归并；请求天数夹到 30 天。""" pages = [ @@ -825,6 +887,7 @@ def test_model_catalog_cache(): test_aggregate_credits() test_fetch_request_usage_rejects_invalid_success_payloads() test_fetch_credits_distinguishes_empty_from_missing_structure() + test_sync_usage_keeps_per_account_snapshots_on_failure() test_fetch_credits_paginates_until_short_page() test_fetch_request_usage_marks_partial_at_page_cap() test_fetch_request_usage_paging() From ea8c817b3c90a165826b613ff620f7098a1f9c94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 19:57:48 +0800 Subject: [PATCH 4/7] Price daily usage per site instead of one blended average The usage endpoint applied a single weighted-average price to every day and model, so with mixed domestic/international usage the per-day and per-model amounts were wrong even when the grand total happened to be right. daily_costs is now computed from each site group's own by_day at that site's price and then merged; the balance identity with total_usage still holds because both derive from the same per-site totals. The response also surfaces partial and stale_accounts when the usage data is incomplete. Verified by a two-site different-days arithmetic test and the existing billing identity tests. --- converter.py | 29 ++++++++++++++++++++++------- tests/test_credits.py | 41 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 7 deletions(-) diff --git a/converter.py b/converter.py index 5fbc2d3..9dc72dc 100644 --- a/converter.py +++ b/converter.py @@ -1850,17 +1850,27 @@ def billing_usage(start_date: Optional[str] = None, end_date: Optional[str] = No """OpenAI 用量端点：total_usage 单位美分；daily_costs 为官方明细按天×模型聚合（最近 30 天）。""" _check_auth(authorization, x_api_key) t = _billing_totals() - # 每 Credit 美分单价：按各站实际用量加权（保证 Σdaily 与 total_usage 一致） - cents_per_credit = ((t["used_usd"] * 100 / t["used"]) if t["used"] - else t["price_cny"] / t["rate"] * 100) + # 逐站逐日按本站单价折算后再合并：两站单价不同，统一平均价会让每天/每模型的金额失真。 + # Σdaily 与 total_usage 都由同一组分站用量算出，恒等关系保持不变。 + cents = {"domestic": t["price_cny"] / t["rate"] * 100, "international": t["price_usd"] * 100} + detail = CONFIG.get("usage_daily") or {} + priced: dict = {} + for site, group in (detail.get("groups") or {}).items(): + unit = cents.get(site) + if unit is None: + continue + for day, models in (group.get("by_day") or {}).items(): + slot = priced.setdefault(day, {}) + for model, credit in models.items(): + slot[model] = slot.get(model, 0.0) + float(credit) * unit daily = [] - for day in sorted(t["by_day"]): + for day in sorted(priced): if start_date and day < start_date: continue if end_date and day > end_date: continue - items = [{"name": m, "cost": round(c * cents_per_credit, 4)} - for m, c in sorted(t["by_day"][day].items()) if c > 0] + items = [{"name": m, "cost": round(c, 4)} + for m, c in sorted(priced[day].items()) if c > 0] try: ts = int(time.mktime(time.strptime(day, "%Y-%m-%d"))) except ValueError: @@ -1870,7 +1880,12 @@ def billing_usage(start_date: Optional[str] = None, end_date: Optional[str] = No total_cents = round(sum(sum(i["cost"] for i in d["line_items"]) for d in daily), 2) else: # 全量口径与 subscription 构成余额恒等式 total_cents = round(t["used_usd"] * 100, 2) - return {"object": "list", "total_usage": total_cents, "daily_costs": daily} + out = {"object": "list", "total_usage": total_cents, "daily_costs": daily} + if t.get("partial"): + out["partial"] = True + if detail.get("stale_accounts"): + out["stale_accounts"] = detail["stale_accounts"] + return out # 对外模型表：云端 /v3/config 同步结果优先，DEFAULT_MODELS 兜底补充 diff --git a/tests/test_credits.py b/tests/test_credits.py index 333258a..72e95b4 100644 --- a/tests/test_credits.py +++ b/tests/test_credits.py @@ -740,6 +740,10 @@ def test_billing_balance_identity(): # 区间过滤只统计窗口内明细 converter.CONFIG["usage_daily"] = {"by_day": {"2026-09-01": {"glm-5.3": 200.0}, "2026-09-05": {"glm-5.3": 50.0}}, + "groups": {"domestic": {"by_day": { + "2026-09-01": {"glm-5.3": 200.0}, + "2026-09-05": {"glm-5.3": 50.0}}, + "total_credits": 250.0, "requests": 3}}, "total_credits": 250.0, "requests": 3, "fetched_at": time.time()} filtered = converter.billing_usage("2026-09-04", "2026-09-30", None, None) @@ -751,6 +755,42 @@ def test_billing_balance_identity(): print("✅ test_billing_balance_identity") +def test_billing_usage_prices_each_day_by_site(): + """两站单价不同且用量发生在不同天：逐日金额必须按本站单价，而不是全局平均价。""" + import converter + with tempfile.TemporaryDirectory() as td: + led = credits.CreditLedger(Path(td) / "ledger.json") + led.update_credits("cn", {"credits": 100.0, "segments": [ + {"remaining": 100.0, "total": 200.0, "expires_at": None}], "intl": False}) + led.update_credits("ai", {"credits": 100.0, "segments": [ + {"remaining": 100.0, "total": 200.0, "expires_at": None}], "intl": True}) + saved = (converter.CONFIG.get("ledger"), converter.CONFIG.get("usage_daily")) + try: + converter.CONFIG["ledger"] = led + # 国内 100 credits @ $0.014/7.15 在 09-01；国际 100 credits @ $0.03 在 09-02 + converter.CONFIG["usage_daily"] = { + "by_day": {"2026-09-01": {"m": 100.0}, "2026-09-02": {"m": 100.0}}, + "groups": {"domestic": {"by_day": {"2026-09-01": {"m": 100.0}}, + "total_credits": 100.0, "requests": 1}, + "international": {"by_day": {"2026-09-02": {"m": 100.0}}, + "total_credits": 100.0, "requests": 1}}, + "total_credits": 200.0, "requests": 2, "fetched_at": time.time()} + usage = converter.billing_usage(None, None, None, None) + days = {d["timestamp"]: d["line_items"] for d in usage["daily_costs"]} + import time as _t + d1 = _t.mktime(_t.strptime("2026-09-01", "%Y-%m-%d")) + d2 = _t.mktime(_t.strptime("2026-09-02", "%Y-%m-%d")) + cn_cents = 100 * 0.014 / 7.15 * 100 # ≈ 19.58 美分 + assert abs(days[d1][0]["cost"] - cn_cents) < 0.01, days[d1] + assert abs(days[d2][0]["cost"] - 300.0) < 0.01, days[d2] # 100 × $0.03 = 300 美分 + # 恒等式：Σdaily ≈ total_usage（全量口径取 used_usd） + assert abs(sum(i["cost"] for d in usage["daily_costs"] for i in d["line_items"]) + - usage["total_usage"]) < 0.02 + finally: + converter.CONFIG["ledger"], converter.CONFIG["usage_daily"] = saved + print("✅ test_billing_usage_prices_each_day_by_site") + + def test_billing_intl_split(): """国内/国际分组折算：单价各按站点，合计与恒等式仍成立。""" import converter @@ -892,6 +932,7 @@ def test_model_catalog_cache(): test_fetch_request_usage_marks_partial_at_page_cap() test_fetch_request_usage_paging() test_billing_balance_identity() + test_billing_usage_prices_each_day_by_site() test_billing_intl_split() test_current_models_intl_condition() test_guard_model() From 5a63c9fa75d725c876e906b7e27bac9cdbcbb6d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:03:58 +0800 Subject: [PATCH 5/7] Record discarded tool-call generations and make the retry budget configurable Up to three extra generations after malformed tool calls consumed real credits, but only the final usage was recorded. Each discarded generation now lands in the request's attempts with its sequence number and total_tokens, and the budget moves from a hardcoded constant to --tool-call-max-retry / CODEBUDDY2API_TOOL_CALL_MAX_RETRY (default 3, 0 disables retries). Verified by a new endpoint test covering retry-with-usage recording and the zero budget (affected test files pass). --- converter.py | 16 +++++++++++---- tests/test_runtime_endpoints.py | 36 +++++++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/converter.py b/converter.py index 9dc72dc..91c16bd 100644 --- a/converter.py +++ b/converter.py @@ -2586,11 +2586,15 @@ async def _fetch_checked_chat(url, headers, body, model_name, rid, cred=None, *, observe_usage(result.get("usage") or {}) return result # 审核拒绝不是工具损坏，不因 required 工具选择而重复生成。 - if detector.detected or not body.get("tools") or tool_attempt >= _TOOL_CALL_MAX_RETRY: + budget = CONFIG.get("tool_call_max_retry", _TOOL_CALL_MAX_RETRY) + if detector.detected or not body.get("tools") or tool_attempt >= budget: raise UpstreamResponseError(502, b"Invalid upstream tool_calls after retries") tool_attempt += 1 - _log(f"[{rid}] tool_calls 损坏，重试 {tool_attempt}/{_TOOL_CALL_MAX_RETRY} | {model_name}") - + # 被丢弃的这次生成也是真实消耗：连同序号记进 attempts，账务不再只看见最后一次 + discarded = result.get("usage") or {} + observe_attempt("tool_args_retry", attempt=tool_attempt, max_attempts=budget, + total_tokens=discarded.get("total_tokens")) + _log(f"[{rid}] tool_calls 损坏，重试 {tool_attempt}/{budget} | {model_name}") async def _chat_sse_lines(url, headers, body, model_name, t0, rid, cred=None, *, aggregate=False): """提供公共 Chat SSE 行流；流式请求不做审核重试，正文检测缓冲有界。""" @@ -3026,6 +3030,9 @@ def main(): ap.add_argument("--log-body-limit", type=_nonnegative_int, metavar="BYTES", default=os.environ.get("CODEBUDDY2API_LOG_BODY_LIMIT", "65536"), help="每条正文日志的预览字节上限，默认 64 KiB；0 只记录摘要") + ap.add_argument("--tool-call-max-retry", type=_nonnegative_int, metavar="N", + default=os.environ.get("CODEBUDDY2API_TOOL_CALL_MAX_RETRY", "3"), + help="工具参数损坏时的额外生成上限，默认 3；0 表示不重试（每次额外生成都消耗额度）") ap.add_argument("--auto-trial", type=_boolean_arg, nargs="?", const=True, default=os.environ.get("CODEBUDDY2API_AUTO_TRIAL", "false"), help="自动领取国际 WorkBuddy 一次性体验积分，默认关闭") @@ -3035,7 +3042,8 @@ def main(): if args.command == "login": return login(site=args.site, open_browser=not args.no_browser) - for key in ("max_images", "image_policy", "max_request_bytes", "log_body_limit", "auto_trial"): + for key in ("max_images", "image_policy", "max_request_bytes", "log_body_limit", "auto_trial", + "tool_call_max_retry"): CONFIG[key] = getattr(args, key) CONFIG["api_key"] = args.api_key CONFIG["desensitize"] = args.desensitize diff --git a/tests/test_runtime_endpoints.py b/tests/test_runtime_endpoints.py index 04e3e87..f53ad64 100644 --- a/tests/test_runtime_endpoints.py +++ b/tests/test_runtime_endpoints.py @@ -188,6 +188,42 @@ def test_omitted_stream_defaults_to_nonstream_and_bad_type_rejected(self): self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.headers["content-type"], "application/json") + def test_discarded_tool_generations_are_recorded_with_usage(self): + """损坏工具调用触发的额外生成：每次丢弃都带用量记入 attempts；预算可配。""" + good = {"tool_calls": [{"index": 0, "id": "ok", "type": "function", + "function": {"name": "synthetic_tool", "arguments": "{}"}}]} + bad = {"tool_calls": [{"index": 0, "id": "bad", "type": "function", + "function": {"name": "synthetic_tool", "arguments": "{"}}]} + attempts = [] + with patch.object(converter, "observe_attempt", + side_effect=lambda stage, **kw: attempts.append((stage, kw))): + calls = {"n": 0} + def flaky(request): + calls["n"] += 1 + return httpx.Response(200, content=sse(bad if calls["n"] == 1 else good, "tool_calls")) + self.respond = flaky + self.requests.clear() + payload = payload_for(ROUTES[0], 0) + payload["tools"] = TOOLS + response = self.client.post(ROUTES[0], json=payload) + self.assertEqual(response.status_code, 200, response.text) + self.assertEqual(calls["n"], 2) + retries = [kw for stage, kw in attempts if stage == "tool_args_retry"] + self.assertEqual(len(retries), 1) + self.assertEqual(retries[0]["attempt"], 1) + self.assertIn("total_tokens", retries[0]) # 被丢弃的生成用量不再消失 + + converter.CONFIG["tool_call_max_retry"] = 0 + try: + self.respond = lambda request: httpx.Response(200, content=sse(bad, "tool_calls")) + self.requests.clear() + nonstream = dict(payload, stream=False) # 非流式：错误直接体现为 HTTP 状态码 + response = self.client.post(ROUTES[0], json=nonstream) + self.assertEqual(response.status_code, 502, response.text) + self.assertEqual(len(self.requests), 1) # 预算 0：不重试 + finally: + converter.CONFIG["tool_call_max_retry"] = 3 + def test_tool_metadata_policy_reaches_all_protocols(self): description = "Read sandbox data without destructive changes." schema = {"type": "object", "title": "Lookup inputs", "properties": { From 9f8e22ad28631ad0f30333aa3790513d0fa77aa7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E8=8F=8A?= <277378677+maiphucgiang@users.noreply.github.com> Date: Mon, 14 Sep 2026 20:17:32 +0800 Subject: [PATCH 6/7] Expire audit dedup rows with the same retention as their details The ingest dedup table had no time dimension and intentionally outlived all detail eviction, so it grew monotonically with unique requests and the detail byte budget could not bound the database file. Schema v2 adds created_at to ingest (migrated from v1 with timestamps backfilled from the matching detail row), expires dedup rows at the same retention cutoff with bounded batches, and reports them in pending_cleanup. Within the retention window dedup is unchanged; beyond it an ancient id may record again instead of being kept forever. Aggregates are still never deleted by budget or retention. Verified by a v1-migration regression test and the updated retention contract tests (test_audit_store.py passes). --- app/audit_store.py | 35 ++++++++++++++++++++++++++++------- tests/test_audit_store.py | 31 ++++++++++++++++++++++++++++--- 2 files changed, 56 insertions(+), 10 deletions(-) diff --git a/app/audit_store.py b/app/audit_store.py index 986f439..526e7f4 100644 --- a/app/audit_store.py +++ b/app/audit_store.py @@ -3,7 +3,8 @@ Methods are synchronous: ASGI callers must offload them. Lock and SQLite busy waits are capped at 250ms; SQL has a cooperative 1s progress deadline (not a hard wall-clock guarantee for filesystem I/O). Failures are observable, not retried. -Ingest deduplication and aggregates intentionally outlive all detail eviction. +Aggregates intentionally outlive all detail eviction. Ingest dedup rows expire with the +same retention cutoff as their details, so the dedup table cannot grow without bound. Detail accounting is transactional; indexed cleanup commits bounded batches. Large budget/retention reductions converge on subsequent writes or detail reads (including storage()), reported as pending_cleanup until complete. This is a @@ -23,7 +24,7 @@ import uuid from typing import Any -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 _CLEANUP_BATCH = 128 _CLEANUP_SECONDS = 0.05 METRICS = ("input_tokens", "output_tokens", "cache_read_tokens", "cache_creation_tokens", @@ -125,11 +126,11 @@ def __init__(self, path, max_bytes=256 * 1024 * 1024, retention_days=30, self._db.execute("BEGIN IMMEDIATE") version = self._db.execute("PRAGMA user_version").fetchone()[0] tables = self._db.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() - if version not in (0, SCHEMA_VERSION) or (version == 0 and tables): + if version not in (0, 1, SCHEMA_VERSION) or (version == 0 and tables): raise ValueError("unsupported audit schema") self._db.execute("CREATE TABLE IF NOT EXISTS state (id INTEGER PRIMARY KEY CHECK(id=1), epoch INTEGER NOT NULL, detail_generation INTEGER NOT NULL, cleared_at REAL NOT NULL)") self._db.execute("INSERT OR IGNORE INTO state VALUES(1,0,0,0)") - self._db.execute("CREATE TABLE IF NOT EXISTS ingest (id TEXT PRIMARY KEY, kind TEXT NOT NULL)") + self._db.execute("CREATE TABLE IF NOT EXISTS ingest (id TEXT PRIMARY KEY, kind TEXT NOT NULL, created_at REAL NOT NULL)") self._db.execute("CREATE TABLE IF NOT EXISTS requests (id TEXT PRIMARY KEY, started_at REAL NOT NULL, model TEXT, profile TEXT, credential TEXT, outcome TEXT, status_code INTEGER, payload TEXT NOT NULL, logical_bytes INTEGER NOT NULL)") self._db.execute("CREATE INDEX IF NOT EXISTS requests_time ON requests(started_at,id)") self._db.execute("CREATE TABLE IF NOT EXISTS attempts (request_id TEXT NOT NULL, ordinal INTEGER NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(request_id,ordinal))") @@ -138,6 +139,7 @@ def __init__(self, path, max_bytes=256 * 1024 * 1024, retention_days=30, self._init_accounting() for table in ("stats_hourly", "stats_daily", "stats_totals"): self._db.execute(f"CREATE TABLE IF NOT EXISTS {table} (bucket INTEGER NOT NULL, dimension TEXT NOT NULL, dimension_key TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(bucket,dimension,dimension_key))") + self._migrate_ingest_time() self._db.execute(f"PRAGMA user_version={SCHEMA_VERSION}") self._epoch = self._db.execute("SELECT epoch FROM state").fetchone()[0] self._db.execute("COMMIT") @@ -145,6 +147,17 @@ def __init__(self, path, max_bytes=256 * 1024 * 1024, retention_days=30, self._db.close() raise + def _migrate_ingest_time(self): + # v1 → v2：去重表获得时间维度。旧行用对应明细的时间回填，没有明细的按现在计， + # 随后与明细同截止期过期，不再无限期滞留。 + columns = [row[1] for row in self._db.execute("PRAGMA table_info(ingest)").fetchall()] + if "created_at" not in columns: + self._db.execute("ALTER TABLE ingest ADD COLUMN created_at REAL") + self._db.execute("""UPDATE ingest SET created_at=COALESCE( + (SELECT started_at FROM requests WHERE requests.id=ingest.id), + (SELECT started_at FROM events WHERE events.id=ingest.id), ?)""", (time.time(),)) + self._db.execute("CREATE INDEX IF NOT EXISTS ingest_time ON ingest(created_at)") + def _init_accounting(self): # Additive v1 migration: scan legacy details only once, under the same # write transaction that installs triggers. Reopening never recounts. @@ -298,7 +311,7 @@ def commit(): state = self._db.execute("SELECT * FROM state").fetchone() if data["epoch"] != state["epoch"]: return {"ok": True, "recorded": False, "reason": "stale_epoch"} - if not self._db.execute("INSERT OR IGNORE INTO ingest VALUES(?, 'request')", (data["id"],)).rowcount: + if not self._db.execute("INSERT OR IGNORE INTO ingest VALUES(?, 'request', ?)", (data["id"], data["started_at"])).rowcount: return {"ok": True, "recorded": False, "reason": "duplicate"} self._aggregate(data) generation = record.get("detail_generation") @@ -338,12 +351,20 @@ def _expire(self, retention_days=None, deadline=None): if time.monotonic() >= deadline: break self._db.execute(f"DELETE FROM {table} WHERE id=?", (record_id,)) + # 去重行与明细同截止期过期：保留期之外不再防重放，也不再无限增长 + for row in self._db.execute("SELECT id FROM ingest WHERE created_at= deadline: + break + self._db.execute("DELETE FROM ingest WHERE id=?", (row[0],)) def _cleanup_pending(self, budget=None, retention_days=None): budget = self.max_bytes if budget is None else budget cutoff = time.time() - (self.retention_days if retention_days is None else retention_days) * 86400 accounting = self._db.execute("SELECT logical_bytes,cleanup_target FROM detail_accounting WHERE id=1").fetchone() - return (accounting[0] > budget or accounting[1] is not None or any( + ingest_expired = self._db.execute( + "SELECT 1 FROM ingest WHERE created_at budget or accounting[1] is not None or ingest_expired or any( self._db.execute(f"SELECT 1 FROM {table} WHERE started_at Date: Mon, 14 Sep 2026 20:19:56 +0800 Subject: [PATCH 7/7] Document billing integrity markers and the tool retry budget flag Sync the advanced guides: partial/stale_accounts semantics on billing endpoints, per-site daily pricing, the per-account snapshot behavior on sync failures, and the new --tool-call-max-retry option with its credit cost note. --- docs/advanced.md | 9 ++++++++- docs/advanced.zh-CN.md | 9 ++++++++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/docs/advanced.md b/docs/advanced.md index 4032eae..f745fe1 100644 --- a/docs/advanced.md +++ b/docs/advanced.md @@ -29,6 +29,7 @@ Compose explicitly passes some environment variables and CLI flags, so deleting | `--auto-trial [true/false]` | `false` | Attempt one-time international WorkBuddy trial-credit claims | | `--max-images` | `16` | Total images per request; `0` permits no images | | `--image-policy` | `truncate` | Keep newest images; `error` rejects excess images with 413 | +| `--tool-call-max-retry` | `3` | Extra generations after malformed tool calls (each consumes credits); `0` disables retries | | `--max-request-bytes` | `33554432` | Positive byte limit for the processed upstream JSON | | `--log-body-limit` | `65536` | Text-log body preview bytes; `0` logs summaries only, not the SQLite diagnostic budget | @@ -132,6 +133,12 @@ Credential domain / token issuer determine the product identity. Chat and refres - `/v1/messages/count_tokens` returns a character-based heuristic estimate for budgeting, not an exact count. - Text logs and SQLite auditing have separate budgets. Logs contain bounded, redacted previews, not complete original requests. Treat logs, credential exports and backups as private data. +## Billing data integrity + +- Balances and usage are paginated in full; when a page cap is hit or an account's sync fails, responses carry `partial: true` (and `stale_accounts`) instead of pretending to be exact. +- A failed account keeps its last good snapshot; HTTP 200 responses with a failing business code or missing structure are treated as errors and never overwrite history. +- daily_costs are priced per site per day at that site's price, not at one blended average. + ## Troubleshooting and retries | Symptom | Behavior / action | @@ -143,7 +150,7 @@ Credential domain / token issuer determine the product identity. Chat and refres | Upstream `service info not found` (code 11102) | That backend does not serve the model at all: avoid it for the `(backend, model)` pair, route the model to another backend, and return 404 when none has it. Half-open after 6 h, exponential backoff up to 24 h, cleared at once by one successful call; inspect via `GET /admin/model-blocks` | | Connection setup failure | Retry only `ConnectError` / `ConnectTimeout` once after backoff | | Post-send disconnect, read/write timeout or HTTP error | No network replay, avoiding duplicate billing; logs include exception type and elapsed time | -| Malformed tool calls | Aggregate validation permits up to 3 additional generations, potentially consuming credits; exhaustion returns an error | +| Malformed tool calls | Aggregate validation permits up to `--tool-call-max-retry` (default 3) additional generations, each consuming credits and recorded with its usage in the attempt details; exhaustion returns an error | | Empty or truncated upstream stream | No valid output, a missing end marker or an error is not reported as success | | Content-filter rejection | With desensitization and `--no-compact`, a complete non-streaming filter-only rejection may receive one shorter-template retry on the same account. No streaming filter retry, circuit opening or account rotation | | Slow responses | Inspect timing and failed attempts in the WebUI, then choose a faster model supported by the account | diff --git a/docs/advanced.zh-CN.md b/docs/advanced.zh-CN.md index 956e9ed..57bf7eb 100644 --- a/docs/advanced.zh-CN.md +++ b/docs/advanced.zh-CN.md @@ -29,6 +29,7 @@ Compose 会显式传入部分环境变量及 CLI 参数，删除 `.env` 中的 | `--auto-trial [true/false]` | `false` | 尝试领取国际 WorkBuddy 一次性体验积分 | | `--max-images` | `16` | 单请求图片总数；`0` 不允许图片 | | `--image-policy` | `truncate` | 保留最新图片；设为 `error` 时超限返回 413 | +| `--tool-call-max-retry` | `3` | 工具参数损坏时的额外生成上限（每次都消耗额度）；`0` 不重试 | | `--max-request-bytes` | `33554432` | 处理后的上游 JSON 字节上限，须为正整数 | | `--log-body-limit` | `65536` | 兼容文本日志正文预览字节；`0` 只记摘要，不控制 SQLite 诊断预算 | @@ -132,6 +133,12 @@ WebUI 可以直接上传文件；以下限制针对 `POST /admin/credentials` - `/v1/messages/count_tokens` 返回字符启发式估算值，仅作预算参考，不是精确计数。 - 兼容文本日志和 SQLite 审计使用独立预算；日志仅记录有界、脱敏预览，不是完整原始请求。日志、凭证导出和备份仍须按私有数据保管。 +## 账务数据完整性 + +- 余额/用量来自官方接口的分页遍历；达到页数上限或任一账号同步失败时，响应带 `partial: true`（及 `stale_accounts` 列表），不伪装为精确全量。 +- 单账号同步失败保留其上次成功快照；接口返回 HTTP 200 但业务码失败或结构缺失时按错误处理，不覆盖历史。 +- daily_costs 按各站（国内/国际）本站单价逐日折算，不再使用全局平均单价。 + ## 故障与重试 | 现象 | 处理与边界 | @@ -143,7 +150,7 @@ WebUI 可以直接上传文件；以下限制针对 `POST /admin/credentials` | 上游 `service info not found`（11102） | 该后端根本不服务这个模型：按 (后端, 模型) 避让，把该模型派给其他后端，全部后端都没有时返回 404。6 小时后半开放行重试，反复命中最长退避 24 小时，一次成功调用即刻解除；可用 `GET /admin/model-blocks` 查看 | | 建连失败 | 仅 `ConnectError` / `ConnectTimeout` 退避重试一次 | | 发送后断连、读写超时、HTTP 错误 | 不做网络重放，避免重复计费；日志记录异常类型与耗时 | -| 工具参数损坏 | 聚合校验失败最多额外生成 3 次，可能消耗更多额度；耗尽后返回错误 | +| 工具参数损坏 | 聚合校验失败按 `--tool-call-max-retry`（默认 3）额外生成，可能消耗更多额度；被丢弃的生成带用量记入尝试明细；耗尽后返回错误 | | 上游空流或残流 | 没有有效输出、缺少结束标记或包含错误的流不伪装为成功 | | 内容审核拒绝 | 脱敏 + `--no-compact` 下，仅完整非流式纯拒绝且模板确实缩短时，最多同账号兜底一次；流式不做审核重试，也不因此熔断或切号 | | 响应慢 | 在 WebUI 查看耗时与失败尝试，再选择当前账号支持的更快模型 |