"""云端控制面 HTTP server（首版）。 零第三方依赖，只用 http.server + sqlite3 + 标准库。 分工与边界（以实际代码为准，不夸大）： - 本 server 只做控制面：项目/任务的创建、查询、取消、心跳、 stale 任务回收，以及 GitHub/Slack webhook 的验签与解析。 - 本版不直接跑模型回合。任务状态机与 host.py 对齐： queued -> running -> completed/failed/interrupted/merged/merge-failed， 其中 completed/merged 由 worker/控制面后续调用方更新（例如 worker 调 host.py 跑完回合后回写），本 server 不提供“跑模型”端点。 - 其它已就绪模块（queue 的 order_tasks/ready_tasks、sandbox 的执行器、 memory 的 MemoryStore、kb 的 KBStore、models 的 ModelRouter、 cost 的 CostStore、webhooks 的 NotificationService）本版尚未接成 HTTP 端点，留给 worker 接线时调用；本 server 只接入其中两处： AuthStore（认证/权限/审计）与 webhooks 的验签/解析函数。 - DockerExecutor 已接通（见 sandbox.py），CloudExecutor 目前仍是占位 （见 sandbox.py），本 server 不假装已接通云端执行。 导入规则：只用绝对包导入（from cloud.auth import ...）。绝不要把 cloud/ 目录加进 sys.path：标准库已有 queue 模块，会和 cloud/queue.py 撞名，导致 stdlib queue 被错误模块覆盖。 状态库：单一 SQLite 文件。AuthStore 用 users/project_members/ audit_log 三张表；本 server 用 projects/tasks 两张表，表名不冲突。 并发：ThreadingHTTPServer 多线程处理请求，每个请求内新建 SQLite 连接（server 表用新连接，AuthStore 每次请求新建一个实例），绝不在 多线程 handler 间共用一个连接。 启动： python3 -m cloud.server [--host 127.0.0.1] [--port 8765] [--db <路径>] 后端由 CLOUD_DB_BACKEND 选（sqlite 默认；postgres 已接通， 驱动缺失时启动即明确报错，见 cloud/db.py）。 测试用 create_server(db_path, host, port) 只返回 ThreadingHTTPServer 实例，不自动 serve。 """ 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, quote, urlparse from cloud.auth import AuthStore from cloud.cost import CostStore from cloud.db import connect as _db_connect from cloud.db import create_database as _create_database from cloud.db import init_postgres_schema as _init_postgres_schema from cloud.db import PostgresIntegrityError as _PgIntegrityError from cloud.db import PostgresOperationalError as _PgOperationalError from cloud.kb import KBStore from cloud.memory import MemoryStore from cloud.pr import PrSkipped, PrStepError, create_task_pr from cloud.workers import ( heartbeat_worker as _registry_heartbeat, list_workers as _registry_list, ) from cloud.oauth import ( BindError, OAuthError, OAuthStateStore, bind_external_identity, build_authorization_url, default_userinfo_client, find_bound_user, handle_oauth_callback, list_bindings, map_external_identity_to_local_user_with_key, start_oauth_login, unbind_external_identity, ) from cloud.repos import validate_repo_url from cloud.push import ( create_subscription as _push_create, delete_subscription as _push_delete, dispatch_event as _push_dispatch, list_subscriptions as _push_list, ) from cloud.github_app import ( DEFAULT_APP_NAME, DEFAULT_APP_URL, AppError, SetupStateStore, build_app_manifest, claim_fix_review, exchange_manifest_code, get_app_config, list_installations, parse_installation_event, public_app_status, record_installation_repos, remove_installation, remove_installation_repos, save_app_config, set_fix_review_task_id, ) from cloud.webhooks import ( WebhookDeliveryStore, parse_github_check_run, parse_github_pr_comment, parse_slack_command, verify_github_signature_any, verify_slack_signature, ) 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:”或“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 _worker_heartbeat_status(server=None): """读 worker 运维心跳，返回 (age_seconds_or_None, worker_ok)。 心跳文件路径：测试注入优先（server.worker_heartbeat_file）， 其次环境变量 CLOUD_WORKER_HEARTBEAT_FILE，未设则默认 ./worker-heartbeat.json——与 worker 的默认路径保持一致，主机 部署时两边都要指到同一个主机路径（见 deploy/README）。 阈值：server.worker_heartbeat_max_age_seconds（测试注入）或 环境变量 CLOUD_WORKER_HEARTBEAT_MAX_AGE_SECONDS，默认 180 秒。 文件不存在、读失败、JSON 非法、last_seen 非数字时一律返回 (None, False)，不抛异常、不影响 health 本身的 200。 """ path = None max_age = 180 if server is not None: path = getattr(server, "worker_heartbeat_file", None) override = getattr(server, "worker_heartbeat_max_age_seconds", None) if override is not None: try: max_age = float(override) except (TypeError, ValueError): max_age = 180 if not path: path = os.environ.get("CLOUD_WORKER_HEARTBEAT_FILE") or \ "./worker-heartbeat.json" if max_age == 180: try: max_age = float(os.environ.get( "CLOUD_WORKER_HEARTBEAT_MAX_AGE_SECONDS") or 180) except (TypeError, ValueError): max_age = 180 try: with open(path, encoding="utf-8") as f: payload = json.load(f) last_seen = float(payload.get("last_seen")) except Exception: return None, False age = time.time() - last_seen return round(age, 1), bool(age <= max_age) def _connect(db_path): """每个请求新建一个连接；后端按 CLOUD_DB_BACKEND 分发。 cloud.db.connect() 本身已按环境变量分发：sqlite 时是 WAL + busy_timeout 的 SQLite 连接；postgres 时是 psycopg 适配连接（db_path 被忽略，连接参数只走 CLOUD_DB_* 环境变量，驱动缺失时明确报错，绝不回落）。因此 server 自身的表与 AuthStore / MemoryStore 等沿用 connect() 的 调用点在 PG 模式下同样走 Postgres，无需逐一改写。 """ return _db_connect(str(db_path)) def migrate_tasks_table(conn): """幂等迁移：给老库 tasks 表补新列，不丢数据。 新列：result_text、error_text、actual_model、started_at、 finished_at、child_pid。PRAGMA table_info 查现有列，缺哪个 补哪个；已迁移的库再跑一次是 no-op。worker 写执行结果、 server 读任务时都依赖这些列存在。 """ cols = {row[1] for row in conn.execute( "PRAGMA table_info(tasks)").fetchall()} new_cols = { "result_text": "TEXT", "error_text": "TEXT", "actual_model": "TEXT", "started_at": "REAL", "finished_at": "REAL", "child_pid": "INTEGER", "executor": "TEXT", "pr_url": "TEXT", "pr_number": "INTEGER", } for name, coltype in new_cols.items(): if name not in cols: conn.execute( "ALTER TABLE tasks ADD COLUMN %s %s" % (name, coltype)) def migrate_projects_table(conn): """幂等迁移：给老库 projects 表补 repo_url 列，不丢数据。 与 migrate_tasks_table 同款写法：PRAGMA table_info 查现有列 （Postgres 适配层把它翻译成 information_schema 查询，调用方 取 row[1] 即列名），缺 repo_url 时 ALTER TABLE 补列，已迁移 的库再跑一次是 no-op。SQLite 与 Postgres 两支都走它。 """ cols = {row[1] for row in conn.execute( "PRAGMA table_info(projects)").fetchall()} if "repo_url" not in cols: conn.execute("ALTER TABLE projects ADD COLUMN repo_url TEXT") if "auto_pr" not in cols: conn.execute("ALTER TABLE projects ADD COLUMN auto_pr INTEGER") def _is_postgres_backend(): backend = os.environ.get("CLOUD_DB_BACKEND") or "sqlite" return str(backend).strip().lower() in ("postgres", "postgresql") def init_db(db_path): """创建 server 自己的 projects/tasks 表（AuthStore 的表由其自建）。 Postgres 模式：建 cloud.db.POSTGRES_SCHEMA 的全部表（其中 projects/tasks 已含 migrate_tasks_table 的 6 列），再跑一次 migrate_tasks_table——PRAGMA 翻译后在 PG 上是 no-op，老库 迁移仍走同一函数。SQLite 模式：原行为不变。 """ conn = _connect(db_path) try: if _is_postgres_backend(): _init_postgres_schema(conn) migrate_tasks_table(conn) migrate_projects_table(conn) conn.commit() return conn.execute( "CREATE TABLE IF NOT EXISTS projects(" "id TEXT PRIMARY KEY, " "name TEXT, " "owner_id TEXT, " "created_at REAL, " "repo_url TEXT, " "auto_pr INTEGER)" ) conn.execute( "CREATE TABLE IF NOT EXISTS tasks(" "id TEXT PRIMARY KEY, " "project_id TEXT, " "prompt TEXT, " "status TEXT, " "idempotency_key TEXT, " "depends_on TEXT, " "priority INTEGER, " "created_by TEXT, " "created_at REAL, " "updated_at REAL, " "last_heartbeat REAL, " "UNIQUE(project_id, idempotency_key))" ) migrate_tasks_table(conn) migrate_projects_table(conn) # worker 每轮 poll 都按 status 取任务（见 load_tasks 的 # WHERE status IN (...)），无索引时是全表扫描；任务表随 # 时间变大后索引是必须的。IF NOT EXISTS 保证老库幂等。 conn.execute( "CREATE INDEX IF NOT EXISTS idx_tasks_status" " ON tasks(status)") conn.commit() finally: conn.close() @contextmanager def _auth_store(db_path): """每请求一个 AuthStore 实例（自带新连接），用完即关闭。""" store = AuthStore(db_path) try: yield store finally: try: store.conn.close() except Exception: pass def _row_get(row, name): """sqlite Row 的容错取值：老库行缺列时返回 None 不炸。""" try: return row[name] except (KeyError, IndexError, TypeError): return None def _task_to_dict(row): depends_on = row["depends_on"] try: depends_on = json.loads(depends_on) if depends_on else [] except (TypeError, ValueError): depends_on = [] if not isinstance(depends_on, list): depends_on = [depends_on] return { "id": row["id"], "project_id": row["project_id"], "prompt": row["prompt"], "status": row["status"], "idempotency_key": row["idempotency_key"], "depends_on": depends_on, "priority": row["priority"], "created_by": row["created_by"], "created_at": row["created_at"], "updated_at": row["updated_at"], "last_heartbeat": row["last_heartbeat"], "result_text": _row_get(row, "result_text"), "error_text": _row_get(row, "error_text"), "actual_model": _row_get(row, "actual_model"), "started_at": _row_get(row, "started_at"), "finished_at": _row_get(row, "finished_at"), "child_pid": _row_get(row, "child_pid"), "executor": _row_get(row, "executor"), "pr_url": _row_get(row, "pr_url"), "pr_number": _row_get(row, "pr_number"), } def _project_to_dict(row): return { "id": row["id"], "name": row["name"], "owner_id": row["owner_id"], "created_at": row["created_at"], # 老库行缺 repo_url 列时 _row_get 给 None，不炸（列本身由 # migrate_projects_table 在 init_db 时补，老行读到也是 None）。 "repo_url": _row_get(row, "repo_url"), # auto_pr 默认 0（关闭）：老行缺列时 None 按 0 计，绝不因 # 迁移就把老项目默认改成自动推。 "auto_pr": _row_get(row, "auto_pr") or 0, } def sweep_stale_tasks(db_path, threshold_seconds=300): """把 running 且心跳过期的任务改为 interrupted，返回被扫任务 id 列表。 心跳取 last_heartbeat，无心跳则用 updated_at；早于 now - threshold_seconds 即过期。审计记一条 "sweep_interrupted" （user_id 记 "system"，因为回收不是某个用户的操作）。 独立可单测，不依赖 HTTP。 """ now = time.time() cutoff = now - threshold_seconds conn = _connect(db_path) try: conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT id, project_id FROM tasks WHERE status = 'running'" " AND COALESCE(last_heartbeat, updated_at) < ?", (cutoff,), ).fetchall() swept = [] for row in rows: cur = conn.execute( "UPDATE tasks SET status = 'interrupted', updated_at = ?" " WHERE id = ? AND status = 'running'", (now, row["id"]), ) if cur.rowcount: swept.append(row["id"]) conn.commit() stale = [(row["id"], row["project_id"]) for row in rows if row["id"] in swept] finally: conn.close() with _auth_store(db_path) as store: for task_id, project_id in stale: store.audit("system", project_id, "sweep_interrupted", task_id) return swept class CloudControlPlaneHandler(BaseHTTPRequestHandler): """控制面请求处理：除 /health 与 /webhooks/* 外一律要求 X-API-Key。""" protocol_version = "HTTP/1.1" # -- 通用工具 ------------------------------------------------------ def log_message(self, *args): # 测试输出保持干净 pass def _send_json(self, status, obj): body = json.dumps(obj).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _send_html(self, status, html_body): body = html_body.encode("utf-8") self.send_response(status) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _read_body(self): try: length = int(self.headers.get("Content-Length") or 0) except ValueError: length = 0 if length <= 0: return b"" return self.rfile.read(length) def _read_json(self): """返回 (obj, error)：空 body 视为 {}；非法 JSON 返回 error。""" raw = self._read_body() if not raw: return {}, None try: obj = json.loads(raw.decode("utf-8")) except (ValueError, UnicodeDecodeError): return None, "invalid json" if not isinstance(obj, dict): return None, "body must be a JSON object" return obj, None def _db(self): conn = _connect(self.server.db_path) conn.row_factory = sqlite3.Row return conn def _authenticate(self): """X-API-Key -> user_id；无 key 或 key 错返回 None。""" api_key = self.headers.get("X-API-Key") if not api_key: return None with _auth_store(self.server.db_path) as store: return store.authenticate(api_key) # -- 入口 ---------------------------------------------------------- def do_GET(self): self._dispatch("GET") def do_POST(self): self._dispatch("POST") def do_DELETE(self): self._dispatch("DELETE") def _dispatch(self, method): parsed = urlparse(self.path) path = parsed.path if len(path) > 1 and path.endswith("/"): path = path.rstrip("/") query = parse_qs(parsed.query) if path == "/health" and method == "GET": # 健康检查不走限流与认证；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 # worker 字段：worker_ok 只表示“worker 心跳新鲜”，worker # 不 ok 不代表 server 挂了，所以 health 本身仍返回 200， # 调用方看 worker_ok 字段判断，不要把 health 改成 503 # 破坏已有调用（monitor.sh 也是按字段判断，不按状态码）。 age, worker_ok = _worker_heartbeat_status(self.server) self._send_json(200, { "ok": True, "service": "cloud-control-plane", "backend": _db_backend_name(), "db_ok": db_ok, "worker_last_seen_seconds_ago": age, "worker_ok": worker_ok, }) return if path.startswith("/oauth/"): # OAuth 登录/回调是 pre-auth 端点（用户还没本站 API key， # 这正是 OAuth 要解决的问题），靠 state + client_secret # 认证，不走 X-API-Key；不要把它当成无认证后门。 # 绑定端点例外：绑定必须知道绑到哪个已有账号，所以在 # 处理器内验 X-API-Key（无 key 401），分发仍走这支。 bind = self._oauth_bind_route(path) if bind is not None: provider, action = bind if method == "POST" and action == "bind_initiate": self._oauth_bind_initiate(provider) elif method == "GET" and action == "bindings": self._oauth_bindings_get(provider) elif method == "DELETE" and action == "bindings": self._oauth_bindings_delete(provider) else: self._send_json(404, {"error": "not found"}) return if method == "GET" and path.endswith("/login"): self._oauth_login(path) elif method == "GET" and path.endswith("/callback"): # 真实浏览器流程：provider 重定向到这里，query 带 # code + state（不带 client_id，靠 state 记录取回）。 self._oauth_callback_get(path) elif method == "POST" and path.endswith("/callback"): self._oauth_callback(path) else: self._send_json(404, {"error": "not found"}) return if path.startswith("/github-app/"): # GitHub App 一键安装：setup/state 与 status 须持本站 key # （在处理器内验）；setup 表单页与 callback 靠 setup # state 凭证（一次性、1 小时），不走 X-API-Key（浏览器 # 回调带不了请求头，state 就是凭证）。 if method == "POST" and path == "/github-app/setup/state": self._github_app_setup_state() elif method == "GET" and path == "/github-app/setup": self._github_app_setup_page() elif method == "GET" and path == "/github-app/callback": self._github_app_callback() elif method == "GET" and path == "/github-app/status": self._github_app_status() else: self._send_json(404, {"error": "not found"}) return if path.startswith("/webhooks/"): # webhook 靠签名认证（外部服务无法持 X-API-Key），不走 API key。 if method != "POST": self._send_json(404, {"error": "not found"}) return if path == "/webhooks/github": self._handle_github_webhook() elif path == "/webhooks/slack": self._handle_slack_webhook() else: 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"}) return if path == "/projects" and method == "POST": self._create_project(user_id) elif path == "/projects" and method == "GET": self._list_projects(user_id) elif path == "/workers" and method == "GET": self._workers_list(user_id) elif path == "/workers/heartbeat" and method == "POST": self._workers_heartbeat(user_id) elif path == "/metrics" and method == "GET": self._metrics(user_id) elif path == "/models" and method == "GET": self._models_overview(user_id) elif path.startswith("/projects/"): self._dispatch_project(user_id, method, path, query) elif path == "/tasks" and method == "POST": self._create_task(user_id) elif path == "/tasks" and method == "GET": 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/"): rest = path[len("/tasks/"):] if "/" not in rest and method == "GET": self._get_task(user_id, rest) elif rest.endswith("/cost") and method == "GET": self._task_cost(user_id, rest[: -len("/cost")]) elif rest.endswith("/cancel") and method == "POST": self._cancel_task(user_id, rest[: -len("/cancel")]) elif rest.endswith("/merge") and method == "POST": self._merge_task(user_id, rest[: -len("/merge")]) elif rest.endswith("/pr") and method == "POST": self._task_pr(user_id, rest[: -len("/pr")]) elif rest.endswith("/retry") and method == "POST": self._retry_task(user_id, rest[: -len("/retry")]) elif rest.endswith("/heartbeat") and method == "POST": self._heartbeat_task(user_id, rest[: -len("/heartbeat")]) else: self._send_json(404, {"error": "not found"}) else: self._send_json(404, {"error": "not found"}) # -- 项目 ---------------------------------------------------------- def _create_project(self, user_id): body, err = self._read_json() if err: self._send_json(400, {"error": err}) return name = body.get("name") if not name or not isinstance(name, str): self._send_json(400, {"error": "name required"}) return project_id = body.get("id") or uuid.uuid4().hex if not isinstance(project_id, str): self._send_json(400, {"error": "id must be a string"}) return # repo_url 可选：不传时为 None（老行为不变）；传了则与 worker # 共用 cloud.repos.validate_repo_url 的同一套校验，不合法时 # 400 且不建项目，绝不把带凭证的 URL 存进库。 repo_url = body.get("repo_url") if repo_url is not None: try: repo_url = validate_repo_url(repo_url) except ValueError as exc: self._send_json(400, {"error": str(exc)}) return # auto_pr 可选，默认 0（关闭）：只接受 0/1（布尔 True/False # 按 1/0 计），其它值 400 且不建项目。本轮没有现成的 # PATCH /projects/{id} 路由，auto_pr 只能在创建时指定， # 建后改它需直接改库或重建项目（见 README）。 auto_pr_raw = body.get("auto_pr", 0) if isinstance(auto_pr_raw, bool): auto_pr = 1 if auto_pr_raw else 0 elif isinstance(auto_pr_raw, int) and auto_pr_raw in (0, 1): auto_pr = auto_pr_raw else: self._send_json(400, {"error": "auto_pr must be 0 or 1"}) return now = time.time() conn = self._db() try: existing = conn.execute( "SELECT id FROM projects WHERE id = ?", (project_id,) ).fetchone() if existing: self._send_json(409, {"error": "project exists"}) return conn.execute( "INSERT INTO projects(id, name, owner_id, created_at," " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, ?)", (project_id, name, user_id, now, repo_url, auto_pr), ) conn.commit() project = { "id": project_id, "name": name, "owner_id": user_id, "created_at": now, "repo_url": repo_url, "auto_pr": auto_pr, } except (sqlite3.IntegrityError, _PgIntegrityError): self._send_json(409, {"error": "project exists"}) return finally: conn.close() # AuthStore.create_project 只写成员表（owner 角色），项目记录 # 本身在上面的 projects 表；两者配合才算一个完整项目。 with _auth_store(self.server.db_path) as store: store.create_project(project_id, user_id) store.audit(user_id, project_id, "create_project", name) self._send_json(201, project) def _list_projects(self, user_id): conn = self._db() try: member_ids = { row["project_id"] for row in conn.execute( "SELECT project_id FROM project_members WHERE user_id = ?", (user_id,), ).fetchall() } rows = conn.execute( "SELECT id, name, owner_id, created_at, repo_url, auto_pr" " FROM projects ORDER BY created_at, id" ).fetchall() finally: conn.close() projects = [_project_to_dict(r) for r in rows if r["id"] in member_ids] self._send_json(200, {"projects": projects}) # -- 项目子资源：记忆 / 知识库 / 成本 ------------------------------- def _project_exists(self, project_id): conn = self._db() try: row = conn.execute( "SELECT id FROM projects WHERE id = ?", (project_id,) ).fetchone() return row is not None finally: conn.close() def _check_project(self, user_id, project_id, action): """返回 None 表示通过；否则返回 (status, error) 先行响应。""" if not self._project_exists(project_id): return 404, "project not found" with _auth_store(self.server.db_path) as store: if not store.check_permission(user_id, project_id, action): return 403, "forbidden" return None def _dispatch_project(self, user_id, method, path, query): rest = path[len("/projects/"):] parts = rest.split("/") project_id = parts[0] if parts else "" if not project_id: self._send_json(404, {"error": "not found"}) return # /projects/{id}/memory if len(parts) == 2 and parts[1] == "memory": if method == "POST": err = self._check_project(user_id, project_id, "create_task") if err: self._send_json(err[0], {"error": err[1]}) return self._memory_create(user_id, project_id) elif method == "GET": err = self._check_project(user_id, project_id, "read") if err: self._send_json(err[0], {"error": err[1]}) return self._memory_list(project_id, query) else: self._send_json(404, {"error": "not found"}) return # /projects/{id}/memory/{mid}/supersede if len(parts) == 4 and parts[1] == "memory" and parts[3] == "supersede": if method != "POST": self._send_json(404, {"error": "not found"}) return err = self._check_project(user_id, project_id, "create_task") if err: self._send_json(err[0], {"error": err[1]}) return self._memory_supersede(user_id, project_id, parts[2]) return # /projects/{id}/kb/index if len(parts) == 3 and parts[1] == "kb" and parts[2] == "index": if method != "POST": self._send_json(404, {"error": "not found"}) return err = self._check_project(user_id, project_id, "create_task") if err: self._send_json(err[0], {"error": err[1]}) return self._kb_index(user_id, project_id) return # /projects/{id}/kb/search if len(parts) == 3 and parts[1] == "kb" and parts[2] == "search": if method != "GET": self._send_json(404, {"error": "not found"}) return err = self._check_project(user_id, project_id, "read") if err: self._send_json(err[0], {"error": err[1]}) return self._kb_search(project_id, query) return # /projects/{id}/cost if len(parts) == 2 and parts[1] == "cost": if method != "GET": self._send_json(404, {"error": "not found"}) return err = self._check_project(user_id, project_id, "read") if err: self._send_json(err[0], {"error": err[1]}) return self._project_cost(project_id, query) return # /projects/{id}/push/subscriptions if len(parts) == 3 and parts[1] == "push" and parts[2] == "subscriptions": if method == "POST": err = self._check_project(user_id, project_id, "create_task") if err: self._send_json(err[0], {"error": err[1]}) return self._push_create_endpoint(user_id, project_id) elif method == "GET": err = self._check_project(user_id, project_id, "read") if err: self._send_json(err[0], {"error": err[1]}) return self._send_json(200, { "subscriptions": _push_list( self.server.db_path, project_id, user_id=user_id)}) else: self._send_json(404, {"error": "not found"}) return # /projects/{id}/push/subscriptions/{sid} DELETE if len(parts) == 4 and parts[1] == "push" and parts[2] == "subscriptions": if method != "DELETE": self._send_json(404, {"error": "not found"}) return err = self._check_project(user_id, project_id, "create_task") if err: self._send_json(err[0], {"error": err[1]}) return subs = _push_list(self.server.db_path, project_id) mine = [s for s in subs if s["id"] == parts[3]] if not mine: self._send_json(404, {"error": "subscription not found"}) return # 只能删自己的订阅，项目 owner 可删任意（owner 角色检查）。 with _auth_store(self.server.db_path) as auth: is_owner = auth.check_permission( user_id, project_id, "manage_project") if mine[0]["user_id"] != user_id and not is_owner: self._send_json(403, {"error": "forbidden"}) return deleted = _push_delete(self.server.db_path, subscription_id=parts[3]) self._send_json(200, {"deleted": bool(deleted)}) return self._send_json(404, {"error": "not found"}) def _memory_create(self, user_id, project_id): body, err = self._read_json() if err: self._send_json(400, {"error": err}) return goal = body.get("goal") conclusion = body.get("conclusion") if not goal or not isinstance(goal, str): self._send_json(400, {"error": "goal required"}) return if not conclusion or not isinstance(conclusion, str): self._send_json(400, {"error": "conclusion required"}) return store = MemoryStore(self.server.db_path) try: mid = store.add_summary( project_id, body.get("thread_id") or "", goal, conclusion, files_changed=body.get("files_changed") or [], open_issues=body.get("open_issues") or [], ) finally: try: store.conn.close() except Exception: pass with _auth_store(self.server.db_path) as auth: auth.audit(user_id, project_id, "create_memory", str(mid)) self._send_json(201, {"id": mid, "project_id": project_id}) def _memory_list(self, project_id, query): try: top_k = int((query.get("top_k") or ["5"])[0]) except (TypeError, ValueError): self._send_json(400, {"error": "top_k must be an integer"}) return store = MemoryStore(self.server.db_path) try: summaries = store.list_summaries(project_id) context = store.build_prompt_context(project_id, top_k=top_k) finally: try: store.conn.close() except Exception: pass self._send_json(200, { "project_id": project_id, "summaries": summaries[:top_k], "prompt_context": context, }) def _memory_supersede(self, user_id, project_id, mid_raw): try: mid = int(mid_raw) except (TypeError, ValueError): self._send_json(400, {"error": "id must be an integer"}) return conn = self._db() try: row = conn.execute( "SELECT project_id FROM summaries WHERE id = ?", (mid,) ).fetchone() finally: conn.close() if row is None or row["project_id"] != project_id: self._send_json(404, {"error": "summary not found"}) return store = MemoryStore(self.server.db_path) try: store.deprecate(mid) finally: try: store.conn.close() except Exception: pass with _auth_store(self.server.db_path) as auth: auth.audit(user_id, project_id, "supersede_memory", str(mid)) self._send_json(200, {"id": mid, "status": "deprecated"}) def _kb_index(self, user_id, project_id): body, err = self._read_json() if err: self._send_json(400, {"error": err}) return path = body.get("path") content = body.get("content") if not path or not isinstance(path, str): self._send_json(400, {"error": "path required"}) return if content is None or not isinstance(content, str): self._send_json(400, {"error": "content required"}) return store = KBStore(self.server.db_path) try: doc_id = store.index_file( project_id, path, content, kind=body.get("kind") or "file") except ValueError as e: self._send_json(400, {"error": str(e)}) return finally: try: store.conn.close() except Exception: pass with _auth_store(self.server.db_path) as auth: auth.audit(user_id, project_id, "kb_index", path) self._send_json(201, {"id": doc_id, "project_id": project_id, "path": path}) def _kb_search(self, project_id, query): q = (query.get("q") or [""])[0] if not q: self._send_json(400, {"error": "q required"}) return try: top_k = int((query.get("top_k") or ["5"])[0]) except (TypeError, ValueError): self._send_json(400, {"error": "top_k must be an integer"}) return store = KBStore(self.server.db_path) try: results = store.search(project_id, q, top_k=top_k) finally: try: store.conn.close() except Exception: pass self._send_json(200, {"project_id": project_id, "query": q, "results": results}) def _project_cost(self, project_id, query): store = CostStore(self.server.db_path) try: overall = store.report_by_project(project_id) conn = store._conn conn.row_factory = sqlite3.Row days = (query.get("days") or [None])[0] day = (query.get("day") or [None])[0] daily = [] if day: row = conn.execute( "SELECT day, COALESCE(SUM(total_tokens),0) AS t," " COALESCE(SUM(usd_micros),0) AS u, COUNT(*) AS c," " COALESCE(SUM(fallback_count),0) AS f" " FROM cost_records WHERE project_id = ? AND day = ?" " GROUP BY day", (project_id, day)).fetchone() if row: daily = [{"day": row["day"], "total_tokens": row["t"], "usd_micros": row["u"], "turns": row["c"], "fallback_count": row["f"]}] elif days is not None: try: n = int(days) except (TypeError, ValueError): self._send_json(400, {"error": "days must be an integer"}) return from datetime import date, timedelta cutoff = (date.today() - timedelta(days=max(n - 1, 0)) ).isoformat() rows = conn.execute( "SELECT day, COALESCE(SUM(total_tokens),0) AS t," " COALESCE(SUM(usd_micros),0) AS u, COUNT(*) AS c," " COALESCE(SUM(fallback_count),0) AS f" " FROM cost_records WHERE project_id = ? AND day >= ?" " GROUP BY day ORDER BY day", (project_id, cutoff)).fetchall() daily = [{"day": r["day"], "total_tokens": r["t"], "usd_micros": r["u"], "turns": r["c"], "fallback_count": r["f"]} for r in rows] finally: try: store._conn.close() except Exception: pass self._send_json(200, {"project_id": project_id, "overall": overall, "daily": daily}) # -- 任务成本与模型总览 ---------------------------------------------- def _task_cost(self, user_id, task_id): row = self._fetch_task(task_id) if row is None: self._send_json(404, {"error": "task not found"}) return task = _task_to_dict(row) with _auth_store(self.server.db_path) as auth: if not auth.check_permission(user_id, task["project_id"], "read"): self._send_json(403, {"error": "forbidden"}) return store = CostStore(self.server.db_path) try: report = store.report_by_task(task_id) finally: try: store._conn.close() except Exception: pass self._send_json(200, {"task_id": task_id, "project_id": task["project_id"], **report}) def _models_overview(self, user_id): # 优先级列表来自环境变量 MODELS_PRIORITY（逗号分隔），未设时 # 用默认列表；这只是控制面的默认配置，不是 provider 的承诺。 raw = os.environ.get("MODELS_PRIORITY", "") priority = [m.strip() for m in raw.split(",") if m.strip()] or [ "gpt-6.1-sol", "gpt-6-sol", "gpt-5.6-sol"] # 只统计调用者是成员的项目，避免跨项目泄露成本。 conn = self._db() try: member_ids = [r["project_id"] for r in conn.execute( "SELECT project_id FROM project_members WHERE user_id = ?", (user_id,)).fetchall()] finally: conn.close() if not member_ids: self._send_json(200, {"priority_list": priority, "model_stats": [], "fallback_events": []}) return placeholders = ",".join("?" for _ in member_ids) store = CostStore(self.server.db_path) try: conn = store._conn conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT model, COUNT(*) AS turns," " COALESCE(SUM(total_tokens),0) AS tokens," " COALESCE(SUM(fallback_count),0) AS fallbacks" " FROM cost_records WHERE project_id IN (%s)" " GROUP BY model ORDER BY model" % placeholders, member_ids).fetchall() fb_rows = conn.execute( "SELECT task_id, project_id, model, fallback_count" " FROM cost_records WHERE project_id IN (%s)" " AND fallback_count > 0" " ORDER BY created_at DESC LIMIT 50" % placeholders, member_ids).fetchall() finally: try: store._conn.close() except Exception: pass self._send_json(200, { "priority_list": priority, "model_stats": [ {"model": r["model"], "turns": r["turns"], "total_tokens": r["tokens"], "fallback_count": r["fallbacks"]} for r in rows], "fallback_events": [ {"task_id": r["task_id"], "project_id": r["project_id"], "model": r["model"], "fallback_count": r["fallback_count"]} for r in fb_rows], }) # -- 任务 ---------------------------------------------------------- def _create_task(self, user_id): body, err = self._read_json() if err: self._send_json(400, {"error": err}) return project_id = body.get("project_id") prompt = body.get("prompt") if not project_id or not isinstance(project_id, str): self._send_json(400, {"error": "project_id required"}) return if not prompt or not isinstance(prompt, str): self._send_json(400, {"error": "prompt required"}) return depends_on = body.get("depends_on") or [] if isinstance(depends_on, str): depends_on = [depends_on] if not isinstance(depends_on, list): self._send_json(400, {"error": "depends_on must be a list"}) return try: priority = int(body.get("priority", 0) or 0) except (TypeError, ValueError): self._send_json(400, {"error": "priority must be an integer"}) return # executor 可选：None 表示任意 worker 可认领；指定时只能是 # local/docker/cloud，非法值 400 且不建任务。 executor = body.get("executor") if executor is not None and executor not in ( "local", "docker", "cloud"): self._send_json(400, { "error": "executor must be local, docker or cloud"}) return idempotency_key = body.get("idempotency_key") if idempotency_key is not None and not isinstance(idempotency_key, str): self._send_json(400, {"error": "idempotency_key must be a string"}) return with _auth_store(self.server.db_path) as store: if not store.check_permission(user_id, project_id, "create_task"): self._send_json(403, {"error": "forbidden"}) return now = time.time() conn = self._db() try: if idempotency_key is not None: row = conn.execute( "SELECT * FROM tasks WHERE project_id = ?" " AND idempotency_key = ?", (project_id, idempotency_key), ).fetchone() if row: self._send_json(200, _task_to_dict(row)) return task_id = uuid.uuid4().hex try: conn.execute( "INSERT INTO tasks(id, project_id, prompt, status," " idempotency_key, depends_on, priority, created_by," " created_at, updated_at, last_heartbeat, executor)" " VALUES(?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?, NULL, ?)", (task_id, project_id, prompt, idempotency_key, json.dumps(depends_on), priority, user_id, now, now, executor), ) conn.commit() except (sqlite3.IntegrityError, _PgIntegrityError): # 并发下同幂等键抢建：回读已有任务，不建重复。 row = conn.execute( "SELECT * FROM tasks WHERE project_id = ?" " AND idempotency_key = ?", (project_id, idempotency_key), ).fetchone() if row: self._send_json(200, _task_to_dict(row)) return raise row = conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,) ).fetchone() task = _task_to_dict(row) finally: conn.close() with _auth_store(self.server.db_path) as store: store.audit(user_id, project_id, "create_task", task_id) self._send_json(201, task) def _list_tasks(self, user_id, query): project_ids = query.get("project_id") project_id = project_ids[0] if project_ids else None if not project_id: self._send_json(400, {"error": "project_id required"}) return with _auth_store(self.server.db_path) as store: if not store.check_permission(user_id, project_id, "read"): self._send_json(403, {"error": "forbidden"}) return conn = self._db() try: rows = conn.execute( "SELECT * FROM tasks WHERE project_id = ?" " ORDER BY created_at, id", (project_id,), ).fetchall() finally: conn.close() self._send_json(200, {"tasks": [_task_to_dict(r) for r in rows]}) def _fetch_task(self, task_id): conn = self._db() try: return conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,) ).fetchone() finally: conn.close() def _get_task(self, user_id, task_id): row = self._fetch_task(task_id) if row is None: self._send_json(404, {"error": "task not found"}) return task = _task_to_dict(row) with _auth_store(self.server.db_path) as store: if not store.check_permission(user_id, task["project_id"], "read"): self._send_json(403, {"error": "forbidden"}) return self._send_json(200, task) def _cancel_task(self, user_id, task_id): row = self._fetch_task(task_id) if row is None: self._send_json(404, {"error": "task not found"}) return task = _task_to_dict(row) with _auth_store(self.server.db_path) as store: if not store.check_permission( user_id, task["project_id"], "create_task"): self._send_json(403, {"error": "forbidden"}) return if task["status"] in TERMINAL_STATUSES: self._send_json(400, {"error": "task already terminal"}) return now = time.time() conn = self._db() try: conn.execute( "UPDATE tasks SET status = 'interrupted', updated_at = ?" " WHERE id = ?", (now, task_id), ) conn.commit() task = _task_to_dict( conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,) ).fetchone() ) finally: conn.close() with _auth_store(self.server.db_path) as store: store.audit(user_id, task["project_id"], "cancel_task", task_id) self._send_json(200, task) def _merge_task(self, user_id, task_id): """把云端任务标记为已合并：只补状态，不执行合并本身。 本接口不做 git 合并。自动 PR 的正确顺序是：任务 completed -> 调本接口改状态 -> 调 POST /tasks/{id}/pr 推分支建 PR -> 再在 GitHub 上合 PR 或本地 host.py merge。 本地 host.py merge 成功后会删分支和 worktree，此后调 PR 会跳过 no_worktree，所以 PR 必须在本地 git merge 之前做，而不是之后。规则：任务必须存在（404）、调用方 必须是项目 maintainer 以上（approve_merge，403）、任务 必须是 completed（其它状态一律 409，含已 merged 的重复 调用，避免二次合并混淆）。成功后写审计 merge_task。 """ row = self._fetch_task(task_id) if row is None: self._send_json(404, {"error": "task not found"}) return task = _task_to_dict(row) with _auth_store(self.server.db_path) as store: if not store.check_permission( user_id, task["project_id"], "approve_merge"): self._send_json(403, {"error": "forbidden"}) return if task["status"] != "completed": self._send_json(409, { "error": "only completed tasks can be marked merged", "status": task["status"], }) return now = time.time() conn = self._db() try: conn.execute( "UPDATE tasks SET status = 'merged', updated_at = ?" " WHERE id = ?", (now, task_id), ) conn.commit() task = _task_to_dict( conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,) ).fetchone() ) finally: conn.close() with _auth_store(self.server.db_path) as store: store.audit(user_id, task["project_id"], "merge_task", task_id) self._send_json(200, task) def _retry_task(self, user_id, task_id): """把终态失败任务重新排队：只改状态，不直接执行。 规则：任务必须存在（404）、调用方必须是项目 maintainer 以上（approve_merge，403——重跑要再花模型钱，不能让 只有读权限或普通贡献者随便点）、任务必须是 failed / interrupted / merge-failed（其它状态一律 409：running 本来就在跑，queued 本来就在等，completed/merged 已有 结果，重跑应是新任务而不是覆盖结果）。 成功后状态回 queued，上一轮的 result_text/error_text/ actual_model 清空（否则新一轮的结果会和旧结果混在一起 无法分辨）；prompt、模型、优先级、依赖原样保留，依赖 已完成时重跑任务会立刻就绪。写审计 task_retried。 更新用 CAS（WHERE status IN (...)），避免与正在完成的 回合互相覆盖：若 CAS 没改到行，说明状态已变，返回 409 而不是假装成功。 """ row = self._fetch_task(task_id) if row is None: self._send_json(404, {"error": "task not found"}) return task = _task_to_dict(row) with _auth_store(self.server.db_path) as store: if not store.check_permission( user_id, task["project_id"], "approve_merge"): self._send_json(403, {"error": "forbidden"}) return if task["status"] not in ("failed", "interrupted", "merge-failed"): self._send_json(409, { "error": "only failed/interrupted tasks can be retried", "status": task["status"], }) return now = time.time() conn = self._db() try: cur = conn.execute( "UPDATE tasks SET status = 'queued', result_text = NULL," " error_text = NULL, actual_model = NULL, updated_at = ?" " WHERE id = ? AND status IN" " ('failed', 'interrupted', 'merge-failed')", (now, task_id), ) conn.commit() if cur.rowcount != 1: self._send_json(409, { "error": "task status changed, retry aborted", "status": task["status"], }) return task = _task_to_dict( conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,) ).fetchone() ) finally: conn.close() with _auth_store(self.server.db_path) as store: store.audit(user_id, task["project_id"], "task_retried", task_id) self._send_json(200, task) # -- worker 池 ----------------------------------------------------- def _workers_list(self, user_id): """列出注册 worker：任何已认证用户可读（worker 是基础设施 信息，不属于某个项目的数据，不按项目成员过滤）。""" workers = _registry_list(self.server.db_path) self._send_json(200, {"workers": workers}) def _workers_heartbeat(self, user_id): """worker 心跳：body 含 worker_id（必填，缺了 400）与可选 executor/capacity。心跳本身就是注册：worker_id 不存在时 自动注册，绝不 404。""" body, err = self._read_json() if err: self._send_json(400, {"error": err}) return worker_id = body.get("worker_id") if not worker_id or not isinstance(worker_id, str): self._send_json(400, {"error": "worker_id required"}) return executor = body.get("executor") capacity = body.get("capacity") if capacity is not None: try: capacity = int(capacity) except (TypeError, ValueError): self._send_json(400, {"error": "capacity must be an integer"}) return try: worker = _registry_heartbeat( self.server.db_path, worker_id, executor=executor, capacity=capacity) except ValueError as exc: self._send_json(400, {"error": str(exc)}) return worker["alive"] = True self._send_json(200, worker) # -- 自动 PR（默认关闭，显式触发）---------------------------------- def _task_pr(self, user_id, task_id): """给已 merged 的任务建 PR（见 cloud/pr.py 的跳过规则）。 权限：项目 maintainer 以上（approve_merge，403）；任务 不存在 404；非 merged 状态 409；跳过时返回 200 + pr_skipped 与 reason，任务状态保持 merged 不变；push/pr 创建失败时返回 502 并写清哪一步失败，任务同样保持 merged。 成功时把 pr_url/pr_number 写回任务并记审计 pr_created。 """ row = self._fetch_task(task_id) if row is None: self._send_json(404, {"error": "task not found"}) return task = _task_to_dict(row) with _auth_store(self.server.db_path) as store: if not store.check_permission( user_id, task["project_id"], "approve_merge"): self._send_json(403, {"error": "forbidden"}) return if task["status"] != "merged": self._send_json(409, { "error": "only merged tasks can create a PR", "status": task["status"], }) return conn = self._db() try: prow = conn.execute( "SELECT auto_pr FROM projects WHERE id = ?", (task["project_id"],)).fetchone() finally: conn.close() auto_pr = 0 if prow is not None: try: auto_pr = prow["auto_pr"] or 0 except (KeyError, IndexError, TypeError): auto_pr = 0 try: result = create_task_pr( task_id, auto_pr, worktrees_root=getattr( self.server, "host_worktrees_dir", None), git_runner=getattr(self.server, "pr_git_runner", None), gh_runner=getattr(self.server, "pr_gh_runner", None), gh_which=getattr(self.server, "pr_gh_which", None)) except PrSkipped as exc: self._send_json(200, { "task_id": task_id, "status": task["status"], "pr_skipped": True, "reason": exc.reason, }) return except PrStepError as exc: self._send_json(502, { "error": exc.message, "step": exc.step, "task_id": task_id, "status": task["status"], }) return now = time.time() conn = self._db() try: conn.execute( "UPDATE tasks SET pr_url = ?, pr_number = ?," " updated_at = ? WHERE id = ?", (result["pr_url"], result["pr_number"], now, task_id), ) conn.commit() task = _task_to_dict( conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()) finally: conn.close() with _auth_store(self.server.db_path) as store: store.audit(user_id, task["project_id"], "pr_created", result["pr_url"]) self._send_json(200, dict(task, pr_url=result["pr_url"], pr_number=result["pr_number"], branch=result["branch"])) def _heartbeat_task(self, user_id, task_id): row = self._fetch_task(task_id) if row is None: self._send_json(404, {"error": "task not found"}) return task = _task_to_dict(row) with _auth_store(self.server.db_path) as store: if not store.check_permission( user_id, task["project_id"], "create_task"): self._send_json(403, {"error": "forbidden"}) return now = time.time() new_status = "running" if task["status"] == "queued" else task["status"] conn = self._db() try: conn.execute( "UPDATE tasks SET status = ?, last_heartbeat = ?," " updated_at = ? WHERE id = ?", (new_status, now, now, task_id), ) conn.commit() task = _task_to_dict( conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,) ).fetchone() ) finally: conn.close() self._send_json(200, task) # -- 管理 ---------------------------------------------------------- def _admin_sweep(self): body, err = self._read_json() if err: self._send_json(400, {"error": err}) return threshold = body.get( "threshold_seconds", DEFAULT_SWEEP_THRESHOLD_SECONDS) try: threshold = int(threshold) except (TypeError, ValueError): self._send_json(400, {"error": "threshold_seconds must be an integer"}) return 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、成本合计。 权限：任意项目成员即可读（不要求 owner），但只统计调用者 是成员的项目——不是全库数字，避免跨项目信息泄露。 成本真数与估算分开：cost_records 的 source 列区分。 """ conn = self._db() try: member_ids = { row["project_id"] for row in conn.execute( "SELECT project_id FROM project_members WHERE user_id = ?", (user_id,), ).fetchall() } rows = conn.execute( "SELECT project_id, status, started_at, finished_at" " FROM tasks" ).fetchall() # 按状态分计数：七个状态键恒存在（缺失为 0），调用方 # 不用再猜某个状态是“零”还是“没统计”。 by_status = {key: 0 for key in TASK_STATUS_KEYS} durations = [] for row in rows: if row["project_id"] not in member_ids: continue by_status[row["status"]] = by_status.get(row["status"], 0) + 1 if row["started_at"] and row["finished_at"]: durations.append(row["finished_at"] - row["started_at"]) total = sum(by_status.values()) done = by_status.get("completed", 0) + by_status.get("merged", 0) failed = by_status.get("failed", 0) + by_status.get("merge-failed", 0) finished = done + failed try: cost_rows = conn.execute( "SELECT project_id, source, total_tokens, usd_micros," " fallback_count FROM cost_records" ).fetchall() except (sqlite3.OperationalError, _PgOperationalError): cost_rows = [] actual_tokens = estimated_tokens = 0 actual_usd = estimated_usd = 0 fallback_total = 0 for row in cost_rows: if row["project_id"] not in member_ids: continue if row["source"] == "actual": actual_tokens += row["total_tokens"] or 0 actual_usd += row["usd_micros"] or 0 else: estimated_tokens += row["total_tokens"] or 0 estimated_usd += row["usd_micros"] or 0 fallback_total += row["fallback_count"] or 0 finally: conn.close() self._send_json(200, { "total_tasks": total, "by_status": by_status, "success_rate": (done / finished) if finished else None, "avg_duration_seconds": ( sum(durations) / len(durations)) if durations else None, "fallback_total": fallback_total, "cost": { "actual_tokens": actual_tokens, "estimated_tokens": estimated_tokens, "actual_usd_micros": actual_usd, "estimated_usd_micros": estimated_usd, }, }) # -- OAuth（pre-auth，靠 state + secret，不走 X-API-Key）----------- def _oauth_provider(self, path): parts = [p for p in path.split("/") if p] if len(parts) != 3 or parts[0] != "oauth": return None return parts[1] def _oauth_login(self, path): provider = self._oauth_provider(path) if provider is None: self._send_json(404, {"error": "not found"}) return query = parse_qs(urlparse(self.path).query) client_id = (query.get("client_id") or [None])[0] redirect_uri = (query.get("redirect_uri") or [None])[0] scope = (query.get("scope") or [None])[0] if not client_id or not redirect_uri: self._send_json(400, {"error": "client_id and redirect_uri required"}) return store = OAuthStateStore(self.server.db_path) try: state, url = start_oauth_login( store, provider, client_id, redirect_uri, scope=scope) except OAuthError as e: self._send_json(400, {"error": str(e), "reason": e.reason}) return except ValueError as e: self._send_json(400, {"error": str(e)}) return finally: try: store.conn.close() except Exception: pass # 只返回 URL，不直接跳转到任何登录页；跳转由调用方决定。 self._send_json(200, {"provider": provider, "state": state, "authorization_url": url}) def _oauth_clients(self, provider): """回调用的 http/userinfo 客户端：测试注入优先，生产走真实路径。 server 属性 oauth_http_client / oauth_userinfo_client 由 create_server 置为 None（生产）或由测试注入 fake；http_client 已注入（fake）而 userinfo 未注入时不走真实 userinfo（fake token 自带身份字段时不需要它，走真实端点反而会用假 token 撞真实 GitHub）。生产（两者都未注入）时 userinfo 走真实端点。 """ http_client = getattr(self.server, "oauth_http_client", None) userinfo_client = getattr(self.server, "oauth_userinfo_client", None) if userinfo_client is None and http_client is None: userinfo_client = default_userinfo_client(provider) return http_client, userinfo_client def _oauth_finish(self, provider, code, state, client_id, redirect_uri): """回调共用后半段：验 state -> 换 token -> 取身份 -> 分叉。 成功返回 6 元组 ``(local_id, created, api_key_plain, rotated, identity, is_bind)``； 失败抛 OAuthError（调用方按 JSON/HTML 分别呈现，绝不把 access_token、client_secret、旧 API key 写进任何响应）， 绑定冲突抛 BindError（调用方按 409 呈现）。 client_id / redirect_uri 为 None 时由 handle_oauth_callback 从 state 记录取回（真实浏览器回调只带 code + state）。 分叉规则（顺序即优先级）： 1. state 带 bind_user_id -> 绑定分支：把身份绑到该本站 账号，返回 is_bind=True、api_key_plain=None（绑定成功页 绝不显示任何 key）；冲突抛 BindError； 2. 身份已绑定到某本站账号 -> 登录落到被绑账号，并按 OAuth 登录规矩轮换 key（created=False、rotated=True）， 与 oauth 账号登录同一语义，不发明新语义； 3. 否则走原来的确定性 oauth 账号映射（新建/复用 + 轮换）。 """ store = OAuthStateStore(self.server.db_path) try: http_client, userinfo_client = self._oauth_clients(provider) identity = handle_oauth_callback( store, provider, code, state, client_id, redirect_uri, http_client=http_client, userinfo_client=userinfo_client) finally: try: store.conn.close() except Exception: pass with _auth_store(self.server.db_path) as auth: bind_user_id = identity.get("bind_user_id") if bind_user_id: binding = bind_external_identity( auth, provider, identity, bind_user_id) auth.audit(bind_user_id, "", "bind_external", "%s:%s" % (provider, identity.get("external_id"))) identity["already_bound"] = binding["already"] return (bind_user_id, False, None, False, identity, True) bound_user_id = find_bound_user(auth, provider, identity) if bound_user_id is not None: api_key_plain = auth.rotate_api_key(bound_user_id) auth.audit(bound_user_id, "", "oauth_login", "%s:%s" % (provider, identity.get("external_id"))) auth.audit(bound_user_id, "", "api_key_rotated", provider) return (bound_user_id, False, api_key_plain, True, identity, False) local_id, created, api_key_plain, rotated = \ map_external_identity_to_local_user_with_key( auth, provider, identity) auth.audit(local_id, "", "oauth_login", "%s:%s" % (provider, identity.get("external_id"))) if rotated: auth.audit(local_id, "", "api_key_rotated", provider) return local_id, created, api_key_plain, rotated, identity, False def _oauth_callback(self, path): provider = self._oauth_provider(path) if provider is None: self._send_json(404, {"error": "not found"}) return body, err = self._read_json() if err: self._send_json(400, {"error": err}) return code = body.get("code") state = body.get("state") client_id = body.get("client_id") redirect_uri = body.get("redirect_uri") if not code or not state: self._send_json(400, {"error": "code and state required"}) return try: local_id, created, api_key_plain, rotated, identity, is_bind = \ self._oauth_finish(provider, code, state, client_id, redirect_uri) except BindError as e: self._send_json(409, {"error": str(e), "reason": e.reason}) return except OAuthError as e: self._send_json(400, {"error": str(e), "reason": e.reason}) return if is_bind: # 绑定成功：绝不返回任何 key（绑定页的 completer 未必是 # 发起人，key 一旦上页就可能落到别人手里）；发起人用 # 自己的 key 调 GET /oauth/{provider}/bindings 核对。 self._send_json(200, {"linked": True, "provider": provider, "login": identity.get("login"), "external_id": identity.get("external_id"), "user_id": local_id, "already_bound": identity.get("already_bound", False)}) return # api_key 明文只在这里出现一次：调用方须立即保存， # 之后任何接口都不再返回它；轮换时旧 key 已失效。 self._send_json(200, {"user_id": local_id, "created": created, "provider": provider, "login": identity.get("login"), "email": identity.get("email"), "api_key": api_key_plain, "api_key_rotated": rotated}) def _oauth_error_page(self, reason, message): """GET 回调的 400 错误页：只含 reason 与提示，不含任何密钥、 token 或旧 API key（message 来自 OAuthError，本身不含密钥， 仍做 HTML 转义）。""" return ( "" "OAuth login failed" "

