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 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,41 +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 = resp.get("Accounts") or payload.get("data", {}).get("accounts") or [] - 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} @@ -527,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} @@ -535,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] @@ -548,14 +580,22 @@ 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, 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,9 +606,12 @@ 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} + else: + partial = True # 达到页数上限仍可能有剩余：标记不完整，不装作全量 + return {"by_day": by_day, "total_credits": round(total_credits, 2), "requests": requests, + "partial": partial} # --------------------------------------------------------------------------- @@ -653,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..91c16bd 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 模块默认 @@ -1773,6 +1813,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 {}} @@ -1808,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: @@ -1828,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 兜底补充 @@ -2529,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 行流；流式请求不做审核重试，正文检测缓冲有界。""" @@ -2969,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 一次性体验积分，默认关闭") @@ -2978,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/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 查看耗时与失败尝试，再选择当前账号支持的更快模型 | diff --git a/tests/test_audit_store.py b/tests/test_audit_store.py index 8ea554e..6badbb3 100644 --- a/tests/test_audit_store.py +++ b/tests/test_audit_store.py @@ -94,7 +94,7 @@ def test_all_dimensions_and_periods_persist_without_details(self): self.store.record_request(self.record(input_tokens=0)) self.store.clear("details") with closing(sqlite3.connect(self.path)) as db: - self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 1) + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 2) for table in ("stats_hourly", "stats_daily", "stats_totals"): rows = db.execute(f"SELECT dimension,payload FROM {table}").fetchall() self.assertEqual({r[0] for r in rows}, {"global", "model", "profile", "credential"}) @@ -135,6 +135,29 @@ def test_thread_lock_timeout_is_visible(self): self.assertFalse(result["ok"]) self.assertEqual(self.store.storage()["last_error"], "TimeoutError") + def test_v1_ingest_table_migrates_with_backfilled_timestamps(self): + """v1 库（ingest 无 created_at）打开即迁移：列回填、索引建立、保留期内仍可去重。""" + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "audit.sqlite3" + with closing(sqlite3.connect(str(path))) as db: + db.execute("BEGIN IMMEDIATE") + db.execute("CREATE TABLE ingest (id TEXT PRIMARY KEY, kind TEXT NOT NULL)") # v1 形状 + db.execute("INSERT INTO ingest VALUES('legacy-req', 'request')") + db.execute("PRAGMA user_version=1") + db.execute("COMMIT") + store = AuditStore(path) + try: + with closing(sqlite3.connect(str(path))) as db: + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 2) + columns = [row[1] for row in db.execute("PRAGMA table_info(ingest)").fetchall()] + self.assertIn("created_at", columns) + self.assertGreater(db.execute( + "SELECT created_at FROM ingest WHERE id='legacy-req'").fetchone()[0], 0) + self.assertTrue(store.record_request(self.record("fresh"))["recorded"]) + self.assertEqual(store.record_request(self.record("legacy-req"))["reason"], "duplicate") + finally: + store.close() + def test_concurrent_duplicate_ingest_exactly_once(self): record = self.record() with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor: @@ -152,7 +175,8 @@ def test_logical_budget_and_retention_never_delete_aggregate(self): self.store.record_request(self.record("old", started_at=time.time() - 31 * 86400)) self.assertIsNone(self.store.get_request("old")) self.assertEqual(self.store.dashboard(90)["summary"]["requests"], 2) - self.assertEqual(self.store.record_request(self.record("old"))["reason"], "duplicate") + # 去重行与明细同截止期过期：超过保留期的旧 ID 不再判定为重复（防重放仅限保留期内） + self.assertTrue(self.store.record_request(self.record("old"))["recorded"]) health = self.store.storage() for name in ("db_bytes", "wal_bytes", "shm_bytes"): self.assertIsInstance(health[name], int) @@ -398,7 +422,8 @@ def test_incremental_retention_hides_expired_details_and_preserves_stats(self): if not health["pending_cleanup"]: break self.assertFalse(health["pending_cleanup"]) - self.assertEqual(self.assert_accounting(), (0, 0, 0, count)) + # 去重行与明细同截止期过期：ingest 数归零，聚合统计不受影响 + self.assertEqual(self.assert_accounting(), (0, 0, 0, 0)) self.assertEqual(self.aggregate_snapshot(), aggregates) self.assertFalse(health["degraded"]) diff --git a/tests/test_credits.py b/tests/test_credits.py index df77d6f..72e95b4 100644 --- a/tests/test_credits.py +++ b/tests/test_credits.py @@ -502,6 +502,163 @@ 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_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_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 = [ @@ -537,7 +694,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 @@ -583,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) @@ -594,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 @@ -728,8 +925,14 @@ 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_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() test_billing_balance_identity() + test_billing_usage_prices_each_day_by_site() test_billing_intl_split() test_current_models_intl_condition() test_guard_model() 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": {