Merge pull request #8 from openqj/ops-batch

Ops batch: deploy automation, backup restore drill, monitoring, multi-user verification

.gitignore

@@ -11,3 +11,5 @@ __pycache__/
 host-config.json
 host-config-real.json
 host-state-real.json.bak*
+worker-heartbeat.json
+worker-heartbeat.json.tmp

cloud/README.md

@@ -570,3 +570,59 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   任务仍需 worker 认领执行；`CloudExecutor` 仍是占位符；公网 https
   入口仍需部署侧反向代理 + TLS。
 
+## 运维批（本轮新增：部署、恢复演练、监控、心跳、多用户验证）
+
+细则在 `deploy/README.md`，这里只记代码侧的接口与默认值：
+
+- 部署与回滚（`deploy/deploy.sh` / `deploy/rollback.sh`，VM 侧跑，
+  不碰 `cloud/` 代码）：部署前在主机上备份当前 `/opt/codex-app-server`
+  到 `/opt/codex-app-server-backups/`（时间戳命名，只留最近 3 份，排除
+  work/），解包新代码、chown codexsvc、重启 server+worker、循环验
+  `localhost:8765/health` 最多 60 秒（要求 ok=true、backend=postgres、
+  db_ok=true），失败自动恢复备份并重启；回滚只回代码，不回数据库
+  （迁移只加列、幂等，老代码仍能读新列，见 deploy/README 的回滚边界）。
+  两个脚本都不接受密码参数、不读 env 文件、不打印密钥；需经代理时用
+  环境变量 `SSH_PROXY_CMD` 传入 ProxyCommand，未设则直连。
+- 备份恢复演练（`cloud.backup.restore_postgres` + `deploy/restore-drill.sh`，
+  主机侧跑）：`restore_postgres(backup_file, target_dbname, runner=None)`，
+  用 psql 把 .sql 备份恢复到目标库；目标库名等于 `CLOUD_DB_NAME` 时
+  直接拒绝（除非显式传 `allow_production=True`，演练脚本永远不设它）；
+  备份文件不存在或为空时报错；密码只经 PGPASSWORD 环境变量传，绝不进
+  命令行、返回值或日志。演练脚本先跑一次 `backup_postgres` 生成新鲜
+  备份，再恢复到 `codex_control_restore_test_<时间戳>` 测试库，比对
+  tasks/projects/users/audit_log 四张表行数（只比行数，不打印行内容），
+  比对完立即删测试库（trap 保证中途失败也删）；绝不恢复到生产库，
+  绝不删生产库。
+- 监控与心跳：worker 的 `run_loop` 每次循环结束时写心跳文件，内容只有
+  `{"last_seen": epoch秒, "pid": 进程号}`，写失败只记日志不炸 worker；
+  路径由 `--heartbeat-file` 指定，未指定时看环境变量
+  `CLOUD_WORKER_HEARTBEAT_FILE`，再未指定则为 `./worker-heartbeat.json`
+  （VM 本地默认，主机路径不在代码里硬编码；主机部署时两边指到同一个
+  `/var/lib/codex-app-server/worker-heartbeat.json`，server 侧在 env 文件
+  里加 `CLOUD_WORKER_HEARTBEAT_FILE` 一行，worker 单元加
+  `--heartbeat-file` 参数，deploy.sh 不自动改 env 文件）。`GET /health`
+  在原有 ok/service/backend/db_ok 之外加 `worker_last_seen_seconds_ago`
+  （无文件或读失败时为 None）与 `worker_ok`（心跳 180 秒内为 true，
+  阈值可经 `CLOUD_WORKER_HEARTBEAT_MAX_AGE_SECONDS` 调，默认 180）；
+  health 本身仍返回 200，worker 不 ok 由调用方看字段判断，不要改成 503。
+  `deploy/monitor.sh`（主机侧，codexsvc 可读，不读 env 文件）检查
+  server/worker 的 systemctl 状态、health 四项、pg_isready、docker ps 里
+  codex-postgres 在运行、磁盘 <90%、app 目录存在，输出一行 JSON，任一
+  项 fail 则退出非零；`--self-test` 只验证脚本机制（health 解析 + JSON
+  合法），不验证主机健康。`deploy/monitor.service` /
+  `deploy/monitor.timer` 每 5 分钟跑一次，日志进 journal。
+- 多用户真实验证（`deploy/verify-multiuser.py`，主机侧以 codexsvc 身份
+  跑）：内存里建 4 个测试用户与 2 个测试项目（P2 只有 owner 是成员），
+  用 HTTP API 验证建任务、读任务、merge 的角色权限与跨项目隔离；
+  merge 用的 completed 测试任务是直接在 DB 里置 completed，不跑模型；
+  测试任务全部带永不存在的依赖 `multiuser-test-never-completes`，生产
+  worker 只派依赖全部 completed 的任务，所以测试任务永远不会被认领去
+  跑模型，不花模型钱、不改仓库；测试行全部用 `multiuser-test-` 前缀，
+  清理时只删这些行，清理失败退出非零并列出残留 id（不含 key）；审计
+  日志是 append-only 的，验证审计条目会留下，脚本不删审计表。
+- 仍未做（不要宣称已做）：推送告警通道（monitor 失败时通知到 Muse 聊天
+  的通道还没定）；回滚不回数据库（改类型/删列的迁移需单独评估）；恢复
+  演练只比行数，恢复到生产库的真实灾难恢复流程还没演练过；部署脚本、
+  恢复演练、monitor、多用户验证四项的主机实跑还没做（本轮只在 VM 上
+  跑了单测、deploy 干跑、monitor 自测与 VM 实跑）。
+

cloud/backup.py