OAuth login failed

" "

reason: %s

%s

" "

Please restart login from the login page.

" "" % (html.escape(str(reason or "error")), html.escape(str(message)))) def _oauth_success_page(self, provider, local_id, created, login, api_key_plain, rotated): """GET 回调的成功页：只显示 login、user_id、created 与新 API key 一次；绝不显示 access_token、client_secret、旧 API key。""" rotation_note = ( "

Your previous API key has been rotated and no longer works.

" if rotated else "") return ( "" "OAuth login succeeded" "

OAuth login succeeded

" "

provider: %s

" "

login: %s

" "

user_id: %s

" "

new user: %s

" "%s" "

Your API key (shown only once, save it now):

" "

%s

" "

Please save it immediately. It will not be shown again " "after you leave or refresh this page." "%s

" "

Use it as the X-API-Key header for API calls.

" "" % (html.escape(str(provider)), html.escape(str(login)), html.escape(str(local_id)), "yes" if created else "no", rotation_note, html.escape(str(api_key_plain)), " Your old key has expired (rotated)." if rotated else "")) def _oauth_bind_success_page(self, provider, login): """绑定成功页：只显示 provider + 外部 login + 成功提示。 绝不显示任何 key，也不显示本站 user_id：回调页的 completer 未必是绑定发起人，key 与内部 id 一旦上页就可能落到别人 手里；发起人用自己的 key 调 GET /oauth/{provider}/bindings 核对绑定。 """ return ( "" "OAuth binding succeeded" "

OAuth binding succeeded

" "

provider: %s

" "

login: %s

" "

Your external identity is now bound to your account.

" "

You can verify it from your bindings list.

" "" % (html.escape(str(provider)), html.escape(str(login)))) def _oauth_callback_get(self, path): provider = self._oauth_provider(path) if provider is None: self._send_html(404, self._oauth_error_page( "not_found", "unknown oauth path")) return query = parse_qs(urlparse(self.path).query) code = (query.get("code") or [None])[0] state = (query.get("state") or [None])[0] client_id = (query.get("client_id") or [None])[0] redirect_uri = (query.get("redirect_uri") or [None])[0] if not code or not state: self._send_html(400, self._oauth_error_page( "missing_params", "code and state required")) return try: local_id, created, api_key_plain, rotated, identity, is_bind = \ self._oauth_finish(provider, code, state, client_id, redirect_uri) except BindError as e: self._send_html(409, self._oauth_error_page( e.reason or "bind_conflict", str(e))) return except OAuthError as e: self._send_html(400, self._oauth_error_page( e.reason or "oauth_failed", str(e))) return if is_bind: self._send_html(200, self._oauth_bind_success_page( provider, identity.get("login"))) return self._send_html(200, self._oauth_success_page( provider, local_id, created, identity.get("login"), api_key_plain, rotated)) # -- OAuth 绑定（已有账号绑外部身份）------------------------------ def _oauth_bind_route(self, path): """绑定相关路径解析：返回 (provider, action) 或 None。 POST /oauth/{provider}/bind/initiate -> (provider, "bind_initiate") GET|DELETE /oauth/{provider}/bindings -> (provider, "bindings") 其它 /oauth/ 路径返回 None，走原来的登录/回调分发。 """ parts = [p for p in path.split("/") if p] if len(parts) == 4 and parts[0] == "oauth" \ and parts[2] == "bind" and parts[3] == "initiate": return parts[1], "bind_initiate" if len(parts) == 3 and parts[0] == "oauth" \ and parts[2] == "bindings": return parts[1], "bindings" return None def _oauth_bind_initiate(self, provider): """绑定起点：须持本站 key（与登录不同，绑定必须知道绑到谁）。 body 须含 client_id + redirect_uri（与登录同款）；state 里 带 bind_user_id，回调时靠它走绑定分支。只返回授权 URL， 不直接跳转；跳转由调用方决定。 """ user_id = self._authenticate() if user_id is None: self._send_json(401, {"error": "unauthorized"}) return body, err = self._read_json() if err: self._send_json(400, {"error": err}) return client_id = body.get("client_id") redirect_uri = body.get("redirect_uri") scope = body.get("scope") if not client_id or not redirect_uri: self._send_json(400, {"error": "client_id and redirect_uri required"}) return store = OAuthStateStore(self.server.db_path) try: state = store.create_state( provider, redirect_uri, client_id=client_id, bind_user_id=user_id) url = build_authorization_url( provider, client_id, redirect_uri, scope=scope, state=state) except OAuthError as e: self._send_json(400, {"error": str(e), "reason": e.reason}) return except ValueError as e: self._send_json(400, {"error": str(e)}) return finally: try: store.conn.close() except Exception: pass self._send_json(200, {"provider": provider, "state": state, "authorization_url": url}) def _oauth_bindings_get(self, provider): """列出调用方自己的绑定（只看得到自己的，别人的看不到）。""" user_id = self._authenticate() if user_id is None: self._send_json(401, {"error": "unauthorized"}) return with _auth_store(self.server.db_path) as auth: bindings = list_bindings(auth, user_id) self._send_json(200, {"provider": provider, "bindings": bindings}) def _oauth_bindings_delete(self, provider): """解绑调用方自己在该 provider 上的绑定：本来就没有返回 404。 解绑后该身份登录恢复原来的 oauth 账号映射（旧 oauth 账号 若存在，key 仍可用，不会被解绑删掉）。 """ user_id = self._authenticate() if user_id is None: self._send_json(401, {"error": "unauthorized"}) return with _auth_store(self.server.db_path) as auth: ok = unbind_external_identity(auth, provider, user_id) if ok: auth.audit(user_id, "", "unbind_external", provider) if not ok: self._send_json(404, {"error": "no binding for provider"}) return self._send_json(200, {"unbound": True, "provider": provider}) # -- GitHub App 一键安装 ------------------------------------------ def _github_app_public_base(self): base = os.environ.get("GITHUB_APP_PUBLIC_BASE") or "" return base.rstrip("/") or None def _github_app_error_page(self, reason, message): """App 流程错误页：只含 reason 与提示，不含任何 secret、 pem、code、state（message 来自流程错误，本身不含密钥， 仍做 HTML 转义）。""" return ( "" "GitHub App setup failed" "

GitHub App setup failed

" "

reason: %s

%s

" "" % (html.escape(str(reason or "error")), html.escape(str(message)))) def _github_app_setup_state(self): """签发 setup state：须持本站 key（只有操作方能发起建 App）。 要求环境变量 GITHUB_APP_PUBLIC_BASE（公网域名，如 https://api.diante.me）：manifest 里的 hook / 回调地址 都由它拼出。返回 {state, setup_url}，setup_url 是用户 在浏览器里打开的表单页；state 一次性、1 小时过期。 """ user_id = self._authenticate() if user_id is None: self._send_json(401, {"error": "unauthorized"}) return base = self._github_app_public_base() if not base: self._send_json(400, { "error": "GITHUB_APP_PUBLIC_BASE not configured"}) return body, err = self._read_json() if err: body = {} org = (body or {}).get("org") store = SetupStateStore(self.server.db_path) try: state = store.create_state() finally: try: store.conn.close() except Exception: pass setup_url = base + "/github-app/setup?state=" + state if org: setup_url += "&org=" + quote(str(org)) self._send_json(200, {"state": state, "setup_url": setup_url}) def _github_app_setup_page(self): """建 App 表单页：公开页，但 state 必须是操作方签发的。 无 state 或 state 未签发时不渲染表单（state 是回调的 唯一凭证，公开页自己现发 state 等于谁都能覆盖 App 配置）。 表单按 GitHub manifest 文档 POST 到 github.com 的建 App 页（个人账号 /settings/apps/new，组织账号 /organizations/{org}/settings/apps/new），manifest JSON 由 JS 填进隐藏字段；用户在 GitHub 页确认 Create 后跳回 本站 /github-app/callback。 """ query = parse_qs(urlparse(self.path).query) state = (query.get("state") or [None])[0] org = (query.get("org") or [None])[0] if not state: self._send_html(400, self._github_app_error_page( "missing_state", "setup state required: ask the operator to issue one")) return base = self._github_app_public_base() if not base: self._send_html(400, self._github_app_error_page( "not_configured", "GITHUB_APP_PUBLIC_BASE not configured")) return manifest = build_app_manifest( base, app_name=os.environ.get("GITHUB_APP_NAME") or DEFAULT_APP_NAME, app_url=os.environ.get("GITHUB_APP_URL") or DEFAULT_APP_URL, public=os.environ.get("GITHUB_APP_PUBLIC") == "1") manifest_json = json.dumps(manifest).replace("<", "\\u003c") if org: action = ("https://github.com/organizations/%s" "/settings/apps/new?state=%s" % (quote(str(org)), quote(state))) else: action = ("https://github.com/settings/apps/new?state=%s" % quote(state)) perms = ", ".join("%s: %s" % kv for kv in manifest["default_permissions"].items()) events = ", ".join(manifest["default_events"]) page = ( "" "Create GitHub App" "

