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.

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

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))

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": {

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()

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:
+        # 硬切断判定收敛到 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):
         """当前就绪任务列表（pick_next 的批量版）。