Production-grade: webhook execution, Postgres backup, rate limit, worker timeout/retry

GitHub webhook: delivery-id dedup persisted in webhook_deliveries,
/fix-review creates a queued task with audit (no model run in the
webhook itself), check_run/check_suite parsing without mutating
task state (CI backfill still via pr-checks).
Control plane: Postgres pg_dump backup branch + POST /admin/backup
(maintainer+, password only via PGPASSWORD env), per-key/IP rate
limit (default 120/min, 429), health returns backend + db_ok,
metrics by_status always has all seven status keys.
Worker: per-task timeout (default 1800s, --task-timeout), timeout
is failed not interrupted, limited retries (default 1, backoff,
fallback failures not re-retried), stale codex- container cleanup
on loop/CLI start.
Tests: cloud 271 (267 pass, 4 skipped are pre-existing real-
Postgres tests), host 27 pass.

cloud/server.py

@@ -37,12 +37,15 @@ audit_log 三张表；本 server 用 projects/tasks 两张表，表名不冲突
 """
 
 import argparse
+import hashlib
 import html
 import json
 import os
 import sqlite3
+import threading
 import time
 import uuid
+from collections import deque
 from contextlib import contextmanager
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import parse_qs, urlparse
@@ -71,6 +74,8 @@ from cloud.push import (
     list_subscriptions as _push_list,
 )
 from cloud.webhooks import (
+    WebhookDeliveryStore,
+    parse_github_check_run,
     parse_github_pr_comment,
     parse_slack_command,
     verify_github_signature,
@@ -81,6 +86,73 @@ TERMINAL_STATUSES = frozenset(
     {"completed", "failed", "interrupted", "merged", "merge-failed"}
 )
 DEFAULT_SWEEP_THRESHOLD_SECONDS = 300
+# 限流默认值：每 IP / 每 API key 每分钟 120 次；环境变量
+# CLOUD_RATE_LIMIT_PER_MINUTE / CLOUD_RATE_LIMIT_WINDOW_SECONDS 可调。
+# 限流是内存滑动窗口，不跨进程共享（多 worker 部署时每进程独立计数，
+# 这是诚实边界，不是全局限流）。
+DEFAULT_RATE_LIMIT_PER_MINUTE = 120
+DEFAULT_RATE_LIMIT_WINDOW_SECONDS = 60
+TASK_STATUS_KEYS = (
+    "queued", "running", "completed", "failed",
+    "interrupted", "merged", "merge-failed",
+)
+
+
+def _rate_limit_from_env(name, default):
+    try:
+        value = int(os.environ.get(name) or default)
+    except (TypeError, ValueError):
+        return default
+    return value if value > 0 else default
+
+
+class RateLimiter:
+    """内存滑动窗口限流：每 key 每窗口最多 per_minute 次。
+
+    只在 allow() 内部持一把短锁做 deque 修剪与计数，不持锁做
+    任何 I/O，不会把正常请求串行化。key 由调用方给（本 server
+    传“key:<API key 的 sha256>”或“ip:<客户端 IP>”，原始 API key
+    绝不存进限流结构里）。窗口与阈值可在实例上改（测试注入小
+    阈值验证 429 就靠改这两个属性）。
+    """
+
+    def __init__(self, per_minute=DEFAULT_RATE_LIMIT_PER_MINUTE,
+                 window_seconds=DEFAULT_RATE_LIMIT_WINDOW_SECONDS,
+                 time_fn=None):
+        self.per_minute = per_minute
+        self.window_seconds = window_seconds
+        self._time_fn = time_fn or time.time
+        self._lock = threading.Lock()
+        self._hits = {}
+
+    def allow(self, key):
+        now = self._time_fn()
+        window = self.window_seconds
+        limit = self.per_minute
+        with self._lock:
+            queue = self._hits.get(key)
+            if queue is None:
+                queue = deque()
+                self._hits[key] = queue
+            cutoff = now - window
+            while queue and queue[0] <= cutoff:
+                queue.popleft()
+            if len(queue) >= limit:
+                return False
+            queue.append(now)
+            # 键数兜底：长期运行的 server 不应让 _hits 无限增长；
+            # 超过 10000 个 key 时清掉最久未用的空队列（不影响计数）。
+            if len(self._hits) > 10000:
+                for old_key in list(self._hits):
+                    if not self._hits[old_key]:
+                        del self._hits[old_key]
+                        break
+            return True
+
+
+def _db_backend_name():
+    """当前数据库后端名：环境变量，未设则 sqlite；绝不返回连接串。"""
+    return (os.environ.get("CLOUD_DB_BACKEND") or "sqlite").strip().lower()
 
 
 def _connect(db_path):
@@ -343,7 +415,22 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         query = parse_qs(parsed.query)
 
         if path == "/health" and method == "GET":
-            self._send_json(200, {"ok": True, "service": "cloud-control-plane"})
+            # 健康检查不走限流与认证；db_ok 只做一次轻量查询，
+            # 绝不返回连接串、密码或任何凭据。
+            db_ok = False
+            try:
+                conn = _connect(self.server.db_path)
+                try:
+                    conn.execute("SELECT 1").fetchone()
+                    db_ok = True
+                finally:
+                    conn.close()
+            except Exception:
+                db_ok = False
+            self._send_json(200, {
+                "ok": True, "service": "cloud-control-plane",
+                "backend": _db_backend_name(), "db_ok": db_ok,
+            })
             return
         if path.startswith("/oauth/"):
             # OAuth 登录/回调是 pre-auth 端点（用户还没本站 API key，
@@ -373,6 +460,37 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._send_json(404, {"error": "not found"})
             return
 
+        # 限流在认证之前、健康/OAuth/webhook 之后：超限返回 429，
+        # 不是 401（不要把“请求太频繁”记成“认证失败”）。
+        # key：有 X-API-Key 时按 key 的 sha256 计（原始 key 不存），
+        # 无 key 时按客户端 IP 计。
+        limiter = getattr(self.server, "rate_limiter", None)
+        if limiter is None:
+            limiter = RateLimiter(
+                per_minute=_rate_limit_from_env(
+                    "CLOUD_RATE_LIMIT_PER_MINUTE",
+                    DEFAULT_RATE_LIMIT_PER_MINUTE),
+                window_seconds=_rate_limit_from_env(
+                    "CLOUD_RATE_LIMIT_WINDOW_SECONDS",
+                    DEFAULT_RATE_LIMIT_WINDOW_SECONDS))
+            self.server.rate_limiter = limiter
+        # 测试注入：server.rate_limit_per_minute / rate_limit_window_seconds
+        # 改动后即时生效，不用重启 server。
+        if getattr(self.server, "rate_limit_per_minute", None) is not None:
+            limiter.per_minute = self.server.rate_limit_per_minute
+        if getattr(self.server, "rate_limit_window_seconds", None) is not None:
+            limiter.window_seconds = self.server.rate_limit_window_seconds
+        api_key_header = self.headers.get("X-API-Key")
+        if api_key_header:
+            limit_key = "key:" + hashlib.sha256(
+                api_key_header.encode("utf-8")).hexdigest()
+        else:
+            limit_key = "ip:" + (self.client_address[0]
+                                 if self.client_address else "unknown")
+        if not limiter.allow(limit_key):
+            self._send_json(429, {"error": "rate_limited"})
+            return
+
         user_id = self._authenticate()
         if user_id is None:
             self._send_json(401, {"error": "unauthorized"})
@@ -394,6 +512,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             self._list_tasks(user_id, query)
         elif path == "/admin/sweep" and method == "POST":
             self._admin_sweep()
+        elif path == "/admin/backup" and method == "POST":
+            self._admin_backup(user_id)
         elif path == "/admin/push/dispatch" and method == "POST":
             self._push_dispatch_endpoint(user_id)
         elif path.startswith("/tasks/"):
@@ -1116,6 +1236,63 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         swept = sweep_stale_tasks(self.server.db_path, threshold)
         self._send_json(200, {"swept": swept})
 
+    def _admin_backup(self, user_id):
+        """POST /admin/backup：备份当前后端数据库，返回路径与规模信息。
+
+        权限：调用方须在至少一个项目里是 maintainer 以上（全局
+        管理端点没有单个项目上下文，按“任一项目 maintainer+”
+        判定；无此角色返回 403）。返回只含 backend、备份文件
+        路径、文件大小与任务行数，绝不返回数据库密码、备份文件
+        内容或任何凭据。out_dir 可由 body 指定，否则走环境变量
+        CLOUD_BACKUP_DIR，未设则用 cloud-backups。
+        """
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        conn = self._db()
+        try:
+            member_rows = conn.execute(
+                "SELECT role FROM project_members WHERE user_id = ?",
+                (user_id,),
+            ).fetchall()
+        finally:
+            conn.close()
+        role_levels = {"viewer": 0, "contributor": 1,
+                       "maintainer": 2, "owner": 3}
+        is_maintainer = any(
+            role_levels.get(row[0], 0) >= 2 for row in member_rows)
+        if not is_maintainer:
+            self._send_json(403, {"error": "forbidden"})
+            return
+        out_dir = body.get("out_dir") or os.environ.get(
+            "CLOUD_BACKUP_DIR") or "cloud-backups"
+        if not isinstance(out_dir, str) or not out_dir:
+            self._send_json(400, {"error": "out_dir must be a string"})
+            return
+        backend = _db_backend_name()
+        try:
+            if backend == "postgres":
+                from cloud.backup import backup_postgres as _backup_pg
+                dest = _backup_pg(out_dir, keep=3)
+            else:
+                from cloud.backup import backup_db as _backup_sqlite
+                dest = _backup_sqlite(self.server.db_path, out_dir, keep=3)
+            size_bytes = os.path.getsize(dest)
+            conn = self._db()
+            try:
+                task_count = conn.execute(
+                    "SELECT COUNT(*) FROM tasks").fetchone()[0]
+            finally:
+                conn.close()
+        except Exception as exc:
+            self._send_json(500, {"error": "backup failed: %s" % (exc,)})
+            return
+        self._send_json(200, {
+            "backend": backend, "path": dest,
+            "size_bytes": size_bytes, "task_count": task_count,
+        })
+
     def _metrics(self, user_id):
         """可观测性：任务计数、成功率、平均耗时、fallback、成本合计。
 
