diff --git a/cloud/README.md b/cloud/README.md index 6f7a8c0..8e18bf1 100644 --- a/cloud/README.md +++ b/cloud/README.md @@ -274,4 +274,37 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳 APNs / FCM 的真实凭证与发送端仍缺；单机 SQLite 在高并发写下 仍有上限，生产量上来后需换 Postgres（接口层与骨架已就绪， 但真实部署、驱动安装与迁移脚本仍缺，见上文“数据库后端选择”）； - worker 执行、云端执行、真实通知与真实模型回合均未接通。 + worker 执行已接通（见下文“worker 常驻与真取消”），云端执行、 + 真实通知与真实模型回合仍受 key 供给与云账号卡点限制。 + +## worker 常驻与真取消（本轮新增） + +- 单次：`python3 -m cloud.worker --db <库> --host-py ../host.py`； + 常驻：`python3 -m cloud.worker --db <库> --loop --interval 300`， + 每轮 run_all，SIGTERM/SIGINT 时优雅退出，正在跑的任务先标 + interrupted 再退出；CLI 起 worker 前先 init_db，全新库不再因 + 缺表而炸，循环内一轮异常记日志后继续，不带走整个 worker。 +- 真取消：子进程以进程组启动（start_new_session），取消时先 + SIGTERM 给整个进程组、等 2 秒、仍活着再 SIGKILL，杀完任务 + 保持 interrupted，执行结果不覆盖它；取消竞态（run_one 已先 + 标 interrupted）时 cancel 仍返回 True，因为中断确由本次取消触发。 +- 任务结果列：tasks 表新增 result_text、error_text、actual_model、 + started_at、finished_at、child_pid，init_db 用 PRAGMA 检查后 + ALTER TABLE 幂等迁移，老库升级不丢数据；GET /tasks/{id} 返回 + 它们，缺 key 任务的 error_text 含 Missing environment variable。 + +## 可观测性与备份（本轮新增） + +- 结构化日志：cloud/logging_config.py，JSON 行日志（时间、级别、 + 服务、任务 id、事件），worker 的认领/完成/失败/取消/fallback/ + 缺 key 记事件日志，绝不写密钥与环境变量值。 +- 指标：`GET /metrics`（需 X-API-Key，任意项目成员可读，只统计 + 调用者是成员的项目），返回任务总数、各状态计数、成功率、 + 平均耗时、fallback 总次数、成本合计（actual 与 estimated 分开）。 +- 备份：`python3 -m cloud.backup --db <库> --out <目录> [--keep 3]`， + 用 SQLite backup API 在线备份，轮转只留最近 3 份（命名含时间戳）， + 失败明确报错不静默；备份文件本身就是可打开的 SQLite 库。 +- DB 接口迁移：server 及 auth/memory/kb/cost/oauth/push/webhooks + 的 Store 连接已统一走 cloud.db 的 WAL + busy_timeout 连接 + （Postgres 后端仍是骨架，驱动缺失时启动即失败、不先建 SQLite + 文件，这一点迁移后未退化）。 diff --git a/cloud/auth.py b/cloud/auth.py index 2646284..0943998 100644 --- a/cloud/auth.py +++ b/cloud/auth.py @@ -24,6 +24,7 @@ approve_merge -> maintainer，manage_project -> owner（仅 owner）。 import hashlib import secrets import sqlite3 +from cloud.db import connect as _db_connect import time import uuid @@ -44,7 +45,7 @@ class AuthStore: def __init__(self, db_path): self.db_path = str(db_path) - self.conn = sqlite3.connect(self.db_path) + self.conn = _db_connect(self.db_path) self.conn.execute( "CREATE TABLE IF NOT EXISTS users(" "id TEXT PRIMARY KEY, " diff --git a/cloud/backup.py b/cloud/backup.py new file mode 100644 index 0000000..a51244f --- /dev/null +++ b/cloud/backup.py @@ -0,0 +1,64 @@ +"""云库备份：SQLite 在线备份 + 轮转。 + +用法：python3 -m cloud.backup --db <库路径> --out <目录> [--keep 3] + +- 用 sqlite3 的 backup API 做在线备份（源库正在被 server 读写时 + 也能安全拷，不用先停服务、不用直接 cp 正在写的文件）。 +- 轮转：备份文件名含时间戳，写完后按时间倒序只保留最近 keep + 份（默认 3），多余的删掉；keep < 1 时报错，不删光。 +- 备份失败（源库不存在、目标目录建不出来、backup API 报错） + 明确抛错、不静默，CLI 退出码非零。 +- 备份文件本身就是 SQLite 库，可直接用 sqlite3 打开验证含 + tasks 表；备份里不含任何密钥（库里本来就只存 API key 哈希）。 +""" + +import argparse +import glob +import os +import sqlite3 +import time + +from cloud.db import connect as _db_connect + + +def backup_db(db_path, out_dir, keep=3, timestamp=None): + """执行一次备份并轮转，返回备份文件路径。 + + timestamp 可注入（单测用固定时间戳造多份）；缺省用当前时间。 + """ + if keep < 1: + raise ValueError("keep 必须 >= 1，否则会删光所有备份") + if not os.path.exists(db_path): + raise FileNotFoundError("源库不存在：%s" % db_path) + os.makedirs(out_dir, exist_ok=True) + stamp = timestamp or time.strftime("%Y%m%d%H%M%S", time.localtime()) + dest = os.path.join(out_dir, "cloud-backup-%s.db" % stamp) + src = _db_connect(str(db_path)) + try: + dst = sqlite3.connect(dest) + try: + src.backup(dst) + finally: + dst.close() + finally: + src.close() + # 轮转：按文件名时间戳倒序（文件名即时间戳，排序即时间序）， + # 只留最近 keep 份。 + backups = sorted(glob.glob(os.path.join(out_dir, "cloud-backup-*.db"))) + for old in backups[:-keep] if len(backups) > keep else []: + os.remove(old) + return dest + + +def main(argv=None): + parser = argparse.ArgumentParser(description="cloud control-plane backup") + parser.add_argument("--db", required=True) + parser.add_argument("--out", required=True) + parser.add_argument("--keep", type=int, default=3) + args = parser.parse_args(argv) + dest = backup_db(args.db, args.out, keep=args.keep) + print(dest) + + +if __name__ == "__main__": + main() diff --git a/cloud/cost.py b/cloud/cost.py index 24b100a..b052830 100644 --- a/cloud/cost.py +++ b/cloud/cost.py @@ -10,6 +10,7 @@ so estimates are never mixed into actual totals. import os import sqlite3 +from cloud.db import connect as _db_connect import time from datetime import datetime, timezone @@ -56,7 +57,7 @@ class CostStore: if db_path != ":memory:": parent = os.path.dirname(os.path.abspath(db_path)) os.makedirs(parent, exist_ok=True) - self._conn = sqlite3.connect(db_path) + self._conn = _db_connect(db_path) self._conn.execute( """CREATE TABLE IF NOT EXISTS cost_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, diff --git a/cloud/kb.py b/cloud/kb.py index a623f7f..d965bd2 100644 --- a/cloud/kb.py +++ b/cloud/kb.py @@ -17,6 +17,7 @@ Design notes: """ import sqlite3 +from cloud.db import connect as _db_connect import time _KINDS = ("file", "readme", "doc", "task", "review") @@ -27,7 +28,7 @@ class KBStore: def __init__(self, db_path): self.db_path = db_path - self.conn = sqlite3.connect(db_path) + self.conn = _db_connect(db_path) self.conn.row_factory = sqlite3.Row self.conn.execute( """CREATE TABLE IF NOT EXISTS kb_docs( diff --git a/cloud/logging_config.py b/cloud/logging_config.py new file mode 100644 index 0000000..4d233d4 --- /dev/null +++ b/cloud/logging_config.py @@ -0,0 +1,54 @@ +"""结构化日志：JSON 行日志，含时间、级别、服务、任务 id、事件。 + +只用标准库 logging。日志绝不写任何密钥、API key、环境变量值： +调用方只传任务 id、状态、事件名这类非敏感字段。 +""" + +import json +import logging +import sys +import time + + +class JsonFormatter(logging.Formatter): + """把每条记录格式化成一行 JSON，方便后续按字段过滤聚合。""" + + def format(self, record): + payload = { + "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created)), + "level": record.levelname, + "service": "cloud-control-plane", + "logger": record.name, + "message": record.getMessage(), + } + for key in ("task_id", "project_id", "event", "status"): + if hasattr(record, key): + payload[key] = getattr(record, key) + if record.exc_info and record.exc_info[0] is not None: + payload["error_type"] = record.exc_info[0].__name__ + return json.dumps(payload, ensure_ascii=False) + + +def setup_logging(level=logging.INFO, stream=None): + """配置根 logger 为 JSON 行日志；重复调用不重复加 handler。""" + handler = logging.StreamHandler(stream or sys.stderr) + handler.setFormatter(JsonFormatter()) + root = logging.getLogger() + root.handlers = [handler] + root.setLevel(level) + return root + + +def get_logger(name="cloud"): + return logging.getLogger(name) + + +def log_event(logger, event, task_id=None, project_id=None, status=None, + message=None): + """记一条结构化事件日志；message 默认就是事件名。""" + logger.info(message or event, extra={ + "event": event, + "task_id": task_id, + "project_id": project_id, + "status": status, + }) diff --git a/cloud/memory.py b/cloud/memory.py index 3763b9e..6f2074c 100644 --- a/cloud/memory.py +++ b/cloud/memory.py @@ -14,6 +14,7 @@ Design notes: import json import sqlite3 +from cloud.db import connect as _db_connect import time @@ -22,7 +23,7 @@ class MemoryStore: def __init__(self, db_path): self.db_path = db_path - self.conn = sqlite3.connect(db_path) + self.conn = _db_connect(db_path) self.conn.row_factory = sqlite3.Row self.conn.execute( """CREATE TABLE IF NOT EXISTS summaries( diff --git a/cloud/oauth.py b/cloud/oauth.py index 5c04d6d..ff40ea1 100644 --- a/cloud/oauth.py +++ b/cloud/oauth.py @@ -33,6 +33,7 @@ import json import os import secrets import sqlite3 +from cloud.db import connect as _db_connect import time import urllib.parse import urllib.request @@ -96,7 +97,7 @@ class OAuthStateStore: def __init__(self, db_path): self.db_path = str(db_path) - self.conn = sqlite3.connect(self.db_path) + self.conn = _db_connect(self.db_path) self.conn.execute( "CREATE TABLE IF NOT EXISTS oauth_states(" "state TEXT PRIMARY KEY, " diff --git a/cloud/push.py b/cloud/push.py index 2ce9d63..15697e0 100644 --- a/cloud/push.py +++ b/cloud/push.py @@ -32,6 +32,7 @@ """ import sqlite3 +from cloud.db import connect as _db_connect import time import uuid from contextlib import contextmanager @@ -78,7 +79,7 @@ def _connect(db): if _is_connection(db): yield db elif isinstance(db, (str, bytes)) or hasattr(db, "__fspath__"): - conn = sqlite3.connect(str(db), timeout=30) + conn = _db_connect(str(db)) try: yield conn finally: diff --git a/cloud/sandbox.py b/cloud/sandbox.py index 9effd04..c0113f0 100644 --- a/cloud/sandbox.py +++ b/cloud/sandbox.py @@ -167,6 +167,32 @@ class LocalExecutor(Executor): return self._result_from_runner( task["id"], self.runner(cmd)) + def popen_task(self, task, kind="dispatch"): + """进程组启动 host.py 子进程，返回 Popen（供真取消用）。 + + start_new_session=True 让子进程自成进程组，worker 取消时 + killpg 才能连同子孙一起杀掉。注意：只有默认 runner（真实 + subprocess）时才用本方法；单测注入的假 runner 走同步路径， + 不走这里。 + """ + if not isinstance(task, dict) or task.get("id") is None: + raise ValueError("任务缺少 id") + if not task.get("prompt"): + raise ValueError("任务缺少 prompt") + cmd = self._base_command() + if kind == "run": + cmd += ["run", "--task", self._host_name(task), + "--prompt", task["prompt"]] + else: + cmd += ["dispatch", "--name", self._host_name(task), + "--task", task["prompt"]] + repo = task.get("repo") or self.repo + if repo: + cmd += ["--repo", repo] + return subprocess.Popen(cmd, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, + start_new_session=True) + def execute(self, task) -> dict: task_id = task.get("id") if isinstance(task, dict) else None if task_id is None: diff --git a/cloud/server.py b/cloud/server.py index b8645bc..0dc0a00 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -87,6 +87,30 @@ def _connect(db_path): 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", + } + for name, coltype in new_cols.items(): + if name not in cols: + conn.execute( + "ALTER TABLE tasks ADD COLUMN %s %s" % (name, coltype)) + + def init_db(db_path): """创建 server 自己的 projects/tasks 表（AuthStore 的表由其自建）。""" conn = _connect(db_path) @@ -113,6 +137,7 @@ def init_db(db_path): "last_heartbeat REAL, " "UNIQUE(project_id, idempotency_key))" ) + migrate_tasks_table(conn) conn.commit() finally: conn.close() @@ -131,6 +156,14 @@ def _auth_store(db_path): 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: @@ -151,6 +184,12 @@ def _task_to_dict(row): "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"), } @@ -306,6 +345,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._create_project(user_id) elif path == "/projects" and method == "GET": self._list_projects(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/"): @@ -989,6 +1030,75 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): swept = sweep_stale_tasks(self.server.db_path, threshold) self._send_json(200, {"swept": swept}) + 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() + by_status = {} + 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: + 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] diff --git a/cloud/tests/test_prod_gaps.py b/cloud/tests/test_prod_gaps.py new file mode 100644 index 0000000..f3b4362 --- /dev/null +++ b/cloud/tests/test_prod_gaps.py @@ -0,0 +1,342 @@ +"""生产级缺口单测：worker 常驻/真取消、tasks 迁移、metrics、备份、日志。 + +只用标准库，不跑真实模型回合、不联网、不登录任何服务； +process_factory 返回的长跑子进程只是 `sleep`，取消后必须消失。 +""" + +import http.client +import io +import json +import logging +import os +import sqlite3 +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.auth import AuthStore + from cloud.backup import backup_db + from cloud.cost import CostStore + from cloud.db import PostgresDriverMissingError + from cloud.logging_config import JsonFormatter, log_event, setup_logging + from cloud.server import create_server, init_db + from cloud.worker import Worker +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.backup import backup_db + from cloud.cost import CostStore + from cloud.db import PostgresDriverMissingError + from cloud.logging_config import JsonFormatter, log_event, setup_logging + from cloud.server import create_server, init_db + from cloud.worker import Worker + + +def _insert_task(db_path, task_id, prompt="do work", status="queued", + project_id="p1", started=None, finished=None): + now = time.time() + conn = sqlite3.connect(db_path) + 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(?, ?, ?, ?, NULL, '[]', 0, 'tester', ?, ?, NULL)", + (task_id, project_id, prompt, status, now, now), + ) + if started is not None: + conn.execute( + "UPDATE tasks SET started_at = ?, finished_at = ?" + " WHERE id = ?", (started, finished, task_id)) + conn.commit() + finally: + conn.close() + + +def _row(db_path, task_id): + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + try: + row = conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone() + return dict(row) if row else None + finally: + conn.close() + + +def _completed_runner(cmd): + return SimpleNamespace( + returncode=0, + stdout=json.dumps({"task": "x", "status": "completed", + "text": "done"}), + stderr="") + + +class MigrationTestCase(unittest.TestCase): + def test_old_db_migrates_without_data_loss(self): + tmp = tempfile.TemporaryDirectory() + db_path = os.path.join(tmp.name, "old.db") + conn = sqlite3.connect(db_path) + try: + conn.execute( + "CREATE TABLE projects(id TEXT PRIMARY KEY, name TEXT," + " owner_id TEXT, created_at REAL)") + conn.execute( + "CREATE TABLE 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))") + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " depends_on, priority, created_by, created_at," + " updated_at, last_heartbeat)" + " VALUES('old1', 'p1', 'old prompt', 'completed', '[]'," + " 0, 'tester', 1, 1, NULL)") + conn.commit() + finally: + conn.close() + init_db(db_path) + cols = {r[1] for r in sqlite3.connect(db_path).execute( + "PRAGMA table_info(tasks)").fetchall()} + for col in ("result_text", "error_text", "actual_model", + "started_at", "finished_at", "child_pid"): + self.assertIn(col, cols) + row = _row(db_path, "old1") + self.assertEqual(row["prompt"], "old prompt") + self.assertEqual(row["status"], "completed") + # 再跑一次 init_db 是 no-op，不炸、不丢数据。 + init_db(db_path) + self.assertEqual(_row(db_path, "old1")["prompt"], "old prompt") + tmp.cleanup() + + +class WorkerResultColumnsTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def test_completed_task_writes_result_columns(self): + _insert_task(self.db_path, "t1") + worker = Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, + env={"XCM_API_KEY": "test-placeholder"}) + result = worker.run_one() + self.assertEqual(result["status"], "completed") + row = _row(self.db_path, "t1") + self.assertIn("done", row["result_text"]) + self.assertIsNone(row["error_text"]) + self.assertEqual(row["actual_model"], worker.model) + self.assertIsNotNone(row["started_at"]) + self.assertIsNotNone(row["finished_at"]) + self.assertGreaterEqual(row["finished_at"], row["started_at"]) + + def test_missing_key_writes_error_text_and_calls_runner_zero_times(self): + calls = [] + + def counting(cmd): + calls.append(cmd) + return _completed_runner(cmd) + + _insert_task(self.db_path, "t2") + worker = Worker(self.db_path, "/fake/host.py", + runner=counting, env={}) + result = worker.run_one() + self.assertEqual(result["status"], "failed") + self.assertEqual(calls, []) + row = _row(self.db_path, "t2") + self.assertIn("Missing environment variable", row["error_text"]) + self.assertIsNotNone(row["started_at"]) + + +class WorkerTrueCancelTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def _sleep_factory(self, task): + return subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(30)"], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + start_new_session=True) + + def test_cancel_kills_process_group_and_keeps_interrupted(self): + _insert_task(self.db_path, "long1") + worker = Worker(self.db_path, "/fake/host.py", + process_factory=self._sleep_factory, + env={"XCM_API_KEY": "test-placeholder"}) + results = [] + thread = threading.Thread( + target=lambda: results.append(worker.run_one()), daemon=True) + thread.start() + pgid = None + deadline = time.time() + 5 + while time.time() < deadline: + row = _row(self.db_path, "long1") + if row and row["child_pid"]: + pgid = row["child_pid"] + break + time.sleep(0.05) + self.assertIsNotNone(pgid, "子进程 pid 未写进任务记录") + # 进程组当前存在（killpg 0 不抛）。 + os.killpg(os.getpgid(pgid), 0) + self.assertTrue(worker.cancel("long1")) + thread.join(timeout=10) + self.assertFalse(thread.is_alive()) + self.assertEqual(results[0]["status"], "interrupted") + self.assertEqual(_row(self.db_path, "long1")["status"], "interrupted") + # 进程组确实消失。 + with self.assertRaises((ProcessLookupError, PermissionError, OSError)): + os.killpg(os.getpgid(pgid), 0) + + def test_loop_stops_and_marks_running_interrupted(self): + _insert_task(self.db_path, "loop1") + worker = Worker(self.db_path, "/fake/host.py", + process_factory=self._sleep_factory, + env={"XCM_API_KEY": "test-placeholder"}) + stop = threading.Event() + results = [] + thread = threading.Thread( + target=lambda: results.extend( + worker.run_loop(interval=0.2, stop_event=stop)), + daemon=True) + thread.start() + time.sleep(1.0) + stop.set() + # 循环在任务进行中收到停止信号：_loop_stop 由 run_loop 的 + # finally 外的轮询感知——这里直接置位模拟 SIGTERM 处理。 + worker._loop_stop = True + thread.join(timeout=10) + self.assertFalse(thread.is_alive()) + self.assertEqual(_row(self.db_path, "loop1")["status"], "interrupted") + + +class MetricsAndBackupTestCase(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.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() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + self.tmp.cleanup() + + def _req(self, method, path, key=None, body=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode() + hdrs["Content-Type"] = "application/json" + if key: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + return resp.status, json.loads(data) if data else None + + def test_metrics_counts_match_db(self): + status, proj = self._req("POST", "/projects", self.owner_key, + {"name": "m"}) + self.assertEqual(status, 201) + pid = proj["id"] + _insert_task(self.db_path, "m1", status="completed", + project_id=pid, started=1000.0, finished=1002.0) + _insert_task(self.db_path, "m2", status="failed", + project_id=pid, started=1000.0, finished=1001.0) + _insert_task(self.db_path, "other", status="completed", + project_id="someone-else") + costs = CostStore(self.db_path) + costs.record("m1", pid, "gpt-6.1-sol", input_tokens=10, + output_tokens=5, source="actual", fallback_count=1) + costs.record("m2", pid, "gpt-6.1-sol", input_tokens=20, + output_tokens=0, source="estimated") + status, metrics = self._req("GET", "/metrics", self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(metrics["total_tasks"], 2) + self.assertEqual(metrics["by_status"]["completed"], 1) + self.assertEqual(metrics["by_status"]["failed"], 1) + self.assertAlmostEqual(metrics["success_rate"], 0.5) + self.assertAlmostEqual(metrics["avg_duration_seconds"], 1.5) + self.assertEqual(metrics["fallback_total"], 1) + self.assertEqual(metrics["cost"]["actual_tokens"], 15) + self.assertEqual(metrics["cost"]["estimated_tokens"], 20) + # 无 key 401。 + status, _ = self._req("GET", "/metrics") + self.assertEqual(status, 401) + + def test_backup_rotation_keeps_three(self): + init_db(self.db_path) + _insert_task(self.db_path, "b1") + out_dir = os.path.join(self.tmp.name, "backups") + for stamp in ("20260101000000", "20260102000000", "20260103000000", + "20260104000000", "20260105000000"): + backup_db(self.db_path, out_dir, keep=3, timestamp=stamp) + files = sorted(os.listdir(out_dir)) + self.assertEqual(len(files), 3) + self.assertEqual(files[0], "cloud-backup-20260103000000.db") + conn = sqlite3.connect(os.path.join(out_dir, files[-1])) + try: + row = conn.execute( + "SELECT id FROM tasks WHERE id = 'b1'").fetchone() + self.assertIsNotNone(row) + finally: + conn.close() + + def test_postgres_backend_without_driver_fails_before_file(self): + other_db = os.path.join(self.tmp.name, "pg.db") + with self.assertRaises(PostgresDriverMissingError): + create_server(other_db, "127.0.0.1", 0, backend="postgres") + self.assertFalse(os.path.exists(other_db)) + + +class LoggingTestCase(unittest.TestCase): + def test_json_log_contains_task_and_event_without_secret(self): + stream = io.StringIO() + handler = logging.StreamHandler(stream) + handler.setFormatter(JsonFormatter()) + logger = logging.getLogger("cloud.test-prod-gaps") + logger.handlers = [handler] + logger.setLevel(logging.INFO) + logger.propagate = False + try: + log_event(logger, "worker_claim", task_id="t9", + project_id="p9", status="running") + finally: + logger.handlers = [] + payload = json.loads(stream.getvalue().strip()) + self.assertEqual(payload["task_id"], "t9") + self.assertEqual(payload["event"], "worker_claim") + self.assertEqual(payload["service"], "cloud-control-plane") + # setup_logging 本身不重复加 handler。 + root = setup_logging(stream=io.StringIO()) + self.assertEqual(len(root.handlers), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/webhooks.py b/cloud/webhooks.py index d1e08a8..10046fc 100644 --- a/cloud/webhooks.py +++ b/cloud/webhooks.py @@ -21,6 +21,8 @@ import hmac import sqlite3 import time +from cloud.db import connect as _db_connect + __all__ = [ "verify_github_signature", "parse_github_pr_comment", @@ -155,7 +157,7 @@ class SlackUserMap: def __init__(self, db_path): self.db_path = db_path - self._conn = sqlite3.connect(db_path) + self._conn = _db_connect(db_path) self._conn.execute( "CREATE TABLE IF NOT EXISTS slack_users (" "slack_id TEXT PRIMARY KEY, " diff --git a/cloud/worker.py b/cloud/worker.py index 585e16a..86afcee 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -38,13 +38,13 @@ （user_id="system"，与 server.sweep_stale_tasks 的做法一致）， 审计失败不影响任务状态（best-effort）。 -取消：host.py 的 argparse 子命令表里没有 interrupt（turn/interrupt -只存在于 host.py Server 类内部的 app-server 协议调用，未暴露为 -CLI），所以 worker 的 cancel() 走 server 的 cancel 语义：把 -queued/running 任务直接标 interrupted（已终态的任务不动）。 -进行中的 host.py 子进程不会因此被杀（中途杀进程可能把 host -state 写一半）；run_one() 在执行返回后会先查任务是否已被取消， -已被取消的一律保持 interrupted，不用执行结果覆盖它。 +取消：worker 的 cancel() 走 server 的 cancel 语义：把 +queued/running 任务直接标 interrupted（已终态的任务不动）， +且正在跑的子进程会真杀——子进程以进程组启动（start_new_session）， +取消时先 SIGTERM 给整个进程组、等 2 秒、仍活着再 SIGKILL， +杀完后任务保持 interrupted，执行结果不覆盖它。run_one() 在 +执行返回后会先查任务是否已被取消，已被取消的一律保持 +interrupted，不用执行结果覆盖它。 可注入：runner（假 CLI，见 sandbox.LocalExecutor）与 env 映射 均可注入，单测不跑真实模型回合、不联网、不登录任何服务。 @@ -54,14 +54,20 @@ import argparse import json import os import re +import signal +import subprocess +import sys import time +from types import SimpleNamespace from cloud.auth import AuthStore from cloud.cost import CostStore, estimate_tokens from cloud.db import connect as _db_connect +from cloud.logging_config import get_logger, log_event, setup_logging from cloud.models import ModelRouter, is_fallbackable, record_fallback from cloud.queue import budget_exceeded, ready_tasks from cloud.sandbox import LocalExecutor +from cloud.sandbox import _default_runner as _sandbox_default_runner # 与 server.TERMINAL_STATUSES 对齐：这些状态的任务不再被认领或取消。 TERMINAL_STATUSES = frozenset( @@ -152,7 +158,7 @@ class Worker: runner=None, env=None, env_key=None, codex_config_path=None, model=None, router=None, max_turns=None, max_cost_usd=None, executor=None, - cost_store=None): + cost_store=None, process_factory=None): self.db_path = str(db_path) self.host_py_path = host_py_path self.config_path = config_path @@ -177,6 +183,14 @@ class Worker: repo=repo) self._cost_store = cost_store self._cancel_requested = set() + # 真取消用：正在跑的子进程（Popen 对象）与任务 id。 + # process_factory 可注入：单测用它返回一个假长跑子进程， + # 真实执行时走 LocalExecutor.popen_task（进程组启动）。 + self.process_factory = process_factory + self._running_proc = None + self._running_task_id = None + self._loop_stop = False + self.logger = get_logger("cloud.worker") # -- 成本存储（惰性建表，避免只认领不执行时也建表）-------------- @property @@ -241,6 +255,99 @@ class Worker: finally: conn.close() + def _ensure_result_columns(self, conn): + """worker 侧的迁移兜底：与 server.migrate_tasks_table 同列。 + + worker 的库理论上已由 server.init_db 迁过；这里再查一次 + 是为了 worker 直连老库（测试里手工建的表）时也不炸。 + """ + try: + cols = {row[1] for row in conn.execute( + "PRAGMA table_info(tasks)").fetchall()} + except Exception: + return + for name, coltype in { + "result_text": "TEXT", "error_text": "TEXT", + "actual_model": "TEXT", "started_at": "REAL", + "finished_at": "REAL", "child_pid": "INTEGER"}.items(): + if name not in cols: + try: + conn.execute( + "ALTER TABLE tasks ADD COLUMN %s %s" + % (name, coltype)) + except Exception: + pass + + def _write_execution_result(self, task_id, result_text=None, + error_text=None, actual_model=None, + started_at=None, finished_at=None, + child_pid=None): + """把执行结果写进 tasks 新列；老库缺列时先迁移再写。""" + conn = _db_connect(self.db_path) + try: + self._ensure_result_columns(conn) + if child_pid is None: + conn.execute( + "UPDATE tasks SET result_text = ?, error_text = ?," + " actual_model = ?, started_at = ?, finished_at = ?," + " updated_at = ? WHERE id = ?", + ((result_text or "")[:4000] if result_text else None, + (error_text or "")[:4000] if error_text else None, + actual_model, started_at, finished_at, + time.time(), task_id), + ) + else: + conn.execute( + "UPDATE tasks SET result_text = ?, error_text = ?," + " actual_model = ?, started_at = ?, finished_at = ?," + " child_pid = ?, updated_at = ? WHERE id = ?", + ((result_text or "")[:4000] if result_text else None, + (error_text or "")[:4000] if error_text else None, + actual_model, started_at, finished_at, child_pid, + time.time(), task_id), + ) + conn.commit() + finally: + conn.close() + + def _set_child_pid(self, task_id, pid): + conn = _db_connect(self.db_path) + try: + self._ensure_result_columns(conn) + conn.execute( + "UPDATE tasks SET child_pid = ? WHERE id = ?", + (pid, task_id), + ) + conn.commit() + finally: + conn.close() + + def _kill_proc_group(self, proc): + """真杀：先 SIGTERM 给整个进程组，等 2 秒，仍活着再 SIGKILL。 + + proc 须是 start_new_session 启动的 Popen（自成进程组）， + 杀组才能连同它拉起的子孙一起杀掉，不只杀父进程。 + """ + try: + pgid = os.getpgid(proc.pid) + except Exception: + return + try: + os.killpg(pgid, signal.SIGTERM) + except (ProcessLookupError, PermissionError, OSError): + return + deadline = time.time() + 2.0 + while time.time() < deadline: + try: + os.killpg(pgid, 0) + except (ProcessLookupError, PermissionError, OSError): + return + time.sleep(0.05) + try: + os.killpg(pgid, signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + pass + def heartbeat(self, task_id): """刷新 running 任务的心跳；任务不在 running 时返回 False。""" now = time.time() @@ -295,11 +402,23 @@ class Worker: # -- 取消（server cancel 语义，见模块 docstring）------------------- def cancel(self, task_id): - """把 queued/running 任务标 interrupted；已终态或不存在返回 False。""" + """把 queued/running 任务标 interrupted；已终态或不存在返回 False。 + + 正在跑的子进程要真杀：先杀进程组（见 _kill_proc_group）， + 再标状态；已终态或不存在返回 False。 + """ task = self.get_task(task_id) if task is None or task["status"] in TERMINAL_STATUSES: return False self._cancel_requested.add(task_id) + if self._running_task_id == task_id and self._running_proc is not None: + self._kill_proc_group(self._running_proc) + try: + self._running_proc.wait(timeout=3) + except Exception: + pass + self._running_proc = None + self._running_task_id = None conn = _db_connect(self.db_path) try: cur = conn.execute( @@ -313,8 +432,16 @@ class Worker: claimed = cur.rowcount == 1 finally: conn.close() + if not claimed: + # 并发竞态：run_one 的轮询已看到本次取消的标记、先一步 + # 把任务标成 interrupted——中断确实由本次取消触发，返回 True。 + current = self.get_task(task_id) + if current is not None and current["status"] == "interrupted": + claimed = True if claimed: self._audit(task["project_id"], "worker_cancelled", task_id) + log_event(self.logger, "worker_cancelled", task_id=task_id, + project_id=task["project_id"], status="interrupted") return claimed def _was_cancelled(self, task_id): @@ -332,6 +459,67 @@ class Worker: return False return value is not None and value != "" + def _use_popen(self): + """是否走进程模式：注入了 process_factory，或执行器是 + 默认 runner 的 LocalExecutor（真实 subprocess）。""" + if self.process_factory is not None: + return True + return (isinstance(self.executor, LocalExecutor) + and getattr(self.executor, "runner", None) + is _sandbox_default_runner + and hasattr(self.executor, "popen_task")) + + def _execute_via_process(self, task, kind="dispatch"): + """进程模式执行一轮：启动、轮询、取消时杀进程组。 + + 返回与同步路径同款 result dict（status/stdout/stderr/ + host_status/returncode）。取消时返回 status='cancelled' + 的标记，调用方据此保持 interrupted。 + """ + task_id = task["id"] + if self.process_factory is not None: + proc = self.process_factory(task) + else: + proc = self.executor.popen_task(task, kind=kind) + self._running_proc = proc + self._running_task_id = task_id + try: + self._set_child_pid(task_id, getattr(proc, "pid", None)) + except Exception: + pass + cancelled = False + try: + while proc.poll() is None: + if task_id in self._cancel_requested or self._loop_stop: + cancelled = True + self._kill_proc_group(proc) + break + time.sleep(0.05) + try: + out, err = proc.communicate(timeout=5) + except Exception: + try: + proc.kill() + except Exception: + pass + out, err = proc.communicate() + finally: + self._running_proc = None + self._running_task_id = None + if cancelled: + return {"executor": "local", "task_id": task_id, + "status": "cancelled", "stdout": out or "", + "stderr": err or "", "returncode": proc.returncode, + "host_status": None} + fake = SimpleNamespace(returncode=proc.returncode, + stdout=out or "", stderr=err or "") + if hasattr(self.executor, "_result_from_runner"): + return self.executor._result_from_runner(task_id, fake) + return {"executor": "local", "task_id": task_id, + "status": "completed" if proc.returncode == 0 else "failed", + "stdout": out or "", "stderr": err or "", + "returncode": proc.returncode, "host_status": None} + # -- 执行 ---------------------------------------------------------- def run_one(self, task=None): """认领并执行一个任务，返回结果 dict（见模块 docstring）。 @@ -351,13 +539,21 @@ class Worker: "task_id": task_id} task = self.get_task(task_id) or task self._audit(task["project_id"], "worker_claim", task_id) + log_event(self.logger, "worker_claim", task_id=task_id, + project_id=task["project_id"], status="running") + started_at = time.time() # 缺 key 门：认领之后、调 host.py 之前；缺 key 直接 failed， # 不调 runner（假 CLI 单测可断言 runner 未被调用）。 if not self.key_available(): error = _missing_key_error(self.env_key) self._set_status(task_id, "failed") + self._write_execution_result( + task_id, error_text=error, actual_model=self.model, + started_at=started_at, finished_at=time.time()) self._audit(task["project_id"], "worker_missing_key", error) + log_event(self.logger, "worker_missing_key", task_id=task_id, + project_id=task["project_id"], status="failed") return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id, @@ -370,14 +566,20 @@ class Worker: result = None error = None try: - result = self.executor.dispatch_task(task) - combined = (result.get("stdout", "") or "") + \ - (result.get("stderr", "") or "") - if result["status"] != "completed" and \ - _ALREADY_EXISTS_MARKER in combined: - # host 任务已存在（上轮 worker 在 dispatch 后、写回前 - # 崩溃）：改用 run 在已有任务上续跑，不重复建 worktree。 - result = self.executor.run_task(task) + if self._use_popen(): + result = self._execute_via_process(task) + combined = (result.get("stdout", "") or "") + \ + (result.get("stderr", "") or "") + if result["status"] not in ("completed", "cancelled") and \ + _ALREADY_EXISTS_MARKER in combined: + result = self._execute_via_process(task, kind="run") + else: + result = self.executor.dispatch_task(task) + combined = (result.get("stdout", "") or "") + \ + (result.get("stderr", "") or "") + if result["status"] != "completed" and \ + _ALREADY_EXISTS_MARKER in combined: + result = self.executor.run_task(task) except Exception as exc: # runner 异常也如实写回 failed result = {"executor": "local", "task_id": task_id, "status": "failed", "stdout": "", "returncode": None, @@ -386,8 +588,14 @@ class Worker: # 取消优先：执行期间被 cancel 的任务保持 interrupted， # 绝不用执行结果覆盖它（见模块 docstring 的取消说明）。 - if self._was_cancelled(task_id): + if self._was_cancelled(task_id) or result.get("status") == "cancelled": self._set_status(task_id, "interrupted") + self._write_execution_result( + task_id, result_text=(result or {}).get("stdout", ""), + error_text="cancelled", actual_model=self.model, + started_at=started_at, finished_at=time.time()) + log_event(self.logger, "worker_cancelled", task_id=task_id, + project_id=task["project_id"], status="interrupted") return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id, @@ -417,6 +625,11 @@ class Worker: self._audit(task["project_id"], "worker_fallback", "%s -> %s" % (self.model, nxt)) self._set_status(task_id, final) + finished_at = time.time() + self._write_execution_result( + task_id, result_text=stdout, error_text=error, + actual_model=self.model, started_at=started_at, + finished_at=finished_at) self.usage_turns += 1 # 成本：host 协议不暴露真实 token 数，用 cost.py 的估算结构 # 记 estimated，绝不标 actual；fallback_count 复用同一记录。 @@ -428,6 +641,11 @@ class Worker: self._audit(task["project_id"], "worker_completed" if final == "completed" else "worker_failed", task_id) + log_event(self.logger, + "worker_completed" if final == "completed" + else "worker_failed", + task_id=task_id, project_id=task["project_id"], + status=final) return {"dispatched": True, "task_id": task_id, "host_name": self.executor._host_name(task) if hasattr(self.executor, "_host_name") else task_id, @@ -449,14 +667,65 @@ class Worker: results.append(result) return results + def run_loop(self, interval=300, max_seconds=None, stop_event=None): + """常驻循环：每 interval 秒跑一轮 run_all，空闲也不退出。 + + 退出条件：stop_event 被置位、_loop_stop 被信号处理置位、 + 或 max_seconds 到期。退出时若有正在跑的任务（进程模式下 + _running_task_id 非空），先杀进程组、再标 interrupted， + 不让任务永远卡在 running。 + """ + started = time.time() + all_results = [] + try: + while not self._loop_stop: + if stop_event is not None and stop_event.is_set(): + break + if max_seconds is not None and \ + time.time() - started >= max_seconds: + break + try: + results = self.run_all() + except Exception as exc: + # 常驻循环一轮炸了不带走整个 worker：记结构化日志， + # 睡完本轮间隔再试下一轮（缺表、锁冲突都属此类）。 + log_event(self.logger, "worker_loop_error", + message="worker_loop_error: %s" % exc) + results = [] + all_results.extend(results) + # 按 0.2 秒切片睡，期间可响应 stop_event/_loop_stop。 + deadline = time.time() + interval + while time.time() < deadline: + if self._loop_stop: + break + if stop_event is not None and stop_event.is_set(): + break + if max_seconds is not None and \ + time.time() - started >= max_seconds: + break + time.sleep(0.2) + finally: + if self._running_task_id is not None: + task_id = self._running_task_id + if self._running_proc is not None: + self._kill_proc_group(self._running_proc) + try: + self._running_proc.wait(timeout=3) + except Exception: + pass + self._running_proc = None + self._running_task_id = None + self._set_status(task_id, "interrupted") + return all_results + def main(argv=None): - """单次派发循环：python3 -m cloud.worker --db <库> --host-py <路径>。 + """worker CLI：默认单次 run_all；--loop 进常驻循环。 只打印任务结果摘要（状态、host 名、错误文本），绝不打印 任何环境变量的值。 """ - parser = argparse.ArgumentParser(description="cloud worker (one pass)") + parser = argparse.ArgumentParser(description="cloud worker") parser.add_argument("--db", default="cloud-control-plane.db") parser.add_argument("--host-py", default=None, help="host.py 路径，缺省为 cloud/ 上一级的 host.py") @@ -468,7 +737,12 @@ def main(argv=None): parser.add_argument("--model", default=None) parser.add_argument("--max-turns", type=int, default=None) parser.add_argument("--max-tasks", type=int, default=None) + parser.add_argument("--loop", action="store_true", + help="常驻循环，每 --interval 秒跑一轮") + parser.add_argument("--interval", type=int, default=300) + parser.add_argument("--max-seconds", type=float, default=None) args = parser.parse_args(argv) + setup_logging() host_py = args.host_py if host_py is None: here = os.path.dirname(os.path.abspath(__file__)) @@ -477,8 +751,24 @@ def main(argv=None): repo=args.repo, env_key=args.env_key, codex_config_path=args.codex_config, model=args.model, max_turns=args.max_turns) + # CLI 起 worker 前先建表/迁移：全新库直接起循环不应因缺表而炸。 try: - results = worker.run_all(max_tasks=args.max_tasks) + from cloud.server import init_db as _init_db + _init_db(args.db) + except Exception as exc: + print("worker db 初始化失败：%s" % exc, file=sys.stderr) + raise SystemExit(1) + if args.loop: + def _on_signal(signum, frame): + worker._loop_stop = True + signal.signal(signal.SIGTERM, _on_signal) + signal.signal(signal.SIGINT, _on_signal) + try: + if args.loop: + results = worker.run_loop(interval=args.interval, + max_seconds=args.max_seconds) + else: + results = worker.run_all(max_tasks=args.max_tasks) finally: worker.close() print(json.dumps(