Merge pull request #71 from openqj/feat-quotas

租户配额与计费：按用户月度 USD 上限 + 建任务/分支/重试准入 402

cloud/README.md

@@ -871,6 +871,47 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   `/tasks`、`/projects`、`/workers`、`/health` 同款，否则页面能打开
   但环境区永远加载失败）。
 
+## 租户配额与计费
+
+- 口径：配额是按用户、按 UTC 自然月的 USD 上限（`user_quotas`
+  表，`quota_usd_micros` 为 NULL 表示不限额，0 表示一分钱也不
+  能花；金额单位与 cost_records 一致，都是微美元，避免浮点）。
+  用量只算 `usd_micros` 非空的实际成本行（estimated 行 usd 为
+  NULL，SUM 自动跳过），按任务创建者 `created_by` 归属——一个
+  用户在多个项目里花的钱都算进他自己的月度账单；项目维度的
+  拆分由项目成本接口另行提供。
+- 准入：建任务、会话分支、重试在权限校验通过后、写库之前查
+  配额，本月用量 >= 配额时回 402 `quota_exceeded`（附 month、
+  quota_usd_micros、used_usd_micros），不建任务、不改状态。
+  老用户与未设配额的用户默认不限额，行为与过去一致。注意
+  边界：只做准入校验，不做跑中掐断——已在跑的回合、后续消
+  息触发的新一轮中途超限不会被停掉（worker 侧没有按金额中途
+  停转的钩子）；重试按操作人查配额、费用仍记到原任务创建者
+  名下（建任务与分支的操作人就是创建者，没有这个错位）。
+  系统任务（GitHub webhook 的 /fix-review，created_by
+  是 'system'）不记到任何用户名下、不占配额；要把它算到项
+  目所有者头上需另做归属。
+- 接口：`GET /users/me/quota`（自查，返回 month、quota、
+  used、remaining，不限额时 quota 与 remaining 为 null）；
+  `GET /admin/quotas?user_id=` 与 `POST /admin/quotas`
+ （body：user_id、quota_usd_micros，null=不限额，负数 400，
+  用户不存在 404，改完记 `quota_set` 审计）。管理口径沿用
+  `/admin/backup` 的做法：在任一项目里是 maintainer/owner 即
+  算管理员，没有任何项目角色的用户无权改任何人的配额。
+  这还不是真正的收款：没有接支付网关、不开发票、不扣款，
+  只有记账 + 配额 + 超限拦截；要收钱需另接支付（价格表在
+  `cloud/cost.py`，注意那是 OpenAI 牌价、不是中转站的账单）。
+- 命令台：API Key 卡下方显示本月用量与配额（10 秒自动刷新；
+  超限建任务时接口的 402 会带“本月配额已用完”提示）。管理
+  员改配额目前只走 API，控制台没有改配额的表单。
+- 公网 nginx：注意，443 块的默认 `location /` 是代理到主机
+  `127.0.0.1:28080`（不是控制面），而 `/users`、`/admin`
+  目前没有专属 location——配额接口部署上生产时须同批加
+  `location ^~ /users/me/quota` 与 `location ^~ /admin/quotas`
+  （只放这两个精确前缀，不要整段放行 `/users` 或 `/admin`，
+  否则会把 `/admin/sweep`、`/admin/backup` 也暴露到公网），
+  否则公网页面调配额接口会落到 28080 而不是控制面。
+
 ## 运维批（本轮新增：部署、恢复演练、监控、心跳、多用户验证）
 
 细则在 `deploy/README.md`，这里只记代码侧的接口与默认值：

cloud/console.html

@@ -87,6 +87,7 @@
   <button id="clearKey">清除</button>
   <span id="keyState" class="muted"></span>
 </div>
+<div><span id="quotaState" class="muted">未加载</span></div>
 </section>
 
 <section class="card">
@@ -292,6 +293,7 @@ async function api(path, method, body) {
     var msg = (data && data.error) ? data.error : ("HTTP " + r.status);
     if (r.status === 401) msg += "（key 未填或无效）";
     if (r.status === 429) msg += "（触发限流，稍后再试）";
+    if (r.status === 402) msg += "（本月配额已用完，联系管理员调整）";
     throw new Error(msg);
   }
   return data;