@@ -1136,7 +1313,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 "SELECT project_id, status, started_at, finished_at"
                 " FROM tasks"
             ).fetchall()
-            by_status = {}
+            # 按状态分计数：七个状态键恒存在（缺失为 0），调用方
+            # 不用再猜某个状态是“零”还是“没统计”。
+            by_status = {key: 0 for key in TASK_STATUS_KEYS}
             durations = []
             for row in rows:
                 if row["project_id"] not in member_ids:
@@ -1413,6 +1592,21 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
 
     # -- webhooks（签名认证，不走 X-API-Key）---------------------------
     def _handle_github_webhook(self):
+        """GitHub webhook：验签 -> delivery 去重 -> 解析 -> 执行建任务。
+
+        顺序是故意的：secret 未配 503、签名无效 401（两者都在
+        去重之前，避免未验签的重复送达污染去重表）；同一
+        X-GitHub-Delivery 第二次送达直接回 200 duplicate，不
+        重复建任务、不重复记审计。/fix-review 解析成功后在
+        控制面建一个 queued 任务（prompt 含 repo、PR 号、评论
+        原文、作者），由 worker 认领执行——webhook 本身不跑模型。
+        任务归属：按项目名 == repo full_name 匹配项目；没有匹配
+        项目时不凭空建项目、不建任务，返回 task_id 为 None 并
+        说明原因（worker 的 repo 仍是配置里的 repo，不是任意
+        GitHub 仓库的自动克隆，见 README）。check_run /
+        check_suite 只做通知解析，返回解析结果而不建任务、不改
+        云端任务状态（CI 状态回填由 pr-checks 主动拉取）。
+        """
         raw = self._read_body()
         secret = os.environ.get("GITHUB_WEBHOOK_SECRET")
         if not secret:
@@ -1430,12 +1624,93 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         if not isinstance(payload, dict):
             self._send_json(400, {"error": "body must be a JSON object"})
             return
-        # 只解析不执行：/fix-review 的真正执行由控制面后续调 host.py 完成。
+        event = self.headers.get("X-GitHub-Event")
+        delivery_id = self.headers.get("X-GitHub-Delivery")
+        repository = payload.get("repository") or {}
+        repo_name = repository.get("full_name")
+        store = WebhookDeliveryStore(self.server.db_path)
+        try:
+            if not store.claim(delivery_id, event=event, repo=repo_name):
+                self._send_json(200, {
+                    "duplicate": True, "delivery_id": delivery_id,
+                })
+                return
+        finally:
+            try:
+                store.close()
+            except Exception:
+                pass
+        # check_run / check_suite：只解析，不建任务、不改状态。
+        check = parse_github_check_run(payload)
+        if check is not None:
+            self._send_json(200, check)
+            return
+        # issue_comment 与 pull_request_review_comment：后者把
+        # pull_request.number 归一化成 parser 要的 issue.number。
         result = parse_github_pr_comment(payload)
+        if result is None and payload.get("pull_request") and \
+                payload.get("comment"):
+            normalized = dict(payload)
+            normalized["issue"] = {
+                "number": (payload["pull_request"] or {}).get("number")}
+            result = parse_github_pr_comment(normalized)
         if result is None:
             self._send_json(200, {"result": None})