@@ -101,6 +101,61 @@ def backup_postgres(out_dir, keep=3, timestamp=None, runner=None,
     return dest
 
 
+def restore_postgres(backup_file, target_dbname, runner=None,
+                     psql_cmd="psql", allow_production=False):
+    """Postgres 恢复：用 psql 把 .sql 备份恢复到目标库，返回目标库名。
+
+    安全规则（恢复演练的核心，绝不放宽）：
+    - 目标库名绝不能等于生产库名（CLOUD_DB_NAME 环境变量读到的值），
+      相等时直接报错拒绝；只有调用方显式传 allow_production=True
+      才放行，而部署脚本（deploy/restore-drill.sh）永远不设这个值。
+    - 恢复前要求备份文件存在且非空，否则报错，不跑 psql。
+    - 连接参数与 backup_postgres 同款：host/port/user 走 CLOUD_DB_*
+      环境变量，密码只以 PGPASSWORD 环境变量传给子进程，绝不
+      出现在命令行参数、返回值或日志里。
+    - 目标库必须已存在（调用方先 CREATE DATABASE，见恢复演练
+      脚本）；psql 非零退出时明确报错（stderr 截断，不含密码）。
+    - 本函数只恢复，不删库、不改生产库；删测试库是调用方的事。
+    """
+    if not backup_file or not os.path.isfile(backup_file):
+        raise FileNotFoundError("备份文件不存在：%s" % backup_file)
+    if os.path.getsize(backup_file) == 0:
+        raise ValueError("备份文件为空，拒绝恢复：%s" % backup_file)
+    if not target_dbname or not isinstance(target_dbname, str) \
+            or not target_dbname.strip() \
+            or any(c.isspace() for c in target_dbname):
+        raise ValueError("目标库名非法：%r" % (target_dbname,))
+    production = os.environ.get("CLOUD_DB_NAME") or ""
+    if production and target_dbname == production \
+            and not allow_production:
+        raise ValueError(
+            "拒绝恢复到生产库 %s：目标库名与 CLOUD_DB_NAME 相同；"
+            "恢复演练必须用测试库名" % production)
+    host = os.environ.get("CLOUD_DB_HOST") or "127.0.0.1"
+    port = os.environ.get("CLOUD_DB_PORT") or "5432"
+    user = os.environ.get("CLOUD_DB_USER")
+    if not user:
+        raise RuntimeError("Postgres 恢复需要 CLOUD_DB_USER")
+    if shutil.which(psql_cmd) is None and runner is None:
+        raise RuntimeError("psql 不可用：本机找不到 %s" % psql_cmd)
+    cmd = [psql_cmd, "-h", host, "-p", str(port), "-U", user,
+           "-d", target_dbname, "-v", "ON_ERROR_STOP=1",
+           "-f", backup_file]
+    env = dict(os.environ)
+    password = os.environ.get("CLOUD_DB_PASSWORD")
+    if password:
+        env["PGPASSWORD"] = password
+    if runner is not None:
+        result = runner(cmd, env)
+    else:
+        result = subprocess.run(cmd, capture_output=True, text=True,
+                                env=env)
+    if getattr(result, "returncode", 1) != 0:
+        err = (getattr(result, "stderr", "") or "")[:500]
+        raise RuntimeError("psql 恢复失败：%s" % err)
+    return target_dbname
+
+
 def main(argv=None):
     parser = argparse.ArgumentParser(description="cloud control-plane backup")
     parser.add_argument("--db", default=None,

cloud/server.py

@@ -156,6 +156,47 @@ def _db_backend_name():
     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 分发。
 
@@ -448,9 +489,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     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/"):

cloud/tests/test_ops_batch.py

@@ -0,0 +1,206 @@
+"""运维批单测：恢复演练、worker 心跳、health worker 字段、monitor 自测。
+
+只用标准库，不跑真实模型回合、不联网、不连真实 GitHub、不连生产库；
+restore 的 psql 与 monitor 的 systemctl/docker 一律用假 runner 或
+--self-test 的 canned server 验证，不在本机起真实服务。
+"""
+
+import http.client
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.backup import restore_postgres
+    from cloud.server import create_server
+    from cloud.worker import Worker
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.backup import restore_postgres
+    from cloud.server import create_server
+    from cloud.worker import Worker
+
+
+def _fake_result(returncode=0, stdout="", stderr=""):
+    return SimpleNamespace(returncode=returncode, stdout=stdout,
+                           stderr=stderr)
+
+
+class RestorePostgresTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.backup = os.path.join(self.tmp.name, "backup.sql")
+        with open(self.backup, "w") as f:
+            f.write("-- fake dump\n")
+        self.old_env = dict(os.environ)
+        os.environ["CLOUD_DB_HOST"] = "127.0.0.1"
+        os.environ["CLOUD_DB_PORT"] = "5432"
+        os.environ["CLOUD_DB_NAME"] = "prod_db"
+        os.environ["CLOUD_DB_USER"] = "codexsvc"
+        os.environ["CLOUD_DB_PASSWORD"] = "test-secret-pw"
+
+    def tearDown(self):
+        os.environ.clear()
+        os.environ.update(self.old_env)
+        self.tmp.cleanup()
+
+    def test_target_equal_to_production_rejected(self):
+        # 目标库名等于生产库名时直接拒绝，allow_production 默认 False。
+        with self.assertRaises(ValueError):
+            restore_postgres(self.backup, "prod_db",
+                             runner=lambda c, e: _fake_result())
+
+    def test_target_equal_to_production_allowed_only_explicitly(self):
+        calls = []
+
+        def runner(cmd, env):
+            calls.append((list(cmd), dict(env)))
+            return _fake_result()
+
+        out = restore_postgres(self.backup, "prod_db", runner=runner,
+                               allow_production=True)
+        self.assertEqual(out, "prod_db")
+        self.assertEqual(len(calls), 1)
+
+    def test_missing_backup_file_rejected(self):
+        with self.assertRaises(FileNotFoundError):
+            restore_postgres(os.path.join(self.tmp.name, "nope.sql"),
+                             "restore_test",
+                             runner=lambda c, e: _fake_result())
+
+    def test_empty_backup_file_rejected(self):
+        empty = os.path.join(self.tmp.name, "empty.sql")
+        open(empty, "w").close()
+        with self.assertRaises(ValueError):
+            restore_postgres(empty, "restore_test",
+                             runner=lambda c, e: _fake_result())
+
+    def test_password_only_in_env_not_in_command(self):
+        calls = []
+
+        def runner(cmd, env):
+            calls.append((list(cmd), dict(env)))
+            return _fake_result()
+
+        out = restore_postgres(self.backup, "restore_test", runner=runner)
+        self.assertEqual(out, "restore_test")
+        cmd, env = calls[0]
+        self.assertNotIn("test-secret-pw", " ".join(cmd))
+        self.assertEqual(env.get("PGPASSWORD"), "test-secret-pw")
+        self.assertIn("-f", cmd)
+        self.assertIn(self.backup, cmd)
+        self.assertIn("restore_test", cmd)
+
+    def test_psql_failure_raises(self):
+        def runner(cmd, env):
+            return _fake_result(returncode=1, stderr="psql boom")
+
+        with self.assertRaises(RuntimeError):
+            restore_postgres(self.backup, "restore_test", runner=runner)
+
+
+class WorkerHeartbeatTestCase(unittest.TestCase):
+    def test_heartbeat_file_written_with_last_seen_and_pid(self):
+        tmp = tempfile.TemporaryDirectory()
+        hb = os.path.join(tmp.name, "worker-heartbeat.json")
+        worker = Worker(os.path.join(tmp.name, "db.db"), "/nonexistent/host.py",
+                        env_key="TEST_KEY", heartbeat_file=hb)
+        worker._write_heartbeat()
+        with open(hb) as f:
+            payload = json.load(f)
+        self.assertIn("last_seen", payload)
+        self.assertEqual(payload["pid"], os.getpid())
+        self.assertLess(abs(payload["last_seen"] - time.time()), 5)
+        tmp.cleanup()
+
+    def test_heartbeat_write_failure_does_not_crash_worker(self):
+        # 心跳目录不可写时只记日志，不抛异常（写到一个文件路径下当目录）。
+        tmp = tempfile.TemporaryDirectory()
+        blocker = os.path.join(tmp.name, "blocker")
+        open(blocker, "w").close()
+        worker = Worker(os.path.join(tmp.name, "db.db"), "/nonexistent/host.py",
+                        env_key="TEST_KEY",
+                        heartbeat_file=os.path.join(blocker, "hb.json"))
+        worker._write_heartbeat()  # 不抛异常即通过
+        tmp.cleanup()
+
+
+class HealthWorkerFieldTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        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 _health(self):
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        conn.request("GET", "/health")
+        resp = conn.getresponse()
+        body = json.loads(resp.read())
+        conn.close()
+        return resp.status, body
+
+    def test_health_worker_fresh_ok(self):
+        hb = os.path.join(self.tmp.name, "hb.json")
+        with open(hb, "w") as f:
+            json.dump({"last_seen": time.time(), "pid": 123}, f)
+        self.server.worker_heartbeat_file = hb
+        status, body = self._health()
+        self.assertEqual(status, 200)
+        self.assertTrue(body["ok"])
+        self.assertTrue(body["worker_ok"])
+        self.assertIsNotNone(body["worker_last_seen_seconds_ago"])
+        self.assertLess(body["worker_last_seen_seconds_ago"], 5)
+
+    def test_health_worker_stale_not_ok_but_still_200(self):
+        hb = os.path.join(self.tmp.name, "hb.json")
+        with open(hb, "w") as f:
+            json.dump({"last_seen": time.time() - 1000, "pid": 123}, f)
+        self.server.worker_heartbeat_file = hb
+        status, body = self._health()
+        # worker 不 ok 不代表 server 挂了，health 本身仍返回 200。
+        self.assertEqual(status, 200)
+        self.assertFalse(body["worker_ok"])
+        self.assertGreater(body["worker_last_seen_seconds_ago"], 500)
+
+    def test_health_worker_missing_file_none_false(self):
+        self.server.worker_heartbeat_file = os.path.join(
+            self.tmp.name, "no-such-hb.json")
+        status, body = self._health()
+        self.assertEqual(status, 200)
+        self.assertIsNone(body["worker_last_seen_seconds_ago"])
+        self.assertFalse(body["worker_ok"])
+
+
+class MonitorSelfTestCase(unittest.TestCase):
+    def test_monitor_self_test_passes(self):
+        # monitor.sh --self-test 只验证脚本机制（health 解析 + JSON 合法），
+        # 不断言主机健康；VM 上其它项 fail 是正常的，self-test 仍应通过。
+        repo_root = Path(__file__).resolve().parents[2]
+        script = repo_root / "deploy" / "monitor.sh"
+        self.assertTrue(script.exists())
+        result = subprocess.run(
+            ["sh", str(script), "--self-test"],
+            capture_output=True, text=True, timeout=60)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertIn("SELF-TEST PASS", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -208,7 +208,7 @@ class Worker:
                  cost_store=None, process_factory=None,
                  task_timeout=1800, max_retries=1,
                  retry_backoff_seconds=5.0, repos_dir="./repos",
-                 repos_runner=None):
+                 repos_runner=None, heartbeat_file=None):
         self.db_path = str(db_path)
         self.host_py_path = host_py_path
         self.config_path = config_path
@@ -252,6 +252,31 @@ class Worker:
         # repos_runner 可注入（单测用假 git runner），缺省走真实 git。
         self.repos_dir = repos_dir
         self.repos_runner = repos_runner
+        # 运维心跳：路径优先显式参数，其次环境变量，默认当前目录下
+        # 的 worker-heartbeat.json；主机路径不在代码里硬编码，主机
+        # 部署时用 --heartbeat-file 或环境变量传入（见 deploy/README）。
+        # 文件里只有 last_seen 与 pid，绝无任何密钥。
+        self.heartbeat_file = heartbeat_file or \
+            os.environ.get("CLOUD_WORKER_HEARTBEAT_FILE") or \
+            "./worker-heartbeat.json"
+
+    def _write_heartbeat(self):
+        """写一次运维心跳：写失败只记日志，绝不炸 worker。
+
+        先写临时文件再 os.replace，保证 server 读到的要么是旧的
+        完整文件、要么是新的完整文件，不会读到写一半的 JSON。
+        """
+        try:
+            payload = {"last_seen": time.time(), "pid": os.getpid()}
+            tmp = "%s.tmp" % self.heartbeat_file
+            parent = os.path.dirname(os.path.abspath(self.heartbeat_file))
+            os.makedirs(parent, exist_ok=True)
+            with open(tmp, "w", encoding="utf-8") as f:
+                f.write(json.dumps(payload))
+            os.replace(tmp, self.heartbeat_file)
+        except Exception as exc:
+            log_event(self.logger, "worker_heartbeat_failed",
+                      message="worker heartbeat write failed: %s" % exc)
 
     # -- 成本存储（惰性建表，避免只认领不执行时也建表）--------------
     @property
@@ -954,6 +979,9 @@ class Worker:
         # 常驻启动清理兜底：清掉上一轮残留的 codex- 容器，
         # 失败只记日志（见 cleanup_stale_containers），不炸循环。
         self.cleanup_stale_containers()
+        # 运维心跳：启动时先写一次，之后每轮结束再写，server 的
+        # /health 据此算 worker_ok（见 server.py 的健康检查）。
+        self._write_heartbeat()
         started = time.time()
         all_results = []
         try:
@@ -972,6 +1000,9 @@ class Worker:
                               message="worker_loop_error: %s" % exc)
                     results = []
                 all_results.extend(results)