@@ -324,6 +326,23 @@ async function loadHealth() {
   } catch (e) { $("#health").textContent = "加载失败：" + e.message; }
 }
 
+async function loadQuota() {
+  // 本月配额与用量：只读展示，管理员改配额走 API（/admin/quotas）。
+  var el = $("#quotaState");
+  if (!getKey()) { el.textContent = "未填 key，配额不显示"; return; }
+  try {
+    var d = await api("/users/me/quota", "GET");
+    var used = (d.used_usd_micros / 1000000).toFixed(2);
+    if (d.quota_usd_micros === null || d.quota_usd_micros === undefined) {
+      el.textContent = "本月用量 $" + used + "（不限额）";
+    } else {
+      var q = (d.quota_usd_micros / 1000000).toFixed(2);
+      var rem = (d.remaining_usd_micros / 1000000).toFixed(2);
+      el.textContent = "本月用量 $" + used + " / $" + q + "（剩余 $" + rem + "）";
+    }
+  } catch (e) { el.textContent = "配额加载失败：" + e.message; }
+}
+
 async function loadProjects() {
   try {
     var d = await api("/projects", "GET");
@@ -610,6 +629,7 @@ $("#saveKey").onclick = function () {
   loadWorkers();
   if ($("#projectSelect").value) { loadTasks(); loadEnvironments(); }
   if (getKey()) loadRelayModels();
+  loadQuota();
 };
 $("#clearKey").onclick = function () { localStorage.removeItem(KEY_STORE); setKeyState(); };
 $("#refreshHealth").onclick = loadHealth;
@@ -789,11 +809,12 @@ setKeyState();
 loadHealth();
 loadProjects();
 loadWorkers();
+loadQuota();
 // 回访用户 key 已在 localStorage 里：初始化也要加载模型列表，
 // 否则下拉永远只有默认项（此前只在点“保存”时调一次，
 // 2026-10-27 实测：存过 key 后重开页面模型选不了）。
 if (getKey()) loadRelayModels();
-setInterval(function () { if ($("#projectSelect").value) { loadTasks(); loadEnvironments(); } loadHealth(); }, 10000);
+setInterval(function () { if ($("#projectSelect").value) { loadTasks(); loadEnvironments(); } loadHealth(); loadQuota(); }, 10000);
 </script>
 </body>
 </html>

cloud/db.py

@@ -885,6 +885,10 @@ POSTGRES_SCHEMA = (
     "CREATE UNIQUE INDEX IF NOT EXISTS idx_env_name"
     " ON environments(project_id, name)"
     " WHERE status != 'archived'",
+    "CREATE TABLE IF NOT EXISTS user_quotas("
+    "user_id TEXT PRIMARY KEY, "
+    "quota_usd_micros BIGINT, "
+    "updated_at DOUBLE PRECISION)",
     "CREATE TABLE IF NOT EXISTS summaries("
     "id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, "
     "project_id TEXT, "

cloud/server.py

@@ -623,6 +623,19 @@ def _ensure_environments_table(conn):
         " WHERE status != 'archived'")
 
 
+def _ensure_quotas_table(conn):
+    """幂等建配额表：server init_db 与配额查询都可调它，老库
+    第一次跑到这里时自动补表，不丢数据。quota_usd_micros 为
+    NULL 表示不限额（与 0 区分：0 是一分钱也不能花）；金额
+    单位与 cost_records 一致，都是 USD 微单位（1 美元 =
+    1_000_000），避免浮点。"""
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS user_quotas("
+        "user_id TEXT PRIMARY KEY, "
+        "quota_usd_micros BIGINT, "
+        "updated_at DOUBLE PRECISION)")
+
+
 _ENV_BRANCH_RE = re.compile(r"^[A-Za-z0-9_.\-/]+$")
 
 
@@ -683,6 +696,7 @@ def init_db(db_path):
             migrate_projects_table(conn)
             _ensure_followups_table(conn)
             _ensure_environments_table(conn)
+            _ensure_quotas_table(conn)
             _ensure_uploads_tables(conn)
             conn.commit()
             return
@@ -714,6 +728,7 @@ def init_db(db_path):
         migrate_projects_table(conn)
         _ensure_followups_table(conn)
         _ensure_environments_table(conn)
+        _ensure_quotas_table(conn)
         _ensure_uploads_tables(conn)
         # worker 每轮 poll 都按 status 取任务（见 load_tasks 的
         # WHERE status IN (...)），无索引时是全表扫描；任务表随
@@ -1098,6 +1113,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._archive_environment(user_id, rest[: -len("/archive")])
             else:
                 self._send_json(404, {"error": "not found"})
+        elif path == "/users/me/quota" and method == "GET":
+            self._user_quota_self(user_id)
         elif path == "/tasks" and method == "POST":
             self._create_task(user_id)
         elif path == "/uploads" and method == "POST":
@@ -1108,6 +1125,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._admin_sweep()
         elif path == "/admin/backup" and method == "POST":
             self._admin_backup(user_id)
+        elif path == "/admin/quotas" and method == "GET":
+            self._admin_quota_get(user_id, query)
+        elif path == "/admin/quotas" and method == "POST":
+            self._admin_quota_set(user_id)
         elif path == "/admin/push/dispatch" and method == "POST":
             self._push_dispatch_endpoint(user_id)
         elif path.startswith("/tasks/"):
@@ -2017,6 +2038,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             if not store.check_permission(user_id, project_id, "create_task"):
                 self._send_json(403, {"error": "forbidden"})
                 return
+        if self._reject_if_over_quota(user_id):
+            return
         # 环境绑定：environment_id 非空时必须是本项目、已 ready
         # 的环境；creating/failed/archived 一律拒绝并把状态写进
         # 报错（creating 时用户就知道该等 worker 准备好再建任务，
@@ -2466,6 +2489,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     user_id, task["project_id"], "create_task"):
                 self._send_json(403, {"error": "forbidden"})
                 return
+        if self._reject_if_over_quota(user_id):
+            return
         message = body.get("message")
         if not message or not isinstance(message, str) \
                 or not message.strip():
@@ -2800,6 +2825,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     user_id, task["project_id"], "approve_merge"):
                 self._send_json(403, {"error": "forbidden"})
                 return
