feat: cost/quota set — dual-column priced reports, creator split, hard cutoff, approval-timeout auto-threshold

- CostStore: per-model breakdown, report_by_creator (JOIN tasks.created_by),
  dual-column reports with priced_usd_micros, unpriced_models listed
  separately and a billing note (proxy price, not the relay bill);
  _report itself untouched, old exact-key tests keep passing.
- Hard cutoff: cost.quota_cutoff_reason is the single USD/unpriced-cap
  decision; worker._quota_exceeded now delegates to it (identical
  messages); claim gate, turn boundary and poll kill stop spend and
  record interrupted + quota reason + audit; auto-resume re-checks
  live quota before requeueing.
- Approval timeout (G7 ops metric): cloud/approval_timeout.py —
  default stays 600s, >20% timeout rate recommends 1800s, <5% keeps,
  mid-range keeps; stats + recommendation + info log; wired into
  /metrics (member projects, no log spam) and a global-admin
  /admin/ops/approval-timeout endpoint (sweeps overdue first).
- Server: task/project cost endpoints return dual-column reports;
  new GET /users/me/cost and GET /admin/cost-by-creator.
- Tests: cloud/tests/test_cost_quota.py, 10 tests.
This commit is contained in:
host committed 2026-11-04 20:54:05 +00:00
1 parent cc32f51502
commit e079b9b44a
5 files changed
+871 -12

No files matched your search