+                # 每轮结束写心跳：哪怕本轮一个任务都没派，worker 活着
+                # 这件事本身也要被 server 看到。
+                self._write_heartbeat()
                 # 按 0.2 秒切片睡，期间可响应 stop_event/_loop_stop。
                 deadline = time.time() + interval
                 while time.time() < deadline:
@@ -1025,6 +1056,12 @@ def main(argv=None):
                         help="host.py 路径，缺省为 cloud/ 上一级的 host.py")
     parser.add_argument("--config", default=None)
     parser.add_argument("--repo", default=None)
+    parser.add_argument("--heartbeat-file", default=None,
+                        help="运维心跳文件路径（默认：环境变量 "
+                             "CLOUD_WORKER_HEARTBEAT_FILE，未设则为 "
+                             "./worker-heartbeat.json；主机部署时用此"
+                             "参数指到 /var/lib/codex-app-server/ 下，"
+                             "与 server 的 health 读取路径保持一致）")
     parser.add_argument("--repos-dir", default="./repos",
                         help="项目仓库检出根目录（默认 ./repos）；项目有"
                              " repo_url 时 worker 在此 clone/fetch 各项目"
@@ -1081,7 +1118,8 @@ def main(argv=None):
                     task_timeout=args.task_timeout,
                     max_retries=args.max_retries,
                     retry_backoff_seconds=args.retry_backoff_seconds,
-                    repos_dir=args.repos_dir)
+                    repos_dir=args.repos_dir,
+                    heartbeat_file=args.heartbeat_file)
     # CLI 启动清理兜底（常驻与单次都做）：清掉上一轮残留容器。
     worker.cleanup_stale_containers()
     # CLI 起 worker 前先建表/迁移：全新库直接起循环不应因缺表而炸。

deploy/README.md