-        else:
-            self._send_json(200, result)
+            return
+        # /fix-review 执行：建 queued 任务（不跑模型，worker 认领）。
+        conn = self._db()
+        try:
+            project_row = None
+            if repo_name:
+                project_row = conn.execute(
+                    "SELECT id FROM projects WHERE name = ?",
+                    (repo_name,),
+                ).fetchone()
+            if project_row is None:
+                self._send_json(200, dict(
+                    result, task_id=None, project_id=None,
+                    error="no matching project for repo"))
+                return
+            project_id = project_row[0]
+            task_id = uuid.uuid4().hex
+            now = time.time()
+            prompt = (
+                "[webhook /fix-review] repo=%s pr=%s author=%s\n%s"
+                % (result.get("repo"), result.get("pr_number"),
+                   result.get("author"), result.get("comment")))
+            try:
+                conn.execute(
+                    "INSERT INTO tasks(id, project_id, prompt, status,"
+                    " idempotency_key, depends_on, priority, created_by,"
+                    " created_at, updated_at, last_heartbeat)"
+                    " VALUES(?, ?, ?, 'queued', ?, '[]', 0, 'system',"
+                    " ?, ?, NULL)",
+                    (task_id, project_id, prompt,
+                     "webhook:%s" % delivery_id if delivery_id else None,
+                     now, now),
+                )
+                conn.commit()
+            except (sqlite3.IntegrityError, _PgIntegrityError):
+                # 并发/去重兜底：同 delivery 已建过任务时回读它，
+                # 不建重复任务、不重复记审计。
+                row = conn.execute(
+                    "SELECT id FROM tasks WHERE project_id = ?"
+                    " AND idempotency_key = ?",
+                    (project_id,
+                     "webhook:%s" % delivery_id if delivery_id else None),
+                ).fetchone()
+                if row:
+                    self._send_json(200, dict(
+                        result, task_id=row[0], project_id=project_id,
+                        duplicate=True))
+                    return
+                raise
+        finally:
+            conn.close()
+        with _auth_store(self.server.db_path) as auth:
+            auth.audit("system", project_id, "webhook_fix_review", task_id)
+        self._send_json(200, dict(
+            result, task_id=task_id, project_id=project_id))
 
     def _handle_slack_webhook(self):
         raw = self._read_body()
@@ -1488,6 +1763,16 @@ def create_server(db_path, host="127.0.0.1", port=8765, backend=None):
     server = ThreadingHTTPServer((host, port), CloudControlPlaneHandler)
     server.db_path = db_path
     server.daemon_threads = True
+    # 限流器：内存滑动窗口，默认值与环境变量覆盖见 RateLimiter；
+    # 测试可直接改 server.rate_limit_per_minute / window_seconds。
+    server.rate_limiter = RateLimiter(
+        per_minute=_rate_limit_from_env(
+            "CLOUD_RATE_LIMIT_PER_MINUTE", DEFAULT_RATE_LIMIT_PER_MINUTE),
+        window_seconds=_rate_limit_from_env(
+            "CLOUD_RATE_LIMIT_WINDOW_SECONDS",
+            DEFAULT_RATE_LIMIT_WINDOW_SECONDS))
+    server.rate_limit_per_minute = None
+    server.rate_limit_window_seconds = None
     # OAuth 回调注入点：生产为 None（走真实换 token + 真实 userinfo），
     # 测试可注入 fake http_client / userinfo_client；绝不注入真实 secret。
     server.oauth_http_client = None