+145
View File
@@ -0,0 +1,145 @@
"""审批超时率自动调阈(G7 运维指标,v1.1 契约口径)。
只依赖标准库与 cloud.approvals 的建表函数,不 import server /
worker,避免循环 import。
口径(与 v1.1 冻结条一致):
- 默认审批超时仍是 10 分钟(600 秒)。注意:这是本调阈口径
的默认值,不是 approvals.py elicitation 队列的 3600 秒
默认值——两条审批线各有默认值,本模块绝不改写后者。
- 超时率 = 已终态审批项中 status='expired' 的占比;已终态 =
approved + denied + expired。pending 不计入分母(还没结果
的项不能算超时也不能算准时);未扫荡的过期 pending 同样
不计入——扫荡发生在审批列表与运维端点,统计前应先扫荡。
- 超时率 > 20% 时建议调到 30 分钟(1800 秒);< 5% 时不动;
5%~20% 之间同样不动(只升不降,降阈须人工决定,本模块
绝不自动把超时缩回去)。边界值本身不动:>0.20 才升,
==0.20 与 ==0.05 都按不动处理。
- 本模块只算建议并记日志,不自动改写任何默认值或已建审批
项的 expires_at:调阈生效须运维按建议改配置后重启/重载,
建议值进 /metrics 与运维端点,审计靠日志行(见
evaluate_approval_timeout 的 info 日志)。
"""
import logging
from cloud.approvals import ensure_approvals_table
logger = logging.getLogger(__name__)
DEFAULT_TIMEOUT_SECONDS = 600
RAISED_TIMEOUT_SECONDS = 1800
RAISE_THRESHOLD = 0.20
KEEP_THRESHOLD = 0.05
def timeout_rate(expired_count, decided_count):
"""超时率:已终态为 0 时回 None(不是 0%,是算不出来)。"""
try:
decided = int(decided_count)
expired = int(expired_count)
except (TypeError, ValueError):
return None
if decided <= 0:
return None
if expired < 0:
expired = 0
if expired > decided:
expired = decided
return expired / decided
def recommend_approval_timeout(expired_count, decided_count,
current_seconds=DEFAULT_TIMEOUT_SECONDS):
"""纯计算函数:输入计数,输出建议(无 I/O,可单测三档)。
返回键:expired_count、decided_count、timeout_rate、
current_seconds、recommended_seconds、action(raise_to_30m
或 keep)、reason。current_seconds 非法时按默认值计,不抛。
"""
try:
current = int(current_seconds)
except (TypeError, ValueError):
current = DEFAULT_TIMEOUT_SECONDS
if current <= 0:
current = DEFAULT_TIMEOUT_SECONDS
rate = timeout_rate(expired_count, decided_count)
if rate is not None and rate > RAISE_THRESHOLD:
action = "raise_to_30m"
recommended = RAISED_TIMEOUT_SECONDS
reason = ("approval timeout rate %.4f > %.2f, "
"recommend %d seconds" % (rate, RAISE_THRESHOLD,
recommended))
else:
action = "keep"
recommended = current
if rate is None:
reason = "no decided approvals, keep %d seconds" % current
else:
reason = ("approval timeout rate %.4f <= %.2f, "
"keep %d seconds" % (rate, RAISE_THRESHOLD,
current))
return {
"expired_count": int(expired_count or 0),
"decided_count": int(decided_count or 0),
"timeout_rate": rate,
"current_seconds": current,
"recommended_seconds": recommended,
"action": action,
"reason": reason,
}
def approval_timeout_stats(conn, project_ids=None):
"""从 task_approvals 统计已终态与超时数(纯读,不扫荡)。
project_ids 为 None 时全库统计;否则只计这些项目(/metrics
的成员项目口径)。表不存在时先幂等建表(空表即零统计)。
"""
ensure_approvals_table(conn)
query = ("SELECT COUNT(*), COALESCE(SUM(CASE WHEN status = 'expired'"
" THEN 1 ELSE 0 END), 0), COALESCE(SUM(CASE WHEN"
" status = 'pending' THEN 1 ELSE 0 END), 0)"
" FROM task_approvals WHERE status IN"
" ('approved', 'denied', 'expired', 'pending')")
args = []
if project_ids is not None:
ids = sorted(set(project_ids))
if not ids:
return {"decided_count": 0, "expired_count": 0,
"pending_count": 0}
query += " AND project_id IN (%s)" % ", ".join(
"?" for _ in ids)
args.extend(ids)
row = conn.execute(query, args).fetchone()
decided = int(row[0] or 0) - int(row[2] or 0)
return {
"decided_count": decided,
"expired_count": int(row[1] or 0),
"pending_count": int(row[2] or 0),
}
def evaluate_approval_timeout(conn, current_seconds=DEFAULT_TIMEOUT_SECONDS,
project_ids=None, log=None):
"""统计 + 建议 + 日志的一站式运维入口。
返回 approval_timeout_stats 与 recommend_approval_timeout
的合并 dict,并写一条 info 日志(log 未给时用模块 logger)。
调用方在需要“过期 pending 也算超时”时须先调 approvals
的 expire_overdue 扫荡——本函数自己不写库。
"""
stats = approval_timeout_stats(conn, project_ids=project_ids)
result = recommend_approval_timeout(
stats["expired_count"], stats["decided_count"],
current_seconds=current_seconds)
result.update(stats)
(log or logger).info(
"approval_timeout metric: decided=%d expired=%d pending=%d "
"rate=%s action=%s recommended_seconds=%d",
result["decided_count"], result["expired_count"],
result["pending_count"],
("%.4f" % result["timeout_rate"]
if result["timeout_rate"] is not None else "unknown"),
result["action"], result["recommended_seconds"])
return result
+204
View File
@@ -135,6 +135,39 @@ def price_usd_micros(model, input_tokens, cached_input_tokens,
return int(round(usd * 1000000)) return int(round(usd * 1000000))
BILLING_NOTE = (
"价格表代理价(经 CLOUD_MODEL_PRICES_JSON / CLOUD_MODEL_PRICES_FILE"
" 覆盖层),非中转实际账单;未定价 token 不计入美元、不等于免费。"
)
def quota_cutoff_reason(quota_usd_micros, used_usd_micros,
unpriced_used_tokens, month, unpriced_cap=None):
"""配额硬切断的唯一判定函数(worker 执行侧与单测共用)。
返回 None 表示未超限、可继续;返回 quota_exceeded 文本表示
必须立即停:调用方不再开新回合、不再派生花钱路径,任务以
interrupted + 该原因收尾(原因写进 error_text 并记审计),
不是只告警。quota 为 None(管理员显式不限额)时不挡;USD
配额与无价 token 上限是两道独立的门,任一超限即停。
文本格式与 worker 旧分支逐字一致,旧单测按此前缀断言。
"""
if quota_usd_micros is None:
return None
if int(used_usd_micros or 0) >= int(quota_usd_micros):
return ("quota_exceeded: 本月配额已用完(已用 %d / "
"配额 %d 微美元,%s)"
% (int(used_usd_micros or 0), int(quota_usd_micros),
month))
cap = unpriced_token_cap() if unpriced_cap is None \
else int(unpriced_cap)
if int(unpriced_used_tokens or 0) >= cap:
return ("quota_exceeded: 本月无价模型 token 已用完"
"(已用 %d / 上限 %d,%s)"
% (int(unpriced_used_tokens or 0), cap, month))
return None
def unpriced_token_cap(): def unpriced_token_cap():
"""Monthly unpriced-token ceiling for users who have a quota row. """Monthly unpriced-token ceiling for users who have a quota row.
@@ -397,3 +430,174 @@ class CostStore:
def report_by_day(self, day) -> dict: def report_by_day(self, day) -> dict:
"""Aggregate cost records for one UTC day (YYYY-MM-DD).""" """Aggregate cost records for one UTC day (YYYY-MM-DD)."""
return self._report("day = ?", (day,)) return self._report("day = ?", (day,))
def _report_joined(self, where, params):
"""_report 的 JOIN 版:按 tasks.created_by 归属汇总。
cost_records 本身没有创建者列,归属只能经 tasks JOIN;
协调员子任务行无 created_by 时跟父任务走的口径在
worker 侧,库层这里只按行上的 created_by 计——无属主
任务的成本不归任何人,不会被静默算进某个创建者。
"""
conn = self._new_conn()
try:
row = conn.execute(
"""SELECT
COALESCE(SUM(c.total_tokens), 0),
COALESCE(SUM(c.usd_micros), 0),
COUNT(*),
COALESCE(SUM(c.fallback_count), 0),
COALESCE(SUM(CASE WHEN c.source = 'actual'
THEN c.total_tokens ELSE 0 END), 0),
COALESCE(SUM(CASE WHEN c.source = 'estimated'
THEN c.total_tokens ELSE 0 END), 0),
COALESCE(SUM(c.cached_input_tokens), 0),
COALESCE(SUM(CASE WHEN c.source = 'actual'
AND c.usd_micros IS NULL
THEN c.total_tokens ELSE 0 END), 0),
COALESCE(SUM(c.completions), 0)
FROM cost_records c JOIN tasks t
ON c.task_id = t.id WHERE %s""" % where,
params,
).fetchone()
finally:
if conn is not self._conn:
conn.close()
(total_tokens, usd_micros, turns, fallback_count, actual,
estimated, cached, unpriced, completions) = row
return {
"total_tokens": int(total_tokens),
"usd_micros": int(usd_micros),
"turns": int(turns),
"fallback_count": int(fallback_count),
"actual_tokens": int(actual),
"estimated_tokens": int(estimated),
"cached_input_tokens": int(cached),
"unpriced_actual_tokens": int(unpriced),
"llm_requests": int(completions),
}
def report_by_creator(self, creator_id, month=None) -> dict:
"""按任务创建者汇总:当月(UTC,YYYY-MM)或全部。
month 为 None 时不过滤月份;month 格式须为 YYYY-MM,
否则 ValueError(调用方转 400)。返回 _report 同款键
外加 creator_id 与 month,方便分账报表直接按人对账。
"""
if month is not None:
if not isinstance(month, str) or len(month) != 7 \
or month[4] != "-":
raise ValueError("month must be YYYY-MM")
report = self._report_joined(
"t.created_by = ? AND c.day LIKE ?",
(creator_id, month + "%"))
else:
report = self._report_joined(
"t.created_by = ?", (creator_id,))
report["creator_id"] = creator_id
report["month"] = month
return report
def model_breakdown(self, task_id=None, project_id=None,
creator_id=None, month=None) -> list:
"""按模型分列:恰好指定 task / project / creator 其中一种。
每行:model、total_tokens、turns、llm_requests、
priced_usd_micros(该模型无任何有价行时为 None,不写 0)、
unpriced_tokens(source='actual' 且 usd IS NULL 的行,
估算行不计)、priced(是否有任一有价行)。未定价模型
由 unpriced_tokens > 0 识别,调用方用它单独列出,绝不
把 priced_usd_micros=None 的模型按 0 美元合并进合计。
creator 口径经 tasks JOIN,month 同 report_by_creator。
"""
scopes = [s for s in (task_id, project_id, creator_id)
if s is not None]
if len(scopes) != 1:
raise ValueError(
"exactly one of task_id, project_id, creator_id required")
join = ""
if task_id is not None:
where, params = "c.task_id = ?", (task_id,)
elif project_id is not None:
where, params = "c.project_id = ?", (project_id,)
else:
join = " JOIN tasks t ON c.task_id = t.id"
where, params = "t.created_by = ?", (creator_id,)
if month is not None:
if not isinstance(month, str) or len(month) != 7 \
or month[4] != "-":
raise ValueError("month must be YYYY-MM")
where += " AND c.day LIKE ?"
params = params + (month + "%",)
conn = self._new_conn()
try:
rows = conn.execute(
"""SELECT c.model,
COALESCE(SUM(c.total_tokens), 0),
COUNT(*),
COALESCE(SUM(c.completions), 0),
SUM(c.usd_micros),
SUM(CASE WHEN c.usd_micros IS NOT NULL
THEN 1 ELSE 0 END),
COALESCE(SUM(CASE WHEN c.source = 'actual'
AND c.usd_micros IS NULL
THEN c.total_tokens ELSE 0 END), 0)
FROM cost_records c%s WHERE %s
GROUP BY c.model
ORDER BY SUM(c.total_tokens) DESC, c.model"""
% (join, where),
params,
).fetchall()
finally:
if conn is not self._conn:
conn.close()
out = []
for row in rows:
(model, total, turns, completions, usd_sum, priced_rows,
unpriced) = row
out.append({
"model": model or "unknown",
"total_tokens": int(total),
"turns": int(turns),
"llm_requests": int(completions),
"priced_usd_micros": (int(usd_sum) if priced_rows else None),
"unpriced_tokens": int(unpriced),
"priced": bool(priced_rows),
})
return out
def _dual_column(self, base, breakdown):
"""双列报告:在 _report 结果上加第二列与分列明细。
_report 本身不动(旧单测按其精确键集断言);双列是
新方法返回的扩展:priced_usd_micros 与 usd_micros 同值
(只计有价行之和),unpriced_models 单独列出未定价
模型,billing_note 明确标注代理价、非中转实际账单。
"""
report = dict(base)
report["priced_usd_micros"] = report["usd_micros"]
report["unpriced_models"] = sorted(
row["model"] for row in breakdown
if row["unpriced_tokens"] > 0)
report["by_model"] = breakdown
report["billing_note"] = BILLING_NOTE
return report
def dual_column_report_by_task(self, task_id) -> dict:
"""单任务双列报告(见 _dual_column)。"""
return self._dual_column(
self.report_by_task(task_id),
self.model_breakdown(task_id=task_id))
def dual_column_report_by_project(self, project_id) -> dict:
"""单项目双列报告(见 _dual_column)。"""
return self._dual_column(
self.report_by_project(project_id),
self.model_breakdown(project_id=project_id))
def dual_column_report_by_creator(self, creator_id,
month=None) -> dict:
"""按创建者双列报告(见 _dual_column)。"""
return self._dual_column(
self.report_by_creator(creator_id, month=month),
self.model_breakdown(creator_id=creator_id, month=month))
+116 -2
View File
@@ -143,6 +143,9 @@ from cloud.todos import get_todo as _todo_get
from cloud.todos import list_todos as _todo_list from cloud.todos import list_todos as _todo_list
from cloud.todos import update_todo as _todo_update from cloud.todos import update_todo as _todo_update
from cloud.automations import triage_inbox as _auto_triage from cloud.automations import triage_inbox as _auto_triage
from cloud.approval_timeout import approval_timeout_stats as _appr_timeout_stats
from cloud.approval_timeout import evaluate_approval_timeout as _appr_timeout_eval
from cloud.approval_timeout import recommend_approval_timeout as _appr_timeout_recommend
from cloud.approvals import create_approval as _appr_create from cloud.approvals import create_approval as _appr_create
from cloud.approvals import decide_approval as _appr_decide from cloud.approvals import decide_approval as _appr_decide
from cloud.approvals import ensure_approvals_table as _appr_ensure from cloud.approvals import ensure_approvals_table as _appr_ensure
@@ -3107,6 +3110,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
self._send_json(404, {"error": "not found"}) self._send_json(404, {"error": "not found"})
elif path == "/users/me/quota" and method == "GET": elif path == "/users/me/quota" and method == "GET":
self._user_quota_self(user_id) self._user_quota_self(user_id)
elif path == "/users/me/cost" and method == "GET":
self._user_cost_self(user_id, query)
elif path == "/admin/cost-by-creator" and method == "GET":
self._admin_cost_by_creator(user_id, query)
elif path == "/admin/ops/approval-timeout" and method == "GET":
self._admin_approval_timeout_ops(user_id)
elif path == "/users/me/profile" and method == "GET": elif path == "/users/me/profile" and method == "GET":
self._identity_profile_get(user_id) self._identity_profile_get(user_id)
elif path == "/users/me/profile" and method in ("PUT", "POST"): elif path == "/users/me/profile" and method in ("PUT", "POST"):
@@ -4726,7 +4735,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
def _project_cost(self, project_id, query): def _project_cost(self, project_id, query):
store = CostStore(self.server.db_path) store = CostStore(self.server.db_path)
try: try:
overall = store.report_by_project(project_id) # 双列:美元只计有价行,未定价模型单独列在
# unpriced_models,billing_note 标注代理价、非账单。
overall = store.dual_column_report_by_project(project_id)
conn = store._conn conn = store._conn
conn.row_factory = sqlite3.Row conn.row_factory = sqlite3.Row
days = (query.get("days") or [None])[0] days = (query.get("days") or [None])[0]
@@ -5034,7 +5045,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
return return
store = CostStore(self.server.db_path) store = CostStore(self.server.db_path)
try: try:
report = store.report_by_task(task_id) # 双列:与项目成本同款,旧键全保留,只加双列键。
report = store.dual_column_report_by_task(task_id)
finally: finally:
try: try:
store._conn.close() store._conn.close()
@@ -9362,6 +9374,93 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
payload["unpriced_token_cap"] = unpriced_token_cap() payload["unpriced_token_cap"] = unpriced_token_cap()
self._send_json(200, payload) self._send_json(200, payload)
def _cost_by_creator_payload(self, target, month):
"""创建者分账双列载荷(自查与管理端共用)。
month 为 None 时不过滤月份;否则须为 YYYY-MM,否则
ValueError(调用方转 400)。配额进度与成本双列同回:
quota 是当月美元配额(与 _quota_status 同口径),cost
是该创建者名下任务的成本(经 tasks.created_by JOIN)。
"""
quota, used, cur_month = self._quota_status(target)
# 端点默认当月(与配额窗口一致),不传 month 时成本与
# 配额同为当月;CostStore 层 month=None 仍表示全部时间。
eff_month = month if month is not None else cur_month
store = CostStore(self.server.db_path)
try:
cost = store.dual_column_report_by_creator(target,
month=eff_month)
finally:
try:
store._conn.close()
except Exception:
pass
return {
"user_id": target,
"month": eff_month,
"quota_usd_micros": quota,
"used_usd_micros": used,
"remaining_usd_micros": (quota - used)
if quota is not None else None,
"unpriced_tokens_used": self._quota_unpriced_tokens(
target, eff_month),
"unpriced_token_cap": unpriced_token_cap(),
"cost": cost,
}
def _user_cost_self(self, user_id, query):
month = (query.get("month") or [None])[0]
try:
payload = self._cost_by_creator_payload(user_id, month)
except ValueError:
self._send_json(400, {"error": "month must be YYYY-MM"})
return
self._send_json(200, payload)
def _admin_cost_by_creator(self, user_id, query):
if not self._is_global_admin(user_id):
self._audit_admin(user_id, "admin_quota_denied",
"get cost by creator")
self._send_json(403, {"error": "forbidden"})
return
target = (query.get("user_id") or [None])[0]
if not target or not isinstance(target, str):
self._send_json(400, {"error": "user_id required"})
return
if not self._user_exists(target):
self._send_json(404, {"error": "user not found"})
return
month = (query.get("month") or [None])[0]
try:
payload = self._cost_by_creator_payload(target, month)
except ValueError:
self._send_json(400, {"error": "month must be YYYY-MM"})
return
self._send_json(200, payload)
def _admin_approval_timeout_ops(self, user_id):
"""审批超时调阈运维指标(全局管理员只读)。
先扫荡过期 pending(与审批列表同款:过期即 expired,
计入超时率),再统计 + 建议 + 记日志。返回里
recommended_seconds 是建议值,不是已生效值——默认
仍 600 秒,调阈生效须运维按建议改配置,见
cloud/approval_timeout 模块 docstring。
"""
if not self._is_global_admin(user_id):
self._audit_admin(user_id, "admin_quota_denied",
"get approval timeout ops")
self._send_json(403, {"error": "forbidden"})
return
conn = self._db()
try:
_appr_ensure(conn)
_appr_expire(conn)
result = _appr_timeout_eval(conn)
finally:
conn.close()
self._send_json(200, result)
def _admin_quota_get(self, user_id, query): def _admin_quota_get(self, user_id, query):
if not self._is_global_admin(user_id): if not self._is_global_admin(user_id):
self._audit_admin(user_id, "admin_quota_denied", self._audit_admin(user_id, "admin_quota_denied",
@@ -9848,6 +9947,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
resident_stats = _res_metrics(conn, member_ids, now) resident_stats = _res_metrics(conn, member_ids, now)
except Exception: # noqa: BLE001 - 观测端点不因 resident 失败 except Exception: # noqa: BLE001 - 观测端点不因 resident 失败
resident_stats = {"available": False} resident_stats = {"available": False}
# G7 运维指标:审批超时调阈(成员项目口径,只加键)。
# 统计失败时置 None(不是零),绝不把“算不出来”写成
# “超时率为零”;未扫荡的过期 pending 不计入,见
# approval_timeout 模块 docstring。这里不用 evaluate
#(它写日志,/metrics 高频轮询会刷屏),只算不记。
try:
_at_stats = _appr_timeout_stats(
conn, project_ids=sorted(member_ids))
approval_timeout = _appr_timeout_recommend(
_at_stats["expired_count"],
_at_stats["decided_count"])
approval_timeout.update(_at_stats)
except Exception:
approval_timeout = None
finally: finally:
conn.close() conn.close()
try: try:
@@ -9880,6 +9993,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
"queued": by_status.get("queued", 0), "queued": by_status.get("queued", 0),
"running": by_status.get("running", 0), "running": by_status.get("running", 0),
}, },
"approval_timeout": approval_timeout,
"failures_last_24h": failures_last_24h, "failures_last_24h": failures_last_24h,
"running_over_300s": running_over_300s, "running_over_300s": running_over_300s,
"workers": { "workers": {
+400
View File
@@ -0,0 +1,400 @@
"""成本与配额成套单测:双列报告、未定价单独列、按创建者分账、
超限硬切断、审批超时调阈三档。
不跑真实模型回合、不联网、不登录任何服务:成本行直接写
CostStore,配额判定走 quota_cutoff_reason 与 worker 的
_quota_exceeded(假 runner,不 dispatch),服务端点走本地
临时端口 + 临时 SQLite。
"""
import http.client
import json
import logging
import os
import sqlite3
import sys
import tempfile
import threading
import time
import unittest
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace
try:
from cloud.approval_timeout import (
DEFAULT_TIMEOUT_SECONDS, RAISED_TIMEOUT_SECONDS,
evaluate_approval_timeout, recommend_approval_timeout)
from cloud.approvals import (
create_approval, decide_approval, expire_overdue)
from cloud.auth import AuthStore
from cloud.cost import (
BILLING_NOTE, CostStore, price_usd_micros, quota_cutoff_reason)
from cloud.server import create_server, init_db
from cloud.worker import Worker
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.approval_timeout import (
DEFAULT_TIMEOUT_SECONDS, RAISED_TIMEOUT_SECONDS,
evaluate_approval_timeout, recommend_approval_timeout)
from cloud.approvals import (
create_approval, decide_approval, expire_overdue)
from cloud.auth import AuthStore
from cloud.cost import (
BILLING_NOTE, CostStore, price_usd_micros, quota_cutoff_reason)
from cloud.server import create_server, init_db
from cloud.worker import Worker
def _req(port, method, path, body=None, key=None):
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
hdrs = {}
payload = None
if body is not None:
payload = json.dumps(body).encode("utf-8")
hdrs["Content-Type"] = "application/json"
if key is not None:
hdrs["X-API-Key"] = key
try:
conn.request(method, path, body=payload, headers=hdrs)
resp = conn.getresponse()
data = resp.read()
finally:
conn.close()
try:
parsed = json.loads(data.decode("utf-8")) if data else None
except ValueError:
parsed = None
return resp.status, parsed
def _insert_task_row(db_path, task_id, created_by, project_id="p1"):
now = time.time()
conn = sqlite3.connect(db_path)
try:
conn.execute(
"CREATE TABLE IF NOT EXISTS tasks(id TEXT PRIMARY KEY,"
" project_id TEXT, prompt TEXT, status TEXT,"
" created_by TEXT, created_at REAL, updated_at REAL)")
conn.execute(
"INSERT INTO tasks(id, project_id, prompt, status,"
" created_by, created_at, updated_at)"
" VALUES(?, ?, 'work', 'completed', ?, ?, ?)",
(task_id, project_id, created_by, now, now))
conn.commit()
finally:
conn.close()
class DualColumnTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.store = CostStore(self.db_path)
def tearDown(self):
self.tmp.cleanup()
def _seed(self):
# 有价模型:gpt-6.1-sol 按价格表计价。
micros = price_usd_micros("gpt-6.1-sol", 10000, 0, 1000)
self.assertIsNotNone(micros)
self.store.record("t-priced", "p1", "gpt-6.1-sol",
input_tokens=10000, output_tokens=1000,
usd_micros=micros, source="actual")
# 未定价模型:usd_micros=None,绝不是 0 美元。
self.store.record("t-free", "p1", "gpt-5.6-terra",
input_tokens=300, output_tokens=200,
usd_micros=None, source="actual")
# 估算行不进未定价 actual 口径。
self.store.record("t-est", "p1", "gpt-5.6-terra",
input_tokens=999, output_tokens=0,
usd_micros=None, source="estimated")
return micros
def test_dual_column_and_unpriced_listed_separately(self):
micros = self._seed()
report = self.store.dual_column_report_by_project("p1")
self.assertEqual(report["priced_usd_micros"], micros)
self.assertEqual(report["usd_micros"], micros)
# 未定价 actual 只有 500(估算 999 不计),且单独列出。
self.assertEqual(report["unpriced_actual_tokens"], 500)
self.assertEqual(report["unpriced_models"], ["gpt-5.6-terra"])
self.assertIn("非中转实际账单", report["billing_note"])
self.assertEqual(report["billing_note"], BILLING_NOTE)
by_model = {row["model"]: row for row in report["by_model"]}
self.assertEqual(by_model["gpt-6.1-sol"]["priced_usd_micros"],
micros)
self.assertTrue(by_model["gpt-6.1-sol"]["priced"])
# 未定价行的美元是 None,不是 0:合计绝不把它按免费算。
self.assertIsNone(by_model["gpt-5.6-terra"]["priced_usd_micros"])
self.assertFalse(by_model["gpt-5.6-terra"]["priced"])
self.assertEqual(by_model["gpt-5.6-terra"]["unpriced_tokens"],
500)
def test_task_dual_column_keeps_old_keys(self):
micros = self._seed()
report = self.store.dual_column_report_by_task("t-priced")
self.assertEqual(report["total_tokens"], 11000)
self.assertEqual(report["usd_micros"], micros)
self.assertEqual(report["unpriced_actual_tokens"], 0)
self.assertEqual(report["unpriced_models"], [])
self.assertEqual(report["turns"], 1)
self.assertEqual(report["llm_requests"], 0)
def test_model_breakdown_requires_exactly_one_scope(self):
with self.assertRaises(ValueError):
self.store.model_breakdown()
with self.assertRaises(ValueError):
self.store.model_breakdown(task_id="a", project_id="p1")
with self.assertRaises(ValueError):
self.store.report_by_creator("u1", month="2026/10")
class CreatorSplitTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.store = CostStore(self.db_path)
def tearDown(self):
self.tmp.cleanup()
def test_report_by_creator_only_counts_own_tasks(self):
_insert_task_row(self.db_path, "t-a", "alice")
_insert_task_row(self.db_path, "t-b", "bob")
_insert_task_row(self.db_path, "t-orphan", None)
self.store.record("t-a", "p1", "gpt-6.1-sol",
input_tokens=100, output_tokens=10,
usd_micros=1000, source="actual")
self.store.record("t-b", "p1", "gpt-6.1-sol",
input_tokens=200, output_tokens=20,
usd_micros=2000, source="actual")
self.store.record("t-orphan", "p1", "gpt-6.1-sol",
input_tokens=999, output_tokens=0,
usd_micros=9999, source="actual")
alice = self.store.report_by_creator("alice")
self.assertEqual(alice["usd_micros"], 1000)
self.assertEqual(alice["total_tokens"], 110)
self.assertEqual(alice["creator_id"], "alice")
bob = self.store.dual_column_report_by_creator("bob")
self.assertEqual(bob["usd_micros"], 2000)
# 无属主任务的成本不归任何人,不会漏进 alice/bob。
self.assertEqual(self.store.report_by_creator("nobody")[
"usd_micros"], 0)
month = datetime.now(timezone.utc).strftime("%Y-%m")
self.assertEqual(
self.store.report_by_creator("alice", month=month)[
"usd_micros"], 1000)
self.assertEqual(
self.store.report_by_creator("alice", month="2000-01")[
"usd_micros"], 0)
class HardCutoffTestCase(unittest.TestCase):
def test_quota_cutoff_reason_three_cases(self):
# USD 超限。
self.assertIn("quota_exceeded",
quota_cutoff_reason(100000, 100000, 0, "2026-10"))
self.assertIn("quota_exceeded",
quota_cutoff_reason(100000, 150000, 0, "2026-10"))
# 无价上限超限(USD 额度还很大也不放行)。
self.assertIn("quota_exceeded",
quota_cutoff_reason(10 ** 12, 0, 500, "2026-10",
unpriced_cap=500))
# 未超限与显式不限额都不挡。
self.assertIsNone(quota_cutoff_reason(100000, 99999, 0,
"2026-10"))
self.assertIsNone(quota_cutoff_reason(None, 10 ** 9, 10 ** 9,
"2026-10"))
def test_worker_quota_exceeded_counts_in_flight_usage(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
db_path = os.path.join(tmp.name, "control.db")
init_db(db_path)
conn = sqlite3.connect(db_path)
try:
now = time.time()
conn.execute(
"INSERT INTO user_quotas(user_id, quota_usd_micros,"
" updated_at) VALUES(?, ?, ?)",
("tester", 100000, now))
conn.execute(
"INSERT INTO tasks(id, project_id, prompt, status,"
" idempotency_key, depends_on, priority, created_by,"
" created_at, updated_at, last_heartbeat)"
" VALUES(?, 'p1', 'work', 'queued', NULL, '[]', 0,"
" 'tester', ?, ?, NULL)",
("T1", now, now))
conn.execute(
"INSERT INTO tasks(id, project_id, prompt, status,"
" idempotency_key, depends_on, priority, created_by,"
" created_at, updated_at, last_heartbeat)"
" VALUES(?, 'p1', 'work', 'completed', NULL, '[]', 0,"
" 'tester', ?, ?, NULL)",
("donor", now, now))
conn.commit()
finally:
conn.close()
store = CostStore(db_path)
store.record("donor", "p1", "gpt-6.1-sol",
input_tokens=1, output_tokens=0,
usd_micros=90000, source="actual")
worker = Worker(db_path, "/fake/host.py",
runner=lambda cmd: SimpleNamespace(
returncode=0, stdout="{}", stderr=""),
env={}, model="gpt-6.1-sol",
retry_backoff_seconds=0)
task = {"id": "T1", "created_by": "tester",
"model": "gpt-6.1-sol"}
# 本轮已花 90000;再花 10000 input + 1000 output =
# 30000 微美元,合计 120000 >= 100000,轮次边界必须停。
err = worker._quota_exceeded(
task, {"input_tokens": 10000, "cached_input_tokens": 0,
"output_tokens": 1000, "total_tokens": 11000})
self.assertIsNotNone(err)
self.assertIn("quota_exceeded", err)
# 小用量不挡。
self.assertIsNone(worker._quota_exceeded(
task, {"input_tokens": 100, "cached_input_tokens": 0,
"output_tokens": 10, "total_tokens": 110}))
class ApprovalTimeoutTestCase(unittest.TestCase):
def test_recommend_three_tiers_and_boundaries(self):
# >20% 调 30 分钟。
rec = recommend_approval_timeout(3, 10)
self.assertEqual(rec["action"], "raise_to_30m")
self.assertEqual(rec["recommended_seconds"],
RAISED_TIMEOUT_SECONDS)
self.assertEqual(rec["recommended_seconds"], 1800)
# <5% 不动。
rec = recommend_approval_timeout(1, 100)
self.assertEqual(rec["action"], "keep")
self.assertEqual(rec["recommended_seconds"],
DEFAULT_TIMEOUT_SECONDS)
self.assertEqual(rec["recommended_seconds"], 600)
# 5%~20% 之间不动;边界值本身也不动。
self.assertEqual(recommend_approval_timeout(1, 10)["action"],
"keep")
self.assertEqual(recommend_approval_timeout(2, 10)["action"],
"keep")
self.assertEqual(recommend_approval_timeout(5, 100)["action"],
"keep")
# 无已终态时算不出来,不是 0%,同样不动。
rec = recommend_approval_timeout(0, 0)
self.assertIsNone(rec["timeout_rate"])
self.assertEqual(rec["action"], "keep")
def test_evaluate_counts_expired_and_logs(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
conn = sqlite3.connect(os.path.join(tmp.name, "appr.db"))
conn.row_factory = sqlite3.Row
try:
items = []
for i in range(10):
items.append(create_approval(
conn, "t1", "p1", summary="probe %d" % i,
timeout_seconds=60, now=1000.0))
for item in items[:7]:
decide_approval(conn, item["id"], "approved", "tester")
# 剩余 3 条过期:扫荡后 status=expired。
self.assertEqual(expire_overdue(conn, now=2000.0), 3)
with self.assertLogs("cloud.approval_timeout",
level="INFO") as logs:
result = evaluate_approval_timeout(conn)
self.assertEqual(result["decided_count"], 10)
self.assertEqual(result["expired_count"], 3)
self.assertAlmostEqual(result["timeout_rate"], 0.3)
self.assertEqual(result["action"], "raise_to_30m")
self.assertEqual(result["recommended_seconds"], 1800)
self.assertTrue(any("approval_timeout metric" in line
for line in logs.output))
finally:
conn.close()
class ServerCostEndpointsTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self.tmp.name, "control.db")
self.auth = AuthStore(self.db_path)
self.owner_id, self.owner_key = self.auth.create_user("owner")
self.other_id, self.other_key = self.auth.create_user("other")
self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id
self.server = create_server(self.db_path, "127.0.0.1", 0)
self.port = self.server.server_address[1]
self.thread = threading.Thread(
target=self.server.serve_forever, daemon=True)
self.thread.start()
s, proj = _req(self.port, "POST", "/projects",
{"name": "cproj"}, self.owner_key)
self.assertEqual(s, 201, proj)
self.project_id = proj["id"]
s, task = _req(self.port, "POST", "/tasks",
{"project_id": self.project_id,
"prompt": "cost probe"}, self.owner_key)
self.assertEqual(s, 201, task)
self.task_id = task["id"]
store = CostStore(self.db_path)
store.record(self.task_id, self.project_id, "gpt-6.1-sol",
input_tokens=10000, output_tokens=1000,
usd_micros=30000, source="actual")
store.record(self.task_id, self.project_id, "gpt-5.6-terra",
input_tokens=300, output_tokens=200,
usd_micros=None, source="actual")
try:
store._conn.close()
except Exception:
pass
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.tmp.cleanup()
if self._old_admins is None:
os.environ.pop("CODEX_GLOBAL_ADMINS", None)
else:
os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
def test_user_cost_and_task_cost_dual_column(self):
s, body = _req(self.port, "GET", "/users/me/cost",
key=self.owner_key)
self.assertEqual(s, 200, body)
self.assertEqual(body["cost"]["priced_usd_micros"], 30000)
self.assertEqual(body["cost"]["unpriced_actual_tokens"], 500)
self.assertEqual(body["cost"]["unpriced_models"],
["gpt-5.6-terra"])
self.assertIn("非中转实际账单",
body["cost"]["billing_note"])
s, body = _req(self.port, "GET",
"/tasks/%s/cost" % self.task_id,
key=self.owner_key)
self.assertEqual(s, 200, body)
self.assertEqual(body["priced_usd_micros"], 30000)
self.assertEqual(body["unpriced_models"], ["gpt-5.6-terra"])
s, body = _req(self.port, "GET",
"/users/me/cost?month=2026/10",
key=self.owner_key)
self.assertEqual(s, 400, body)
def test_admin_approval_timeout_ops_gated(self):
s, body = _req(self.port, "GET",
"/admin/ops/approval-timeout",
key=self.other_key)
self.assertEqual(s, 403, body)
s, body = _req(self.port, "GET",
"/admin/ops/approval-timeout",
key=self.owner_key)
self.assertEqual(s, 200, body)
self.assertEqual(body["current_seconds"], 600)
self.assertEqual(body["decided_count"], 0)
self.assertIsNone(body["timeout_rate"])
self.assertEqual(body["action"], "keep")
if __name__ == "__main__":
unittest.main()
+6 -10
View File
@@ -161,7 +161,7 @@ from cloud.coordinator import parse_parts as _coord_parse_parts
from cloud.coordinator import parts_by_name as _coord_parts_by_name from cloud.coordinator import parts_by_name as _coord_parts_by_name
from cloud.coordinator import nonterminal_children as _coord_nonterminal from cloud.coordinator import nonterminal_children as _coord_nonterminal
from cloud.cost import CostStore, estimate_tokens, price_usd_micros, \ from cloud.cost import CostStore, estimate_tokens, price_usd_micros, \
sum_usage, unpriced_token_cap quota_cutoff_reason, sum_usage
from cloud.db import connect as _db_connect from cloud.db import connect as _db_connect
from cloud.goal_command import parse_goal_command as _parse_goal_command from cloud.goal_command import parse_goal_command as _parse_goal_command
from cloud.logging_config import get_logger, log_event, setup_logging from cloud.logging_config import get_logger, log_event, setup_logging
@@ -3052,15 +3052,11 @@ class Worker:
unpriced_used += ( unpriced_used += (
max(0, int(usage_total.get("input_tokens", 0))) max(0, int(usage_total.get("input_tokens", 0)))
+ max(0, int(usage_total.get("output_tokens", 0)))) + max(0, int(usage_total.get("output_tokens", 0))))
if used >= quota: # 硬切断判定收敛到 cost.quota_cutoff_reason:USD 配额与
return ("quota_exceeded: 本月配额已用完(已用 %d / " # 无价上限任一超限即回 quota_exceeded 文本,调用方(认领
"配额 %d 微美元,%s)" % (used, quota, month)) # 门、轮次边界、30 秒轮询)据此停掉任务、不再开新回合;
cap = unpriced_token_cap() # 文本与旧分支逐字一致,旧单测按此前缀断言。
if unpriced_used >= cap: return quota_cutoff_reason(quota, used, unpriced_used, month)
return ("quota_exceeded: 本月无价模型 token 已用完"
"(已用 %d / 上限 %d,%s)"
% (unpriced_used, cap, month))
return None
def _ready_tasks(self): def _ready_tasks(self):
"""当前就绪任务列表(pick_next 的批量版)。 """当前就绪任务列表(pick_next 的批量版)。