@@ -0,0 +1,179 @@
+# 部署与运维（deploy/）
+
+本目录是主机部署与运维的脚本集，与 `cloud/` 代码分开：部署脚本只负责
+传代码、备份、重启、验证，不改任务逻辑；运维脚本只负责检查与演练，
+不跑模型、不推代码。
+
+## 部署：deploy.sh
+
+VM 侧运行：
+
+```sh
+deploy/deploy.sh <host> [user] [--dry-run]
+# 例如：deploy/deploy.sh 124.222.165.170 root
+# 干跑（只打印步骤，不传代码、不动主机）：
+#   deploy/deploy.sh <host> --dry-run
+```
+
+- 密码：脚本绝不接受密码参数、不读 env 文件、不打印密钥。SSH 密码由
+  ssh/scp 交互提示输入，或由调用方自己的 SSH 通道（key）提供。
+- 经代理：设环境变量 `SSH_PROXY_CMD` 传入 ProxyCommand 内容，ssh 会自行
+  展开 `%h %p`，例如：
+  `SSH_PROXY_CMD="python3 /tmp/proxy_tunnel.py %h %p" deploy/deploy.sh <host>`；
+  未设则直连 ssh。
+- 备份保留策略：每次部署前在主机上把当前 `/opt/codex-app-server` 打包到
+  `/opt/codex-app-server-backups/codex-app-server-<时间戳>.tar.gz`（排除
+  work/ 目录以免过大），只留最近 3 份，多余的自动删掉。
+- 健康门：重启后循环 curl `localhost:8765/health` 最多 60 秒，要求
+  `ok=true` 且 `backend=postgres` 且 `db_ok=true`；失败时自动恢复刚打的
+  备份包并重启，再验一次。回滚后 health 恢复也算部署失败（退出非零），
+  因为新代码没上去；回滚后 health 仍失败则退出非零并明说
+  “已回滚但 health 仍失败，需人工介入”。
+- 成功时只打印新代码的 git commit（从本机 git 取，不编造版本号）。
+
+## 回滚：rollback.sh
+
+VM 侧运行，恢复主机上最新的一份备份包：
+
+```sh
+deploy/rollback.sh <host> [user]
+```
+
+恢复后同样重启并验 health（60 秒），成功/失败都明说；主机上没有备份包
+时直接报错退出，不编造。
+
+### 回滚边界：只回代码，不回数据库
+
+数据库迁移是只加列的幂等操作（例如 projects.repo_url、tasks 的 6 列），
+回滚代码后老代码仍能读新列——新列对老代码只是多余的列，不影响已有
+查询（老代码的 SELECT 只取它认识的列，或用 `_row_get` 容错）。所以只回
+代码是安全的；数据库本身绝不会被 deploy.sh / rollback.sh 动到。这一点
+逻辑上成立的前提是：迁移永远只加列、不改列类型、不删列；如果将来有一次
+迁移要改类型或删列，必须先单独评估回滚方案，不能直接套用本脚本。
+
+## 备份恢复演练：restore-drill.sh
+
+主机侧运行（root，或能读 `/etc/codex-app-server/env` 的用户）：
+
+```sh
+sh /opt/codex-app-server/deploy/restore-drill.sh
+# 可选：BACKUP_DIR=/var/lib/codex-app-server/backups（默认就是这个）
+```
+
+步骤：跑一次 `backup_postgres` 生成新鲜备份 → 新建测试库
+`codex_control_restore_test_<时间戳>` → 用 `cloud.backup.restore_postgres`
+把备份恢复进去 → 比对生产库与测试库的 tasks / projects / users /
+audit_log 行数（行数相等即通过，只比行数，不比对具体内容，不打印任何
+行内容）→ 比对完立即删掉测试库（trap 保证中途失败也删）。
+
+安全红线：恢复目标永远是新生成的测试库名，绝不等于生产库名；
+`restore_postgres` 本身也会拒绝目标库名等于 `CLOUD_DB_NAME`
+（`allow_production` 绝不在本脚本里设为 True），双重保险。本脚本绝不
+恢复到生产库，绝不删生产库；全程只打印 PASS/FAIL 与行数，不打印密码，
+不打印备份内容。
+
+## 监控：monitor.sh + monitor.service / monitor.timer
+
+主机侧运行，codexsvc 可读权限即可，不读 env 文件：
+
+```sh
+sh /opt/codex-app-server/deploy/monitor.sh            # 输出一行 JSON，任一项 fail 则退出非零
+sh /opt/codex-app-server/deploy/monitor.sh --self-test  # 只验证脚本机制（见下）
+```
+
+检查项：systemctl is-active server/worker、curl health（ok/backend/db_ok/
+worker_ok）、pg_isready、docker ps 里 codex-postgres 在运行、磁盘使用率
+<90%、/opt/codex-app-server 存在。health 与 pg_isready 都不需要密码；
+需要密码的 SELECT 1 那一步直接跳过，不在脚本里读密码。
+
+`--self-test` 只验证脚本本身的机制（health JSON 解析 + 输出是合法 JSON）：
+它起一个本地 canned health server，断言 health 那一项为 pass；其它项在
+非主机机器上 fail 是正常的，self-test 不因此失败。不要把 --self-test
+通过误读成“主机健康”。
+
+定时（主机上安装，示例）：
+
+```sh
+cp /opt/codex-app-server/deploy/monitor.service /etc/systemd/system/monitor.service
+cp /opt/codex-app-server/deploy/monitor.timer /etc/systemd/system/monitor.timer
+systemctl daemon-reload
+systemctl enable --now monitor.timer
+journalctl -u monitor.service -f   # 看每次检查的输出
+```
+
+timer 每 5 分钟跑一次，日志进 journal。
+
+### 监控边界：本轮只有“主机侧检查 + journal 日志”
+
+monitor.sh 本身不会发消息给用户。推送到 Muse 聊天的告警仍需用户另行
+决定通道（例如 SSH key 常驻通道让外部 cron 拉 journal，或者一个
+webhook URL 让 monitor 失败时 POST 过去），在通道定下来之前，不要假装
+monitor.sh 会通知任何人。手工验证步骤：主机上跑一次 monitor.sh 看 JSON
+与退出码，再跑一次 --self-test 看 SELF-TEST PASS，最后
+`systemctl list-timers | grep monitor` 确认 timer 已启用。
+
+## worker 心跳与 health 的 worker 字段
+
+worker 的 `run_loop` 每次循环结束时写心跳文件，内容只有
+`{"last_seen": epoch秒, "pid": 进程号}`，不含任何密钥；写失败只记日志，
+不炸 worker。心跳文件路径：worker 用 `--heartbeat-file` 指定（主机部署
+时指到 `/var/lib/codex-app-server/worker-heartbeat.json`），未指定时看
+环境变量 `CLOUD_WORKER_HEARTBEAT_FILE`，再未指定则为
+`./worker-heartbeat.json`（VM 本地跑的默认，不在代码里硬编码主机路径）。
+
+server 的 `GET /health` 在原有 ok/service/backend/db_ok 之外加两个字段：
+`worker_last_seen_seconds_ago`（心跳文件不存在或读失败时为 None）与
+`worker_ok`（心跳在 180 秒内为 true，否则 false；阈值可经环境变量
+`CLOUD_WORKER_HEARTBEAT_MAX_AGE_SECONDS` 调，默认 180）。health 本身仍
+返回 200——worker 不 ok 不代表 server 挂了，调用方看 worker_ok 字段判断，
+不要把 health 改成 503 破坏已有调用。
+
+主机部署时两边路径必须一致：worker 单元加
+`--heartbeat-file /var/lib/codex-app-server/worker-heartbeat.json`，同时
+在 server 的 env 文件（`/etc/codex-app-server/env`，server 单元的
+EnvironmentFile）里加一行
+`CLOUD_WORKER_HEARTBEAT_FILE=/var/lib/codex-app-server/worker-heartbeat.json`。
+deploy.sh 不会自动改 env 文件（那是 root-only 的密钥文件），这一行由
+部署操作人在主机上手动加，加完重启 server 即可。
+
+## 多用户真实验证：verify-multiuser.py
+
+主机侧以 codexsvc 身份运行，直接对生产库 + 本地 HTTP API 验证：
+
+```sh
+python3 /opt/codex-app-server/deploy/verify-multiuser.py
+# 可选：--base http://127.0.0.1:8765 --db /var/lib/codex-app-server/control.db
+```
+
+它在内存里创建 4 个测试用户（owner / maintainer / contributor / viewer）
+与 2 个测试项目（P2 只有 owner 是成员，用于隔离验证），再用 HTTP API
+验证：owner/maintainer/contributor/viewer 的建任务、读任务、merge 权限，
+以及跨项目隔离（P1 成员读 P2 任务 403）、无 key 401。merge 验证用的
+completed 测试任务是直接在 DB 里把测试任务置 completed，不跑真实模型。
+
+防误伤要点（脚本逻辑里已实现）：
+- 测试用户、项目、任务全部用 `multiuser-test-` 前缀，清理时只删这些
+  前缀的行，绝不碰其它生产数据；清理失败时退出非零并列出残留测试 id
+  （不含 key），不静默假装干净。
+- 测试任务创建时都带一个永不存在的依赖 `multiuser-test-never-completes`：
+  worker 只派依赖全部 completed 的 queued 任务，所以这些测试任务永远
+  不会被生产 worker 认领去跑模型，不花模型钱、不改任何仓库。
+- API key 只存在内存里，脚本全程只打印每项 PASS/FAIL 与总数，绝不打印
+  任何 API key；任务 prompt 与用户 id 也不打印，只打印检查名与结果。
+- 审计日志是 append-only 的：验证过程的审计条目会留在 audit_log 里，
+  脚本不删审计表——审计就该留痕，这一点不是清理遗漏。
+
+## 公网入口与 webhook secret（已有文件，不属于本轮新增）
+
+- `caddy/Caddyfile`：只暴露 `/webhooks/*` 与 `/oauth/*`，任务/项目/管理
+  API 对外一律 403；域名与邮箱占位符需部署时替换。
+- `setup-webhook-secret.sh`：在主机上生成 webhook secret 并写入 env 文件，
+  绝不打印 secret；已设则不动，避免轮换后 GitHub 端旧 secret 失效。
+
+## 仍未做（不要假装已做）
+
+- 推送告警通道：monitor 失败时通知到 Muse 聊天的通道还没定（见监控一节）。
+- 回滚不回数据库：见回滚边界一节；改类型/删列的迁移需要单独评估。
+- 恢复演练不碰生产库：演练只比行数，不比对具体内容；恢复到生产库的
+  真实灾难恢复流程（停服务、恢复、重启、验 health）还没演练过。
+- CloudExecutor、GUI、Slack 执行、推送实发：与本轮无关，仍是 0 或占位。

deploy/deploy.sh