Create GitHub App

" "

App name: %s

" "

Webhook URL: %s

" "

Permissions: %s

" "

Events: %s

" "

Click Submit to create the app on GitHub. GitHub will " "redirect back here to finish setup.

" "
" "" "" "
" "" "" % (html.escape(manifest["name"]), html.escape(manifest["hook_attributes"]["url"]), html.escape(perms), html.escape(events), html.escape(action), manifest_json)) self._send_html(200, page) def _github_app_callback(self): """建 App 回调：验 setup state -> 换 code -> 存配置。 state 一次性、1 小时过期；换 code 失败或响应缺字段时 报错误页，不存任何东西。成功页只显示 App 身份（slug、 id、主页）与安装链接，绝不显示 webhook_secret 与 pem （两者只在库里，server 进程内用）。 """ query = parse_qs(urlparse(self.path).query) code = (query.get("code") or [None])[0] state = (query.get("state") or [None])[0] if not code or not state: self._send_html(400, self._github_app_error_page( "missing_params", "code and state required")) return store = SetupStateStore(self.server.db_path) try: ok = store.consume_state(state) finally: try: store.conn.close() except Exception: pass if not ok: self._send_html(400, self._github_app_error_page( "invalid_state", "setup state invalid or expired")) return http_client = getattr(self.server, "github_app_exchange_client", None) try: config = exchange_manifest_code(code, http_client=http_client) except AppError as e: self._send_html(400, self._github_app_error_page( e.reason or "exchange_failed", str(e))) return conn = self._db() try: save_app_config(conn, config) finally: conn.close() slug = config.get("slug") or "" page = ( "" "GitHub App created" "

GitHub App created

" "

App: %s

" "

App ID: %s

" "

Install it on your repositories:

" "

" "Install this GitHub App

" "

After installing, webhooks from the app are verified " "automatically. API actions (push, PR) still use the " "existing PAT until the App-token phase; you may remove " "the manual webhook once the app is installed.

" "" % (html.escape(slug), html.escape(str(config.get("app_id"))), html.escape(slug))) self._send_html(200, page) def _github_app_status(self): """App 配置与安装状态：须持本站 key；绝不返回 secret/pem。""" user_id = self._authenticate() if user_id is None: self._send_json(401, {"error": "unauthorized"}) return conn = self._db() try: status = public_app_status(conn) status["installations"] = list_installations(conn) finally: conn.close() self._send_json(200, status) # -- 推送订阅与分发 --------------------------------------------------- def _push_create_endpoint(self, user_id, project_id): body, err = self._read_json() if err: self._send_json(400, {"error": err}) return try: sub = _push_create( self.server.db_path, project_id, user_id, body.get("channel"), body.get("target"), body.get("preference") or "instant") except ValueError as e: self._send_json(400, {"error": str(e)}) return self._send_json(201, sub) def _push_dispatch_endpoint(self, user_id): body, err = self._read_json() if err: self._send_json(400, {"error": err}) return event_type = body.get("event_type") project_id = body.get("project_id") if not event_type or not project_id: self._send_json(400, {"error": "event_type and project_id required"}) return # 分发者须是该项目的成员（至少可读），否则不能往别人项目里推。 with _auth_store(self.server.db_path) as auth: if not auth.check_permission(user_id, project_id, "read"): self._send_json(403, {"error": "forbidden"}) return # HTTP 层无可注入 sender：webhook/slack 会报“无 sender”， # webpush/apns/fcm 会报缺凭证——都不静默成功。真实发送端 # 由 worker 注入 sender 后调 push.dispatch_event。 result = _push_dispatch( self.server.db_path, event_type, project_id, body.get("payload") or {}, sender=None) self._send_json(200, result) # -- webhooks（签名认证，不走 X-API-Key）--------------------------- def _handle_github_webhook(self): """GitHub webhook：验签 -> delivery 去重 -> 解析 -> 执行建任务。 顺序是故意的：secret 未配 503、签名无效 401（两者都在 去重之前，避免未验签的重复送达污染去重表）；同一 X-GitHub-Delivery 第二次送达直接回 200 duplicate，不 重复建任务、不重复记审计。验签接受两个 secret：App webhook secret（库里，App 装好后）与 env 里 GITHUB_WEBHOOK_SECRET（手动 webhook，过渡期共存；App 装好、手动 webhook 删掉后自然不再被命中）。 /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 主动拉取）。installation 事件记安装记录 （created/added 增行、removed 减行、deleted 清安装）， 不建任务。手动 webhook 与 App webhook 共存时同一评论 会送两遍（delivery_id 不同）：/fix-review 另有内容去重 （repo + PR + 评论 id），只建一次任务。 """ raw = self._read_body() app_secret = None conn = self._db() try: config = get_app_config(conn) if config is not None: app_secret = config.get("webhook_secret") finally: conn.close() secrets = [s for s in (app_secret, os.environ.get("GITHUB_WEBHOOK_SECRET")) if s] if not secrets: self._send_json(503, {"error": "github webhook secret not configured"}) return signature = self.headers.get("X-Hub-Signature-256") if not verify_github_signature_any(secrets, raw, signature): self._send_json(401, {"error": "unauthorized"}) return try: payload = json.loads(raw.decode("utf-8")) if raw else {} except (ValueError, UnicodeDecodeError): self._send_json(400, {"error": "invalid json"}) return if not isinstance(payload, dict): self._send_json(400, {"error": "body must be a JSON object"}) return 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 # installation 事件：记安装记录，不建任务。 if event == "installation": inst = parse_installation_event(payload) if inst is None: self._send_json(200, {"result": None}) return conn = self._db() try: action = inst["action"] if action == "created": record_installation_repos( conn, inst["installation_id"], inst["account_login"], inst["repos"]) elif action == "added": record_installation_repos( conn, inst["installation_id"], inst["account_login"], inst["repos"]) elif action == "removed": remove_installation_repos( conn, inst["installation_id"], inst["repos"]) elif action == "deleted": remove_installation(conn, inst["installation_id"]) # new_permissions_accepted：权限已接受，无需改行。 finally: conn.close() with _auth_store(self.server.db_path) as auth: auth.audit("system", "", "github_app_installation", "%s:%s" % (inst["installation_id"], inst["action"])) self._send_json(200, { "result": "installation_recorded", "installation_id": inst["installation_id"], "action": inst["action"], "repos": inst["repos"], }) return # 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}) 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] # 内容去重：手动 webhook 与 App webhook 共存时同一评论 # 送两遍（delivery_id 不同），只建一次任务；重复送达 # 回读当初的 task_id，不重复记审计。 comment_id = (payload.get("comment") or {}).get("id") is_new, existing_task_id = claim_fix_review( conn, repo_name, result.get("pr_number"), comment_id) if not is_new: self._send_json(200, dict( result, task_id=existing_task_id, project_id=project_id, duplicate=True)) return 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 set_fix_review_task_id( conn, repo_name, result.get("pr_number"), comment_id, task_id) 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() secret = os.environ.get("SLACK_SIGNING_SECRET") if not secret: self._send_json(503, {"error": "slack webhook secret not configured"}) return signature = self.headers.get("X-Slack-Signature") timestamp = self.headers.get("X-Slack-Request-Timestamp") if not verify_slack_signature(secret, timestamp, raw, signature): self._send_json(401, {"error": "unauthorized"}) return # 只解析不执行任务：Slack 斜杠命令的执行由控制面后续决定。 content_type = self.headers.get("Content-Type") or "" text = "" if "application/x-www-form-urlencoded" in content_type: parsed = parse_qs(raw.decode("utf-8", "replace")) values = parsed.get("text") text = values[0] if values else "" elif raw: try: payload = json.loads(raw.decode("utf-8")) if isinstance(payload, dict): text = payload.get("text", "") or "" else: text = raw.decode("utf-8", "replace") except (ValueError, UnicodeDecodeError): text = raw.decode("utf-8", "replace") self._send_json(200, parse_slack_command(text)) def create_server(db_path, host="127.0.0.1", port=8765, backend=None): """建 server 实例（不自动 serve，供测试与 __main__ 共用）。 backend 为 None 时读环境变量 CLOUD_DB_BACKEND，未设则默认 sqlite（旧行为不变）。选 postgres 时在建表之前就调用工厂： 驱动缺失则初始化即抛明确错误（文案含 psycopg），启动即失败， 绝不在运行时中途才炸、也绝不静默回落到 SQLite；驱动已装 且能连上服务器时 Postgres 即为可用后端（适配层见 cloud/db.py，部署步骤见 cloud/README.md）。 """ db_path = str(db_path) # 后端校验必须在 init_db 之前：postgres 模式不能先建 SQLite 表 # 再失败，那会假装 sqlite 已可用。 _create_database(backend=backend, db_path=db_path) init_db(db_path) # 确保 AuthStore 的三张表也存在，后续请求才能直接查 project_members。 with _auth_store(db_path): pass 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 server.oauth_userinfo_client = None # 自动 PR 注入点：生产为 None（走真实 git/gh），测试可注入 # fake git/gh runner 与 worktree 根；绝不注入真实凭证。 server.pr_git_runner = None server.pr_gh_runner = None server.pr_gh_which = None server.host_worktrees_dir = None return server def main(argv=None): parser = argparse.ArgumentParser(description="cloud control plane") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8765) parser.add_argument("--db", default="cloud-control-plane.db") parser.add_argument("--db-backend", default=None, help="sqlite（默认）或 postgres；未给则读 " "CLOUD_DB_BACKEND 环境变量") args = parser.parse_args(argv) server = create_server(args.db, args.host, args.port, backend=args.db_backend) print( "cloud-control-plane listening on %s:%d db=%s" % (args.host, server.server_address[1], args.db), flush=True, ) try: server.serve_forever() except KeyboardInterrupt: pass if __name__ == "__main__": main()