+        if self._reject_if_over_quota(user_id):
+            return
         if task["status"] not in ("failed", "interrupted", "merge-failed"):
             self._send_json(409, {
                 "error": "only failed/interrupted tasks can be retried",
@@ -2834,6 +2861,169 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             store.audit(user_id, task["project_id"], "task_retried", task_id)
         self._send_json(200, task)
 
+    # -- 租户配额与计费 ----------------------------------------------
+    # 计费口径：只算 cost_records 里 usd_micros 非空的实际成本
+    # 行（estimated 行 usd 为 NULL，SUM 自动跳过，不算已花
+    # 的钱）；归属按任务创建者 created_by，不按项目——一个
+    # 用户在多个项目里花的钱都算进他自己的月度账单，项目维
+    # 度的拆分由 _project_cost 另行提供。窗口是 UTC 自然月，
+    # 与 cost_records.day（UTC 日期）同口径。
+    def _quota_month(self):
+        from datetime import datetime, timezone
+        return datetime.now(timezone.utc).strftime("%Y-%m")
+
+    def _quota_status(self, user_id):
+        """返回 (quota_usd_micros|None, used_usd_micros, month)。
+        quota 为 None 表示不限额：老用户与未设配额的用户默认
+        不限额，行为与过去一致，不会因为上了配额功能就突然
+        被挡住。"""
+        month = self._quota_month()
+        conn = self._db()
+        try:
+            _ensure_quotas_table(conn)
+            conn.commit()
+            # cost_records 由 CostStore 懒建：全新库上它还不存在，
+            # 直接 JOIN 会报 no such table——先实例化一次建表
+            # （与 _project_cost 同款做法），再跑配额查询。
+            _cost_store = CostStore(self.server.db_path)
+            try:
+                _cost_store._conn.close()
+            except Exception:
+                pass
+            row = conn.execute(
+                "SELECT quota_usd_micros FROM user_quotas"
+                " WHERE user_id = ?",
+                (user_id,)).fetchone()
+            quota = row[0] if row is not None else None
+            used = conn.execute(
+                "SELECT COALESCE(SUM(c.usd_micros), 0)"
+                " FROM cost_records c JOIN tasks t ON c.task_id = t.id"
+                " WHERE t.created_by = ? AND c.day LIKE ?",
+                (user_id, month + "%")).fetchone()[0]
+            return quota, int(used or 0), month
+        finally:
+            conn.close()
+
+    def _quota_payload(self, user_id, quota, used, month):
+        return {"user_id": user_id, "month": month,
+                "quota_usd_micros": quota, "used_usd_micros": used,
+                "remaining_usd_micros": (quota - used)
+                if quota is not None else None}
+
+    def _check_quota(self, user_id):
+        """配额准入：放行返回 None；超限返回 (quota, used,
+        month)，调用方一律回 402——这是钱用完了，不是请求
+        太快，所以不用 429。注意：只做建任务/分支/重跑时的
+        准入校验；跑起来的回合中途超限不会被掐断（worker
+        侧没有按金额中途停转的钩子，那是下一步的事）。"""
+        quota, used, month = self._quota_status(user_id)
+        if quota is not None and used >= quota:
+            return quota, used, month
+        return None
+
+    def _reject_if_over_quota(self, user_id):
+        blocked = self._check_quota(user_id)
+        if blocked is not None:
+            quota, used, month = blocked
+            self._send_json(402, {"error": "quota_exceeded",
+                                  "month": month,
+                                  "quota_usd_micros": quota,
+                                  "used_usd_micros": used})
+            return True
+        return False
+
+    def _is_global_maintainer(self, user_id):
+        """全局管理判定：沿用 _admin_backup 的口径——在任一
+        项目里是 maintainer/owner 即算（没有单设的全局管理
+        员表；没有任何项目角色的用户无权改任何人的配额）。"""
+        conn = self._db()
+        try:
+            row = conn.execute(
+                "SELECT COUNT(*) FROM project_members"
+                " WHERE user_id = ? AND role IN ('maintainer', 'owner')",
+                (user_id,)).fetchone()
+            return (row[0] if row is not None else 0) > 0
+        finally:
+            conn.close()
+
+    def _user_exists(self, target):
+        conn = self._db()
+        try:
+            return conn.execute(
+                "SELECT 1 FROM users WHERE id = ?",
+                (target,)).fetchone() is not None
+        finally:
+            conn.close()
+
+    def _user_quota_self(self, user_id):
+        quota, used, month = self._quota_status(user_id)
+        self._send_json(200,
+                        self._quota_payload(user_id, quota, used, month))
+
+    def _admin_quota_get(self, user_id, query):
+        if not self._is_global_maintainer(user_id):
+            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
+        quota, used, month = self._quota_status(target)
+        self._send_json(200,
+                        self._quota_payload(target, quota, used, month))
+
+    def _admin_quota_set(self, user_id):
+        if not self._is_global_maintainer(user_id):
+            self._send_json(403, {"error": "forbidden"})
+            return
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        target = body.get("user_id")
+        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
+        quota = body.get("quota_usd_micros")
+        if quota is not None:
+            try:
+                quota = int(quota)
+            except (TypeError, ValueError):
+                self._send_json(400, {
+                    "error": "quota_usd_micros must be an integer or null"})
+                return
+            if quota < 0:
+                self._send_json(400, {
+                    "error": "quota_usd_micros must be >= 0"})
+                return
+        now = time.time()
+        conn = self._db()
+        try:
+            cur = conn.execute(
+                "UPDATE user_quotas SET quota_usd_micros = ?,"
+                " updated_at = ? WHERE user_id = ?",
+                (quota, now, target))
+            if cur.rowcount != 1:
+                conn.execute(
+                    "INSERT INTO user_quotas(user_id, quota_usd_micros,"
+                    " updated_at) VALUES(?, ?, ?)",
+                    (target, quota, now))
+            conn.commit()
+        finally:
+            conn.close()
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, None, "quota_set",
+                        "%s:%s" % (target, quota))
+        quota, used, month = self._quota_status(target)
+        self._send_json(200,
+                        self._quota_payload(target, quota, used, month))
+
     # -- worker 池 -----------------------------------------------------
     def _workers_list(self, user_id):
         """列出注册 worker：任何已认证用户可读（worker 是基础设施

cloud/tests/test_task_quotas.py

@@ -0,0 +1,239 @@
+"""租户配额与计费单测：月度 USD 上限 + 准入 402 + 管理端点。
+
+不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时
+SQLite；用量直接插 cost_records 行（按任务 created_by 归属、
+UTC day），验证 JOIN 归属、自然月窗口与三路准入（建任务 /
+分支 / 重试）。配额单位是微美元（1 美元 = 1_000_000），
+与 cost_records 一致。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+
+
+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 _set_task_status(db_path, task_id, status):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
+                     (status, task_id))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _insert_cost(db_path, task_id, project_id, usd_micros, day):
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO cost_records(task_id, project_id, model,"
+            " input_tokens, output_tokens, total_tokens, usd_micros,"
+            " source, day, created_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (task_id, project_id, "gpt-5.6-terra", 100, 50, 150,
+             usd_micros, "actual", day, 0))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _utc_today():
+    return datetime.now(timezone.utc).strftime("%Y-%m-%d")
+
+
+def _prev_month_day():
+    today = date.today()
+    first = today.replace(day=1)
+    return (first - timedelta(days=1)).isoformat()
+
+
+class ServerQuotasTestCase(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.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": "qproj"}, self.owner_key)
+        self.assertEqual(s, 201, proj)
+        self.project_id = proj["id"]
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.tmp.cleanup()
+
+    def _create_task(self, key=None):
+        return _req(self.port, "POST", "/tasks",
+                    {"project_id": self.project_id,
+                     "prompt": "quota probe"},
+                    key if key is not None else self.owner_key)
+
+    def _set_quota(self, target, quota, key=None):
+        return _req(self.port, "POST", "/admin/quotas",
+                    {"user_id": target, "quota_usd_micros": quota},
+                    key if key is not None else self.owner_key)
+
+    def test_self_quota_default_unlimited(self):
+        s, q = _req(self.port, "GET", "/users/me/quota",
+                    key=self.owner_key)
+        self.assertEqual(s, 200, q)
+        self.assertIsNone(q["quota_usd_micros"])
+        self.assertEqual(q["used_usd_micros"], 0)
+        self.assertIsNone(q["remaining_usd_micros"])
+        self.assertEqual(q["month"],
+                         datetime.now(timezone.utc).strftime("%Y-%m"))
+
+    def test_admin_set_and_get(self):
+        s, q = self._set_quota(self.owner_id, 5000000)
+        self.assertEqual(s, 200, q)
+        self.assertEqual(q["quota_usd_micros"], 5000000)
+        s, q = _req(self.port, "GET",
+                    "/admin/quotas?user_id=" + self.owner_id,
+                    key=self.owner_key)
+        self.assertEqual(s, 200, q)
+        self.assertEqual(q["quota_usd_micros"], 5000000)
+        s, q = _req(self.port, "GET", "/users/me/quota",
+                    key=self.owner_key)
+        self.assertEqual(q["quota_usd_micros"], 5000000)
+        # null = 恢复不限额
+        s, q = self._set_quota(self.owner_id, None)
+        self.assertEqual(s, 200, q)
+        self.assertIsNone(q["quota_usd_micros"])
+
+    def test_admin_set_forbidden_and_validation(self):
+        # other 没有任何项目角色，不是管理员
+        s, q = self._set_quota(self.owner_id, 100, self.other_key)
+        self.assertEqual(s, 403, q)
+        s, q = _req(self.port, "GET",
+                    "/admin/quotas?user_id=" + self.owner_id,
+                    key=self.other_key)
+        self.assertEqual(s, 403, q)
+        s, q = self._set_quota(self.owner_id, -1)
+        self.assertEqual(s, 400, q)
+        s, q = self._set_quota("no-such-user", 100)
+        self.assertEqual(s, 404, q)
+        s, q = _req(self.port, "POST", "/admin/quotas",
+                    {"quota_usd_micros": 100}, self.owner_key)
+        self.assertEqual(s, 400, q)
+        s, q = _req(self.port, "GET", "/admin/quotas",
+                    key=self.owner_key)
+        self.assertEqual(s, 400, q)
+
+    def test_usage_counts_current_month_only(self):
+        s, task = self._create_task()
+        self.assertEqual(s, 201, task)
+        _insert_cost(self.db_path, task["id"], self.project_id,
+                     5000000, _utc_today())
+        _insert_cost(self.db_path, task["id"], self.project_id,
+                     100000000, _prev_month_day())
+        s, q = _req(self.port, "GET", "/users/me/quota",
+                    key=self.owner_key)
+        self.assertEqual(s, 200, q)
+        self.assertEqual(q["used_usd_micros"], 5000000)
+
+    def test_create_blocked_at_quota(self):
+        s, task = self._create_task()
+        self.assertEqual(s, 201, task)
+        _insert_cost(self.db_path, task["id"], self.project_id,
+                     5000000, _utc_today())
+        s, _ = self._set_quota(self.owner_id, 5000000)
+        self.assertEqual(s, 200)
+        s, body = self._create_task()
+        self.assertEqual(s, 402, body)
+        self.assertEqual(body["error"], "quota_exceeded")
+        self.assertEqual(body["quota_usd_micros"], 5000000)
+        self.assertEqual(body["used_usd_micros"], 5000000)
+        # 额度调高后放行
+        s, _ = self._set_quota(self.owner_id, 6000000)
+        self.assertEqual(s, 200)
+        s, task2 = self._create_task()
+        self.assertEqual(s, 201, task2)
+
+    def test_fork_blocked_at_quota(self):
+        s, task = self._create_task()
+        self.assertEqual(s, 201, task)
+        _insert_cost(self.db_path, task["id"], self.project_id,
+                     3000000, _utc_today())
+        _set_task_status(self.db_path, task["id"], "completed")
+        s, _ = self._set_quota(self.owner_id, 3000000)
+        self.assertEqual(s, 200)
+        s, body = _req(self.port, "POST",
+                       "/tasks/%s/fork" % task["id"],
+                       {"message": "branch probe"}, self.owner_key)
+        self.assertEqual(s, 402, body)
+        s, _ = self._set_quota(self.owner_id, None)
+        self.assertEqual(s, 200)
+        s, forked = _req(self.port, "POST",
+                         "/tasks/%s/fork" % task["id"],
+                         {"message": "branch probe"}, self.owner_key)
+        self.assertEqual(s, 201, forked)
+
+    def test_retry_blocked_at_quota(self):
+        s, task = self._create_task()
+        self.assertEqual(s, 201, task)
+        _insert_cost(self.db_path, task["id"], self.project_id,
+                     2000000, _utc_today())
+        _set_task_status(self.db_path, task["id"], "failed")
+        s, _ = self._set_quota(self.owner_id, 2000000)
+        self.assertEqual(s, 200)
+        s, body = _req(self.port, "POST",
+                       "/tasks/%s/retry" % task["id"], {},
+                       self.owner_key)
+        self.assertEqual(s, 402, body)
+        s, row = _req(self.port, "GET", "/tasks/" + task["id"],
+                      key=self.owner_key)
+        self.assertEqual(row["status"], "failed")
+        s, _ = self._set_quota(self.owner_id, None)
+        self.assertEqual(s, 200)
+        s, retried = _req(self.port, "POST",
+                          "/tasks/%s/retry" % task["id"], {},
+                          self.owner_key)
+        self.assertEqual(s, 200, retried)
+        self.assertEqual(retried["status"], "queued")
+
+
+if __name__ == "__main__":
+    unittest.main()