@@ -0,0 +1,158 @@
+#!/bin/sh
+# 部署自动化：VM 侧运行，把本地代码同步到外部主机并重启验证，失败自动回滚。
+#
+# 用法：
+#   deploy/deploy.sh <host> [user] [--dry-run]
+#   host：主机 IP 或域名（必填）；user：SSH 用户（默认 root）。
+#   --dry-run：只打印将要做的步骤，不 scp、不 ssh、不改主机。
+#
+# 密码：本脚本绝不接受密码参数、不读任何 env 文件、不打印任何密钥。
+# SSH 密码由 ssh/scp 交互提示输入，或由调用方自己的 SSH 通道提供
+# （例如已配好的 key）。需要经代理时，用环境变量 SSH_PROXY_CMD
+# 传入 ProxyCommand 内容（ssh 会自行展开 %h %p），未设则直连：
+#   SSH_PROXY_CMD="python3 /tmp/proxy_tunnel.py %h %p" deploy/deploy.sh <host>
+#
+# 步骤：
+#   1. 本机打代码 tar 包（排除 bin/、work/、.git、__pycache__、state、
+#      host-config-real.json，与之前手工同步的排除规则一致）。
+#   2. scp 到主机 /tmp/codex-app-server-code.tar.gz。
+#   3. ssh 到主机：先把当前 /opt/codex-app-server 打备份包到
+#      /opt/codex-app-server-backups/（时间戳命名，只留最近 3 份，
+#      备份排除 work/ 目录以免过大）；解包新代码并 chown codexsvc；
+#      重启 server+worker；循环 curl localhost:8765/health 最多 60 秒，
+#      要求 ok=true 且 backend=postgres 且 db_ok=true。
+#   4. 健康检查失败时自动恢复刚打的备份包、重启、再验 health；回滚后
+#      health 恢复也算部署失败（退出非零），因为新代码没上去；回滚后
+#      health 仍失败则退出非零并明说“已回滚但 health 仍失败，需人工介入”。
+#   5. 成功时只打印新代码的 git commit（从本机 git 取，不编造版本号）。
+#
+# 回滚边界：回滚只回代码，不回数据库。数据库迁移是只加列的幂等操作
+# （projects.repo_url、tasks 的 6 列等），回滚代码后老代码仍能读新列，
+# 所以只回代码是安全的；数据库本身绝不会被本脚本动到。
+set -eu
+
+HOST=${1:-}
+USER_ARG=${2:-root}
+DRY_RUN=0
+for a in "$@"; do
+  if [ "$a" = "--dry-run" ]; then DRY_RUN=1; fi
+done
+if [ -z "$HOST" ] || [ "$HOST" = "--dry-run" ]; then
+  echo "用法：deploy/deploy.sh <host> [user] [--dry-run]" >&2
+  exit 2
+fi
+# user 位置可能是 --dry-run，处理一下
+if [ "$USER_ARG" = "--dry-run" ]; then USER_ARG=root; fi
+
+SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
+REPO_ROOT=$(CDPATH= cd -- "$SCRIPT_DIR/.." && pwd)
+cd "$REPO_ROOT"
+
+COMMIT=$(git rev-parse HEAD 2>/dev/null || echo "unknown")
+TARBALL="$REPO_ROOT/work/external-install/codex-app-server-code.tar.gz"
+REMOTE_TARBALL="/tmp/codex-app-server-code.tar.gz"
+
+# ssh/scp 命令构造：有 SSH_PROXY_CMD 时走代理，否则直连。
+ssh_cmd() {
+  if [ -n "${SSH_PROXY_CMD:-}" ]; then
+    ssh -o "ProxyCommand $SSH_PROXY_CMD" -o StrictHostKeyChecking=no \
+      -o UserKnownHostsFile=/dev/null "$@"
+  else
+    ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null "$@"
+  fi
+}
+scp_cmd() {
+  if [ -n "${SSH_PROXY_CMD:-}" ]; then
+    scp -o "ProxyCommand $SSH_PROXY_CMD" -o StrictHostKeyChecking=no \
+      -o UserKnownHostsFile=/dev/null "$@"
+  else
+    scp -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null "$@"
+  fi
+}
+
+echo "部署目标：$USER_ARG@$HOST，代码 commit：$COMMIT"
+if [ "$DRY_RUN" = "1" ]; then
+  echo "[dry-run] 将打 tar 包：$TARBALL（排除 bin/ work/ .git __pycache__ state host-config-real.json）"
+  echo "[dry-run] 将 scp 到 $USER_ARG@$HOST:$REMOTE_TARBALL"
+  echo "[dry-run] 将在主机上备份 /opt/codex-app-server 到 /opt/codex-app-server-backups/（留 3 份），解包、chown codexsvc、重启 server+worker、验 health 60 秒，失败自动回滚"
+  exit 0
+fi
+
+mkdir -p "$REPO_ROOT/work/external-install"
+tar --exclude='./bin' --exclude='./work' --exclude='./.git' \
+  --exclude='__pycache__' --exclude='*.pyc' \
+  --exclude='host-state*.json*' --exclude='host-daemon.*' \
+  --exclude='host-config-real.json' --exclude='host-run.log' \
+  -czf "$TARBALL" .
+echo "tar 包已生成：$TARBALL"
+
+scp_cmd "$TARBALL" "$USER_ARG@$HOST:$REMOTE_TARBALL"
+echo "tar 包已传到主机"
+
+# 主机侧部署脚本：备份 -> 解包 -> 重启 -> 验 health -> 失败回滚。
+# 注意：这里用单引号 heredoc 传过去，主机侧变量不被本机展开；
+# 健康判断用 python3 解析 JSON，不依赖 jq（主机未必装 jq）。
+ssh_cmd "$USER_ARG@$HOST" 'sh -s' <<'HOST_SCRIPT'
+set -eu
+APP_DIR=/opt/codex-app-server
+BACKUP_DIR=/opt/codex-app-server-backups
+STAMP=$(date +%Y%m%d%H%M%S)
+mkdir -p "$BACKUP_DIR"
+BACKUP_FILE="$BACKUP_DIR/codex-app-server-$STAMP.tar.gz"
+
+check_health() {
+  # 要求 ok=true 且 backend=postgres 且 db_ok=true，满足返回 0。
+  OUT=$(curl -s --max-time 5 http://127.0.0.1:8765/health || true)
+  [ -n "$OUT" ] && printf '%s' "$OUT" | python3 -c '
+import json, sys
+try:
+    d = json.load(sys.stdin)
+except Exception:
+    sys.exit(1)
+sys.exit(0 if (d.get("ok") is True and d.get("backend") == "postgres"
+                and d.get("db_ok") is True) else 1)
+'
+}
+
+wait_health() {
+  i=0
+  while [ "$i" -lt 12 ]; do
+    if check_health; then return 0; fi
+    sleep 5
+    i=$((i + 1))
+  done
+  return 1
+}
+
+# 1. 备份当前代码（排除 work/，只留最近 3 份）
+tar --exclude='./work' -czf "$BACKUP_FILE" -C /opt codex-app-server
+echo "已备份当前代码到：$BACKUP_FILE"
+ls -t "$BACKUP_DIR"/codex-app-server-*.tar.gz 2>/dev/null \
+  | tail -n +4 | xargs -r rm -f
+
+# 2. 解包新代码并改属主
+tar -xzf /tmp/codex-app-server-code.tar.gz -C "$APP_DIR"
+chown -R codexsvc:codexsvc "$APP_DIR"
+
+# 3. 重启并验 health
+systemctl restart codex-cloud-server codex-cloud-worker
+if wait_health; then
+  echo "部署成功，health 通过"
+  exit 0
+fi
+
+# 4. 失败：恢复备份并重启，再验一次
+echo "新代码 health 未通过，开始回滚到备份：$BACKUP_FILE"
+systemctl stop codex-cloud-worker codex-cloud-server || true
+tar -xzf "$BACKUP_FILE" -C /opt
+chown -R codexsvc:codexsvc "$APP_DIR"
+systemctl start codex-cloud-server codex-cloud-worker
+if wait_health; then
+  echo "部署失败，已回滚到备份，health 已恢复（新代码未上线）" >&2
+  exit 1
+fi
+echo "已回滚但 health 仍失败，需人工介入" >&2
+exit 1
+HOST_SCRIPT
+
+echo "部署完成，代码 commit：$COMMIT"

deploy/monitor.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=Codex cloud monitor check (one-shot)
+After=network.target codex-cloud-server.service codex-cloud-worker.service
+
+[Service]
+Type=oneshot
+User=codexsvc
+WorkingDirectory=/opt/codex-app-server
+# monitor.sh 只做检查、不读 env 文件、不发消息；失败时 systemd 会把
+# 退出码与输出记进 journal，告警通道见 deploy/README.md（另行决定）。
+ExecStart=/bin/sh /opt/codex-app-server/deploy/monitor.sh
+
+[Install]
+WantedBy=multi-user.target

deploy/monitor.sh

@@ -0,0 +1,171 @@
+#!/bin/sh
+# 主机侧监控检查：输出一行 JSON 状态，任一项 fail 则退出非零。
+#
+# 用法（主机上，codexsvc 可读权限即可）：
+#   sh /opt/codex-app-server/deploy/monitor.sh
+#   sh /opt/codex-app-server/deploy/monitor.sh --self-test
+#
+# 检查项（全部不需要密码、不读 env 文件）：
+#   server：systemctl is-active codex-cloud-server 为 active
+#   worker：systemctl is-active codex-cloud-worker 为 active
+#   health：curl health，要求 ok=true、backend=postgres、db_ok=true、
+#           worker_ok=true（worker 心跳 180 秒内，见 server.py）
+#   postgres：pg_isready -h 127.0.0.1 -p 5432（只测就绪，不连库查数据，
+#           所以不需要密码；需要密码的 SELECT 1 那一步直接跳过，不做）
+#   docker_postgres：docker ps 里 codex-postgres 在运行
+#   disk：磁盘使用率 < 90%（默认看 /var/lib/codex-app-server，不存在则看 /）
+#   app_dir：/opt/codex-app-server 目录存在
+#
+# 输出示例（单行 JSON，值只会是 pass/fail，不含任何密钥）：
+#   {"overall": "pass", "checks": {"server": "pass", ...}}
+#
+# --self-test：只验证脚本本身的机制（health JSON 解析 + 输出是合法 JSON），
+# 不验证主机健康。它会起一个本地 canned health server（只返回固定健康
+# JSON），断言 health 那一项为 pass；其它项在非主机机器上 fail 是正常的，
+# self-test 不因此失败。不要把 --self-test 通过误读成“主机健康”。
+#
+# 告警边界：本脚本只做“检查 + journal 日志”（配合 monitor.timer 每 5 分钟
+# 跑一次），它本身不会发消息给用户。推送到 Muse 聊天的告警仍需用户另行
+# 决定通道（SSH key 常驻通道或 webhook URL），见 deploy/README.md。
+set -eu
+
+HEALTH_URL=${MONITOR_HEALTH_URL:-http://127.0.0.1:8765/health}
+APP_DIR=${MONITOR_APP_DIR:-/opt/codex-app-server}
+DISK_PATH=${MONITOR_DISK_PATH:-/var/lib/codex-app-server}
+if [ ! -d "$DISK_PATH" ]; then DISK_PATH="/"; fi
+
+check_server() {
+  if systemctl is-active codex-cloud-server >/dev/null 2>&1; then
+    echo pass
+  else
+    echo fail
+  fi
+}
+
+check_worker() {
+  if systemctl is-active codex-cloud-worker >/dev/null 2>&1; then
+    echo pass
+  else
+    echo fail
+  fi
+}
+
+check_health() {
+  OUT=$(curl -s --max-time 5 "$HEALTH_URL" || true)
+  if [ -z "$OUT" ]; then echo fail; return; fi
+  if printf '%s' "$OUT" | HEALTH_JSON="$OUT" python3 -c '
+import json, os, sys
+try:
+    d = json.loads(os.environ["HEALTH_JSON"])
+except Exception:
+    sys.exit(1)
+sys.exit(0 if (d.get("ok") is True and d.get("backend") == "postgres"
+                and d.get("db_ok") is True
+                and d.get("worker_ok") is True) else 1)
+' 2>/dev/null; then
+    echo pass
+  else
+    echo fail
+  fi
+}
+
+check_postgres() {
+  if command -v pg_isready >/dev/null 2>&1 \
+    && pg_isready -h 127.0.0.1 -p 5432 >/dev/null 2>&1; then
+    echo pass
+  else
+    echo fail
+  fi
+}
+
+check_docker_postgres() {
+  if command -v docker >/dev/null 2>&1 \
+    && docker ps --format '{{.Names}}' 2>/dev/null \
+      | grep -qx 'codex-postgres'; then
+    echo pass
+  else
+    echo fail
+  fi
+}
+
+check_disk() {
+  USE=$(df -P "$DISK_PATH" 2>/dev/null | awk 'NR==2 {print $5}' \
+    | tr -d '% ' || true)
+  case "$USE" in
+    ''|*[!0-9]*) echo fail ;;
+    *) if [ "$USE" -lt 90 ]; then echo pass; else echo fail; fi ;;
+  esac
+}
+
+check_app_dir() {
+  if [ -d "$APP_DIR" ]; then echo pass; else echo fail; fi
+}
+
+run_checks() {
+  S=$(check_server); W=$(check_worker); H=$(check_health)
+  P=$(check_postgres); D=$(check_docker_postgres)
+  K=$(check_disk); A=$(check_app_dir)
+  OVERALL=pass
+  for v in "$S" "$W" "$H" "$P" "$D" "$K" "$A"; do
+    if [ "$v" != "pass" ]; then OVERALL=fail; break; fi
+  done
+  printf '{"overall": "%s", "checks": {"server": "%s", "worker": "%s", "health": "%s", "postgres": "%s", "docker_postgres": "%s", "disk": "%s", "app_dir": "%s"}}\n' \
+    "$OVERALL" "$S" "$W" "$H" "$P" "$D" "$K" "$A"
+  if [ "$OVERALL" = "pass" ]; then return 0; else return 1; fi
+}
+
+if [ "${1:-}" = "--self-test" ]; then
+  # 起一个 canned health server：只返回固定健康 JSON，不读任何真实服务。
+  SELF_PORT=18999
+  python3 - "$SELF_PORT" <<'PY' &
+import http.server
+import json
+import sys
+
+
+class H(http.server.BaseHTTPRequestHandler):
+    def do_GET(self):
+        body = json.dumps({
+            "ok": True, "service": "cloud-control-plane",
+            "backend": "postgres", "db_ok": True,
+            "worker_last_seen_seconds_ago": 1.0, "worker_ok": True,
+        }).encode()
+
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *a):
+        pass
+
+
+http.server.HTTPServer(("127.0.0.1", int(sys.argv[1])), H).serve_forever()
+PY
+  SERVER_PID=$!
+  trap 'kill $SERVER_PID 2>/dev/null || true' EXIT
+  sleep 1
+  MONITOR_HEALTH_URL="http://127.0.0.1:$SELF_PORT/health" \
+    HEALTH_URL="http://127.0.0.1:$SELF_PORT/health" \
+    sh "$0" > /tmp/monitor-self-test.json 2>/dev/null || true
+  # 注意：上面那行是跑完整检查，health 以外项在非主机上 fail 是正常的；
+  # self-test 只断言两件事：输出是合法 JSON，且 health 解析为 pass。
+  if python3 -c '
+import json
+import sys
+with open("/tmp/monitor-self-test.json") as f:
+    d = json.load(f)
+sys.exit(0 if d["checks"]["health"] == "pass" else 1)
+'; then
+    echo "SELF-TEST PASS：输出是合法 JSON 且 health 解析为 pass（其它项在非主机上 fail 属正常，不影响 self-test）"
+    cat /tmp/monitor-self-test.json
+    exit 0
+  else
+    echo "SELF-TEST FAIL：health 解析未通过或输出不是合法 JSON" >&2
+    cat /tmp/monitor-self-test.json >&2 || true
+    exit 1
+  fi
+fi
+
+run_checks

