From 143ab3009466c0c60870abe5aafe1be99ac35eef Mon Sep 17 00:00:00 2001 From: host Date: Fri, 30 Oct 2026 16:00:51 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20cost/quota=20set=20=E2=80=94=20dual-col?= =?UTF-8?q?umn=20priced=20reports,=20creator=20split,=20hard=20cutoff,=20a?= =?UTF-8?q?pproval-timeout=20auto-threshold?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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. --- cloud/approval_timeout.py | 145 ++++++++++++ cloud/cost.py | 204 +++++++++++++++++ cloud/server.py | 118 +++++++++- cloud/tests/test_cost_quota.py | 400 +++++++++++++++++++++++++++++++++ cloud/worker.py | 16 +- 5 files changed, 871 insertions(+), 12 deletions(-) create mode 100644 cloud/approval_timeout.py create mode 100644 cloud/tests/test_cost_quota.py diff --git a/cloud/approval_timeout.py b/cloud/approval_timeout.py new file mode 100644 index 0000000..bff0e29 --- /dev/null +++ b/cloud/approval_timeout.py @@ -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 diff --git a/cloud/cost.py b/cloud/cost.py index 43b6dc5..5011e40 100644 --- a/cloud/cost.py +++ b/cloud/cost.py @@ -135,6 +135,39 @@ def price_usd_micros(model, input_tokens, cached_input_tokens, 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(): """Monthly unpriced-token ceiling for users who have a quota row. @@ -397,3 +430,174 @@ class CostStore: def report_by_day(self, day) -> dict: """Aggregate cost records for one UTC day (YYYY-MM-DD).""" 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)) diff --git a/cloud/server.py b/cloud/server.py index 0f6a256..ed3ec10 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -132,6 +132,9 @@ from cloud.todos import get_todo as _todo_get from cloud.todos import list_todos as _todo_list from cloud.todos import update_todo as _todo_update 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 decide_approval as _appr_decide from cloud.approvals import ensure_approvals_table as _appr_ensure @@ -3020,6 +3023,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(404, {"error": "not found"}) elif path == "/users/me/quota" and method == "GET": 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": self._identity_profile_get(user_id) elif path == "/users/me/profile" and method in ("PUT", "POST"): @@ -4637,7 +4646,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): def _project_cost(self, project_id, query): store = CostStore(self.server.db_path) 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.row_factory = sqlite3.Row days = (query.get("days") or [None])[0] @@ -4712,7 +4723,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): return store = CostStore(self.server.db_path) try: - report = store.report_by_task(task_id) + # 双列：与项目成本同款，旧键全保留，只加双列键。 + report = store.dual_column_report_by_task(task_id) finally: try: store._conn.close() @@ -8875,6 +8887,93 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): payload["unpriced_token_cap"] = unpriced_token_cap() 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): if not self._is_global_admin(user_id): self._audit_admin(user_id, "admin_quota_denied", @@ -9361,6 +9460,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): resident_stats = _res_metrics(conn, member_ids, now) except Exception: # noqa: BLE001 - 观测端点不因 resident 失败 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: conn.close() try: @@ -9393,6 +9506,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "queued": by_status.get("queued", 0), "running": by_status.get("running", 0), }, + "approval_timeout": approval_timeout, "failures_last_24h": failures_last_24h, "running_over_300s": running_over_300s, "workers": { diff --git a/cloud/tests/test_cost_quota.py b/cloud/tests/test_cost_quota.py new file mode 100644 index 0000000..bf693ff --- /dev/null +++ b/cloud/tests/test_cost_quota.py @@ -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() diff --git a/cloud/worker.py b/cloud/worker.py index c73e647..cb18809 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -142,7 +142,7 @@ from cloud.coordinator import ensure_columns as _coord_ensure_columns from cloud.coordinator import parse_parts as _coord_parse_parts from cloud.coordinator import nonterminal_children as _coord_nonterminal 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.goal_command import parse_goal_command as _parse_goal_command from cloud.logging_config import get_logger, log_event, setup_logging @@ -2155,15 +2155,11 @@ class Worker: unpriced_used += ( max(0, int(usage_total.get("input_tokens", 0))) + max(0, int(usage_total.get("output_tokens", 0)))) - if used >= quota: - return ("quota_exceeded: 本月配额已用完（已用 %d / " - "配额 %d 微美元，%s）" % (used, quota, month)) - cap = unpriced_token_cap() - if unpriced_used >= cap: - return ("quota_exceeded: 本月无价模型 token 已用完" - "（已用 %d / 上限 %d，%s）" - % (unpriced_used, cap, month)) - return None + # 硬切断判定收敛到 cost.quota_cutoff_reason：USD 配额与 + # 无价上限任一超限即回 quota_exceeded 文本，调用方（认领 + # 门、轮次边界、30 秒轮询）据此停掉任务、不再开新回合； + # 文本与旧分支逐字一致，旧单测按此前缀断言。 + return quota_cutoff_reason(quota, used, unpriced_used, month) def _ready_tasks(self): """当前就绪任务列表（pick_next 的批量版）。