deploy/monitor.timer

@@ -0,0 +1,12 @@
+[Unit]
+Description=Run Codex cloud monitor check every 5 minutes
+
+[Timer]
+# 每 5 分钟跑一次 monitor.sh；Persistent=true 表示主机停机错过的那次
+# 会在下次开机时补跑一次，不会静默丢检查。
+OnCalendar=*:0/5
+Persistent=true
+Unit=monitor.service
+
+[Install]
+WantedBy=timers.target

deploy/restore-drill.sh

@@ -0,0 +1,106 @@
+#!/bin/sh
+# 备份恢复演练：主机侧运行（root，或能读 /etc/codex-app-server/env 的用户）。
+#
+# 用法（主机上）：sh /opt/codex-app-server/deploy/restore-drill.sh
+#   可选环境变量：BACKUP_DIR（默认 /var/lib/codex-app-server/backups）。
+#
+# 步骤：
+#   1. 跑一次 backup_postgres 生成新鲜备份（.sql）。
+#   2. 新建一个测试库 codex_control_restore_test_<时间戳>，用
+#      cloud.backup.restore_postgres 把备份恢复进去。
+#   3. 比对生产库与测试库的 tasks / projects / users / audit_log 行数，
+#      行数相等即通过；只比行数，不比对具体内容，不打印任何行内容。
+#   4. 比对完立即删掉测试库（trap 保证即使中途失败也删）。
+#
+# 安全红线（写进脚本逻辑里，不是只写在注释里）：
+#   - 恢复目标永远是新生成的测试库名，绝不等于生产库名；restore_postgres
+#     本身也会拒绝目标库名等于 CLOUD_DB_NAME（allow_production 绝不在
+#     本脚本里设为 True），双重保险。
+#   - 本脚本绝不恢复到生产库，绝不删生产库；删的只会是本次生成的测试库。
+#   - 全程只打印 PASS/FAIL 与行数，不打印密码，不打印备份内容。
+#   - 本脚本以 root 身份读 env 文件是为了拿到数据库连接参数与密码，
+#     密码只经 PGPASSWORD 环境变量传给 psql/python 子进程，绝不出现在
+#     命令行参数里，也绝不被 echo 出来。
+set -eu
+
+APP_DIR=${APP_DIR:-/opt/codex-app-server}
+BACKUP_DIR=${BACKUP_DIR:-/var/lib/codex-app-server/backups}
+ENV_FILE=${ENV_FILE:-/etc/codex-app-server/env}
+
+if [ ! -f "$ENV_FILE" ]; then
+  echo "FAIL：找不到 env 文件：$ENV_FILE" >&2
+  exit 1
+fi
+# shellcheck disable=SC1090
+set -a
+. "$ENV_FILE"
+set +a
+if [ -z "${CLOUD_DB_NAME:-}" ] || [ -z "${CLOUD_DB_USER:-}" ]; then
+  echo "FAIL：env 文件里缺少 CLOUD_DB_NAME 或 CLOUD_DB_USER" >&2
+  exit 1
+fi
+export PGPASSWORD="${CLOUD_DB_PASSWORD:-}"
+HOST_ARG=${CLOUD_DB_HOST:-127.0.0.1}
+PORT_ARG=${CLOUD_DB_PORT:-5432}
+
+STAMP=$(date +%Y%m%d%H%M%S)
+TEST_DB="codex_control_restore_test_$STAMP"
+
+drop_test_db() {
+  # trap 保证：无论演练成功失败，测试库都会被删掉，不留垃圾库。
+  psql -h "$HOST_ARG" -p "$PORT_ARG" -U "$CLOUD_DB_USER" \
+    -d postgres -c "DROP DATABASE IF EXISTS \"$TEST_DB\"" >/dev/null 2>&1 || true
+}
+trap drop_test_db EXIT
+
+# 1. 新鲜备份（backup_postgres 的连接参数与密码都走环境变量，
+#    密码只经 PGPASSWORD 传给子进程，绝不出现在命令行或输出里）。
+mkdir -p "$BACKUP_DIR"
+BACKUP_FILE=$(python3 - "$APP_DIR" "$BACKUP_DIR" <<'PY'
+import sys
+sys.path.insert(0, sys.argv[1])
+from cloud.backup import backup_postgres
+print(backup_postgres(sys.argv[2]))
+PY
+)
+if [ ! -f "$BACKUP_FILE" ]; then
+  echo "FAIL：备份生成失败" >&2
+  exit 1
+fi
+echo "备份已生成（只打印路径，不打印内容）：$BACKUP_FILE"
+
+# 2. 建测试库并恢复
+psql -h "$HOST_ARG" -p "$PORT_ARG" -U "$CLOUD_DB_USER" \
+  -d postgres -c "CREATE DATABASE \"$TEST_DB\"" >/dev/null
+python3 - "$APP_DIR" "$BACKUP_FILE" "$TEST_DB" <<'PY'
+import sys
+sys.path.insert(0, sys.argv[1])
+from cloud.backup import restore_postgres
+# allow_production 故意不传（默认 False）：目标库名绝不能等于生产库名。
+print(restore_postgres(sys.argv[2], sys.argv[3]))
+PY
+echo "恢复完成，目标测试库：$TEST_DB"
+
+# 3. 行数比对（只比行数，不打印行内容）
+FAIL=0
+for TABLE in tasks projects users audit_log; do
+  PROD_N=$(psql -h "$HOST_ARG" -p "$PORT_ARG" -U "$CLOUD_DB_USER" \
+    -d "$CLOUD_DB_NAME" -t -A -c "SELECT COUNT(*) FROM $TABLE" | tr -d ' \n')
+  REST_N=$(psql -h "$HOST_ARG" -p "$PORT_ARG" -U "$CLOUD_DB_USER" \
+    -d "$TEST_DB" -t -A -c "SELECT COUNT(*) FROM $TABLE" | tr -d ' \n')
+  if [ "$PROD_N" = "$REST_N" ]; then
+    echo "PASS $TABLE：生产库=$PROD_N 恢复库=$REST_N"
+  else
+    echo "FAIL $TABLE：生产库=$PROD_N 恢复库=$REST_N" >&2
+    FAIL=1
+  fi
+done
+
+# 4. 删测试库由 trap 执行；这里先显式删一次，trap 再删是空操作。
+drop_test_db
+trap - EXIT
+if [ "$FAIL" = "1" ]; then
+  echo "恢复演练 FAIL：有表行数不一致" >&2
+  exit 1
+fi
+echo "恢复演练 PASS：全部 4 张表行数一致，测试库已删除"

deploy/rollback.sh

@@ -0,0 +1,79 @@
+#!/bin/sh
+# 回滚：VM 侧运行，把主机恢复到最近一份代码备份包并重启验证。
+#
+# 用法：deploy/rollback.sh <host> [user]
+#   host：主机 IP 或域名（必填）；user：SSH 用户（默认 root）。
+#
+# 密码：与 deploy.sh 同款规则——绝不接受密码参数、不读 env 文件、
+# 不打印任何密钥；SSH 密码由交互提示或调用方通道提供；需要代理时
+# 用环境变量 SSH_PROXY_CMD 传入 ProxyCommand，未设则直连。
+#
+# 回滚边界：只回代码，不回数据库（理由见 deploy.sh 头注释与
+# deploy/README.md：数据库迁移是只加列的幂等操作，老代码仍能读新列）。
+# 备份包来自 deploy.sh 每次部署前在主机上打的
+# /opt/codex-app-server-backups/codex-app-server-<时间戳>.tar.gz，
+# 本脚本恢复其中最新的一份；主机上没有备份包时直接报错退出，不编造。
+set -eu
+
+HOST=${1:-}
+USER_ARG=${2:-root}
+if [ -z "$HOST" ]; then
+  echo "用法：deploy/rollback.sh <host> [user]" >&2
+  exit 2
+fi
+
+ssh_cmd() {
+  if [ -n "${SSH_PROXY_CMD:-}" ]; then
+    ssh -o "ProxyCommand $SSH_PROXY_CMD" -o StrictHostKeyChecking=no \
+      -o UserKnownHostsFile=/dev/null "$@"
+  else
+    ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null "$@"
+  fi
+}
+
+ssh_cmd "$USER_ARG@$HOST" 'sh -s' <<'HOST_SCRIPT'
+set -eu
+APP_DIR=/opt/codex-app-server
+BACKUP_DIR=/opt/codex-app-server-backups
+
+check_health() {
+  OUT=$(curl -s --max-time 5 http://127.0.0.1:8765/health || true)
+  [ -n "$OUT" ] && printf '%s' "$OUT" | python3 -c '
+import json, sys
+try:
+    d = json.load(sys.stdin)
+except Exception:
+    sys.exit(1)
+sys.exit(0 if (d.get("ok") is True and d.get("backend") == "postgres"
+                and d.get("db_ok") is True) else 1)
+'
+}
+
+wait_health() {
+  i=0
+  while [ "$i" -lt 12 ]; do
+    if check_health; then return 0; fi
+    sleep 5
+    i=$((i + 1))
+  done
+  return 1
+}
+
+BACKUP_FILE=$(ls -t "$BACKUP_DIR"/codex-app-server-*.tar.gz 2>/dev/null \
+  | head -n 1 || true)
+if [ -z "$BACKUP_FILE" ]; then
+  echo "主机上没有代码备份包，无法回滚：$BACKUP_DIR" >&2
+  exit 1
+fi
+echo "将恢复备份包：$BACKUP_FILE"
+systemctl stop codex-cloud-worker codex-cloud-server || true
+tar -xzf "$BACKUP_FILE" -C /opt
+chown -R codexsvc:codexsvc "$APP_DIR"
+systemctl start codex-cloud-server codex-cloud-worker
+if wait_health; then
+  echo "回滚成功，health 通过，当前代码为备份包：$BACKUP_FILE"
+  exit 0
+fi
+echo "回滚后 health 仍失败，需人工介入" >&2
+exit 1
+HOST_SCRIPT

deploy/verify-multiuser.py

@@ -0,0 +1,260 @@
+#!/usr/bin/env python3
+"""多用户真实验证：主机侧以 codexsvc 身份运行，直接对生产库 + 本地 HTTP API 验证。
+
+用法（主机上）：
+    python3 /opt/codex-app-server/deploy/verify-multiuser.py [--base URL] [--db PATH]
+    默认 --base http://127.0.0.1:8765，--db /var/lib/codex-app-server/control.db
+    （Postgres 后端时 db 参数只作占位，连接参数走 CLOUD_DB_* 环境变量，
+    与 server/worker 一致；SQLite 后端时就是库文件路径）。
+
+安全与防误伤（写进逻辑里，不是只写在注释里）：
+- 测试用户、项目、任务全部用 multiuser-test- 前缀（项目 id 固定为
+  multiuser-test-p1 / multiuser-test-p2，用户按用户名 multiuser-test-*
+  查找），清理时只删这些前缀的行，绝不碰其它生产数据。
+- 测试任务创建时都带一个永不存在的依赖 multiuser-test-never-completes：
+  worker 只派依赖全部 completed 的 queued 任务（见 cloud/queue.py 的
+  ready_tasks），所以这些测试任务永远不会被生产 worker 认领去跑模型——
+  不花模型钱、不改任何仓库。验证完任务会被删掉，依赖本身只是一个
+  字符串，不对应任何真实任务行。
+- API key 只存在内存里：创建用户时 AuthStore 返回的明文 key 只保存在
+  本进程的变量里，用完即弃；脚本全程只打印每项 PASS/FAIL 与总数，
+  绝不打印任何 API key；任务 prompt、用户 id 等内容也不打印，只打印
+  检查名与结果。
+- 审计日志是 append-only 的：验证过程产生的审计条目（建项目、设成员、
+  建任务、merge 等）会留在 audit_log 里，脚本不删审计表——这是故意的，
+  审计就该留痕；README 里也写清了这一点。
+- 清理失败时脚本退出非零并列出残留测试 id（不含 key），不静默假装干净。
+- 本脚本不跑真实模型、不推代码、不改生产项目与任务，只动 multiuser-test-*
+  的行；跑之前不需要停 worker（见上，测试任务不会被认领）。
+
+检查项（预期值写死在断言里，实际不符即 FAIL）：
+  1. owner 建 P1、P2 成功（201）
+  2. owner 建任务成功（201，contributor 建任务也成功）
+  3. viewer 建任务 403；非成员（viewer 对 P2）建任务 403
+  4. viewer 读 P1 任务成功（200）
+  5. viewer 读 P2 任务 403（跨项目隔离：P1 成员不是 P2 成员）
+  6. maintainer 对 completed 测试任务调 merge 成功（200，状态变 merged）
+  7. contributor 对 completed 测试任务调 merge 403
+  8. maintainer 对 queued 测试任务调 merge 409（状态不对）
+  9. 无 key 调 API 401
+"""
+
+import argparse
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from cloud.auth import AuthStore  # noqa: E402
+from cloud.db import connect as db_connect  # noqa: E402
+
+PREFIX = "multiuser-test-"
+P1 = PREFIX + "p1"
+P2 = PREFIX + "p2"
+NEVER_DEP = PREFIX + "never-completes"
+
+RESULTS = []
+
+
+def check(name, ok):
+    RESULTS.append((name, bool(ok)))
+    print("%s %s" % ("PASS" if ok else "FAIL", name))
+
+
+def req(base, method, path, key=None, body=None):
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    r = urllib.request.Request(base + path, data=data, method=method)
+    if key is not None:
+        r.add_header("X-API-Key", key)
+    if body is not None:
+        r.add_header("Content-Type", "application/json")
+    try:
+        with urllib.request.urlopen(r, timeout=15) as resp:
+            raw = resp.read()
+            return resp.status, (json.loads(raw) if raw else {})
+    except urllib.error.HTTPError as e:
+        raw = e.read()
+        try:
+            payload = json.loads(raw) if raw else {}
+        except ValueError:
+            payload = {}
+        return e.code, payload
+    except Exception as e:
+        return None, {"error": str(e)}
+
+
+def main():
+    ap = argparse.ArgumentParser(description="multiuser verification")
+    ap.add_argument("--base", default="http://127.0.0.1:8765")
+    ap.add_argument("--db", default="/var/lib/codex-app-server/control.db")
+    args = ap.parse_args()
+    base = args.base.rstrip("/")
+
+    auth = AuthStore(args.db)
+    created_task_ids = []
+    created_user_ids = []
+    try:
+        # -- 预清理：上一轮残留的测试行（可重复跑）-------------------
+        conn = db_connect(args.db)
+        try:
+            conn.execute(
+                "DELETE FROM tasks WHERE project_id LIKE ?",
+                (PREFIX + "%",))
+            conn.execute(
+                "DELETE FROM project_members WHERE project_id LIKE ?",
+                (PREFIX + "%",))
+            conn.execute(
+                "DELETE FROM projects WHERE id LIKE ?", (PREFIX + "%",))
+            conn.execute(
+                "DELETE FROM users WHERE username LIKE ?", (PREFIX + "%",))
+            conn.commit()
+        finally:
+            conn.close()
+
+        # -- 建 4 个测试用户（key 只在内存里）------------------------
+        users = {}
+        for role in ("owner", "maintainer", "contributor", "viewer"):
+            uid, key = auth.create_user(PREFIX + role)
+            users[role] = (uid, key)
+            created_user_ids.append(uid)
+
+        # -- 建 2 个测试项目（owner 建，P2 只有 owner 是成员）--------
+        s, p1 = req(base, "POST", "/projects",
+                    users["owner"][1], {"id": P1, "name": P1})
+        check("owner 创建项目 P1（201）", s == 201)
+        s, p2 = req(base, "POST", "/projects",
+                    users["owner"][1], {"id": P2, "name": P2})
+        check("owner 创建项目 P2（201）", s == 201)
+        if s != 201 or p1.get("id") != P1:
+            print("项目创建失败，无法继续验证")
+            return 1
+        for role in ("maintainer", "contributor", "viewer"):
+            auth.set_member(P1, users["owner"][0], users[role][0], role)
+
+        # -- 建任务权限 ----------------------------------------------
+        # 全部测试任务带永不满足的依赖，生产 worker 绝不会认领（见模块注释）。
+        s, t_owner = req(base, "POST", "/tasks", users["owner"][1], {
+            "project_id": P1, "prompt": "multiuser test task",
+            "depends_on": [NEVER_DEP]})
+        check("owner 在 P1 建任务（201）", s == 201)
+        if s == 201:
+            created_task_ids.append(t_owner["id"])
+        s, t_contrib = req(base, "POST", "/tasks",
+                           users["contributor"][1], {
+                               "project_id": P1, "prompt": "multiuser test task",
+                               "depends_on": [NEVER_DEP]})
+        check("contributor 在 P1 建任务（201）", s == 201)
+        if s == 201:
+            created_task_ids.append(t_contrib["id"])
+        s, _ = req(base, "POST", "/tasks", users["viewer"][1], {
+            "project_id": P1, "prompt": "multiuser test task"})
+        check("viewer 在 P1 建任务（403）", s == 403)
+        s, _ = req(base, "POST", "/tasks", users["viewer"][1], {
+            "project_id": P2, "prompt": "multiuser test task"})
+        check("非成员 viewer 在 P2 建任务（403）", s == 403)
+
+        # -- 读任务权限与跨项目隔离 ----------------------------------
+        if created_task_ids:
+            tid = created_task_ids[0]
+            s, _ = req(base, "GET", "/tasks/" + tid, users["viewer"][1])
+            check("viewer 读 P1 任务（200）", s == 200)
+        s, t_p2 = req(base, "POST", "/tasks", users["owner"][1], {
+            "project_id": P2, "prompt": "multiuser test task",
+            "depends_on": [NEVER_DEP]})
+        check("owner 在 P2 建任务（201）", s == 201)
+        if s == 201:
+            created_task_ids.append(t_p2["id"])
+            s, _ = req(base, "GET", "/tasks/" + t_p2["id"],
+                       users["viewer"][1])
+            check("P1 成员 viewer 读 P2 任务（403，跨项目隔离）", s == 403)
+
+        # -- merge 权限（需先把测试任务直接置 completed，不跑模型）--
+        conn = db_connect(args.db)
+        try:
+            for tid in created_task_ids:
+                conn.execute(
+                    "UPDATE tasks SET status = 'completed' "
+                    "WHERE id = ?", (tid,))
+            conn.commit()
+        finally:
+            conn.close()
+        if len(created_task_ids) >= 1:
+            s, merged = req(base, "POST",
+                            "/tasks/" + created_task_ids[0] + "/merge",
+                            users["maintainer"][1])
+            check("maintainer 对 completed 任务调 merge（200 且变 merged）",
+                  s == 200 and merged.get("status") == "merged")
+        if len(created_task_ids) >= 2:
+            s, _ = req(base, "POST",
+                       "/tasks/" + created_task_ids[1] + "/merge",
+                       users["contributor"][1])
+            check("contributor 对 completed 任务调 merge（403）", s == 403)
+        # queued 任务调 merge 应为 409：再建一个（不置 completed）。
+        s, t_queued = req(base, "POST", "/tasks", users["owner"][1], {
+            "project_id": P1, "prompt": "multiuser test task",
+            "depends_on": [NEVER_DEP]})
+        if s == 201:
+            created_task_ids.append(t_queued["id"])
+            s, _ = req(base, "POST",
+                       "/tasks/" + t_queued["id"] + "/merge",
+                       users["maintainer"][1])
+            check("maintainer 对 queued 任务调 merge（409）", s == 409)
+
+        # -- 无 key ---------------------------------------------------
+        s, _ = req(base, "GET", "/tasks/" + created_task_ids[0]
+                   if created_task_ids else "/tasks/none")
+        check("无 key 调 API（401）", s == 401)
+    finally:
+        # -- 清理：只删 multiuser-test-* 的行，审计表不动 ------------
+        leftover = []
+        try:
+            conn = db_connect(args.db)
+            try:
+                for tid in created_task_ids:
+                    conn.execute("DELETE FROM tasks WHERE id = ?", (tid,))
+                conn.execute(
+                    "DELETE FROM tasks WHERE project_id LIKE ?",
+                    (PREFIX + "%",))
+                conn.execute(
+                    "DELETE FROM project_members WHERE project_id LIKE ?",
+                    (PREFIX + "%",))
+                conn.execute(
+                    "DELETE FROM projects WHERE id LIKE ?", (PREFIX + "%",))
+                conn.execute(
+                    "DELETE FROM users WHERE username LIKE ?",
+                    (PREFIX + "%",))
+                conn.commit()
+                # 校验残留（只查测试前缀的行，不打印 key 或内容）。
+                for table, col in (("tasks", "project_id"),
+                                   ("projects", "id"),
+                                   ("users", "username")):
+                    row = conn.execute(
+                        "SELECT COUNT(*) FROM %s WHERE %s LIKE ?"
+                        % (table, col), (PREFIX + "%",)).fetchone()
+                    if row and row[0]:
+                        leftover.append("%s:%d" % (table, row[0]))
+            finally:
+                conn.close()
+        except Exception as e:
+            leftover.append("cleanup-error:%s" % e)
+        try:
+            auth.conn.close()
+        except Exception:
+            pass
+        if leftover:
+            print("清理未完成，残留测试行：%s" % ", ".join(leftover))
+            print("注意：审计日志里的测试条目是故意留下的，不算残留。")
+
+    passed = sum(1 for _, ok in RESULTS if ok)
+    total = len(RESULTS)
+    print("多用户验证：%d/%d 通过" % (passed, total))
+    if leftover:
+        return 1
+    return 0 if passed == total else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())