From 64995fafab5c780d38107ced4b47b94d25a58c6e Mon Sep 17 00:00:00 2001
From: host <host@test.local>
Date: Tue, 27 Oct 2026 15:00:25 +0000
Subject: [PATCH] P2-ops: migration alignment, deploy test list, monitor
 alerts/single-host, backup/rollback closure, Caddy template

- migrations: MIGRATION_TABLES now equals POSTGRES_SCHEMA tables
  (uploads, task_attachments, task_followups, environments added);
  SEQUENCE_TABLES matches IDENTITY tables; task_followups
  sender_task_id column ensured before insert; UNMIGRATED_TABLES
  and migration_coverage() report tables with no PG schema instead
  of silently skipping them; CLI prints the coverage report.
- deploy: DEFAULT_TEST_SUITES gains deploy/tests (discover -s only,
  never package path); regression tests pin the default list.
- monitor: MONITOR_SINGLE_HOST=1 makes the server check single-host;
  default stays dual to avoid silent downgrade; server_mode is in
  the JSON. Optional fail-only alert outlet (webhook or mail,
  default off, alert body is pass/fail JSON only, no secrets).
- backup/rollback: README and restore-drill document what is
  recoverable (DB, code) vs not (worktrees, traces, uploads, host
  configs); rollback.sh shares the server-unit list and dual-port
  health success standard with host_deploy.sh (sync test guards it).
- public entry: Caddyfile uses native {$DOMAIN}/{$ACME_EMAIL}
  variables plus a caddy.service template; README documents the
  steps and the current BT-nginx production status.
---
 cloud/migrate_sqlite_to_postgres.py |  16 +-
 cloud/migrations.py                 | 125 +++++++++++-
 cloud/tests/test_p2_migrations.py   | 203 +++++++++++++++++++
 deploy/README.md                    | 114 +++++++++--
 deploy/caddy/Caddyfile              |  26 ++-
 deploy/caddy/caddy.service          |  24 +++
 deploy/deploy.py                    |  11 +-
 deploy/host_deploy.sh               |   2 +
 deploy/monitor.service              |   6 +-
 deploy/monitor.sh                   |  74 ++++++-
 deploy/restore-drill.sh             |   6 +
 deploy/rollback.sh                  |  67 +++++--
 deploy/tests/test_deploy.py         |  27 +++
 deploy/tests/test_p2_ops.py         | 296 ++++++++++++++++++++++++++++
 14 files changed, 953 insertions(+), 44 deletions(-)
 create mode 100644 cloud/tests/test_p2_migrations.py
 create mode 100644 deploy/caddy/caddy.service
 create mode 100644 deploy/tests/test_p2_ops.py

diff --git a/cloud/migrate_sqlite_to_postgres.py b/cloud/migrate_sqlite_to_postgres.py
index a1fbee4..598d114 100644
--- a/cloud/migrate_sqlite_to_postgres.py
+++ b/cloud/migrate_sqlite_to_postgres.py
@@ -15,7 +15,10 @@
 import argparse
 import sys
 
-from cloud.migrations import migrate_sqlite_to_postgres
+from cloud.migrations import (
+    migrate_sqlite_to_postgres,
+    migration_coverage,
+)
 
 
 def main(argv=None):
@@ -30,6 +33,17 @@ def main(argv=None):
         print("%s: %d" % (table, count))
         total += count
     print("total inserted: %d" % total)
+    # 覆盖报告：只含表名，不含行内容与任何密钥。missing 非空
+    # 即清单与 schema 对齐回归；unmigrated 是目标库无此表、
+    # 本迁移无法迁移的表，须另行处理，不能读成“已迁完”。
+    coverage = migration_coverage()
+    if coverage["missing_from_migration"]:
+        print("WARNING missing_from_migration: %s"
+              % ", ".join(coverage["missing_from_migration"]))
+    print("unmigrated (no PG schema, NOT migrated): %s"
+          % ", ".join(coverage["unmigrated_no_pg_schema"]))
+    print("skipped by design: %s"
+          % ", ".join(coverage["skipped_by_design"]))
     return 0
 
 
diff --git a/cloud/migrations.py b/cloud/migrations.py
index a6ae120..468de77 100644
--- a/cloud/migrations.py
+++ b/cloud/migrations.py
@@ -7,9 +7,22 @@
 - schema 来自 cloud.db.POSTGRES_SCHEMA / init_postgres_schema，
   本模块不另发明表结构；迁移前先建表（若目标库已有表，
   CREATE TABLE IF NOT EXISTS 是 no-op）。
+- P2-ops 对齐：MIGRATION_TABLES 必须等于 POSTGRES_SCHEMA 的
+  表集合（见 migration_coverage 与 test_p2_migrations.py）。
+  以前漏掉的 uploads、task_attachments、task_followups、
+  environments 已补进清单；补进时发现 task_followups 在
+  SQLite 侧比 PG schema 多一列 sender_task_id（peer_messages
+  的运行时 ensure_followups_table 会在 PG 上 ALTER 补列，
+  POSTGRES_SCHEMA 里没有它），迁移时按同样方式先补列再插，
+  否则整批迁移会在该表上撞未知列失败。
 - kb_fts（SQLite FTS5 虚拟表）不迁移：Postgres 没有 FTS5，
   KBStore 在 PG 上走 LIKE 分支，kb_docs 的 content 已足够。
 - sqlite_sequence（AUTOINCREMENT 内部表）不迁移。
+- UNMIGRATED_TABLES：SQLite 各 store 建出、但 POSTGRES_SCHEMA
+  里没有的表（api_keys、评审、自动化、GitHub App、网关 key
+  等）。它们不是“被跳过”，而是本模块无法迁移——迁移报告
+  （migration_coverage / CLI 输出）逐项列出，调用方不得
+  把“迁移成功”读成“全部数据都过去了”。
 - 不打印、不返回任何密码或密钥：函数只返回 {表: inserted}，
   CLI 只打印表名与行数。
 - Postgres 连接参数只走 CLOUD_DB_* 环境变量（见 cloud.db），
@@ -17,6 +30,7 @@
 """
 
 import os
+import re
 import sqlite3
 
 from cloud.db import (
@@ -43,10 +57,94 @@ MIGRATION_TABLES = (
     # （本批后无行按默认额度处理，不再是无限额，但显式
     # 管理员额度仍需原样迁过去，不能丢）。
     "user_quotas",
+    # P2-ops 对齐：以下 4 张表在 POSTGRES_SCHEMA 里有、
+    # 原清单里没有，补进后清单与 schema 表集合一致。
+    # uploads / task_attachments：附件包与任务附件；
+    # task_followups：任务间互发消息的后续消息队列；
+    # environments：项目环境（分支/executor）。
+    # 四张表主键均为 TEXT 或复合主键，没有身份列，
+    # 不需要序列拨动（见 SEQUENCE_TABLES）。
+    "uploads",
+    "task_attachments",
+    "task_followups",
+    "environments",
 )
 
 _SKIP_TABLES = frozenset({"kb_fts", "sqlite_sequence"})
 
+# 身份列的表：POSTGRES_SCHEMA 里唯一用
+# ``BIGINT GENERATED BY DEFAULT AS IDENTITY`` 的 4 张表。
+# 迁移时显式插 id 不会推序列，迁完必须 setval 到 MAX(id)，
+# 否则后续省略 id 的 INSERT 会撞主键。新增身份列表必须
+# 同步改这里，test_p2_migrations.py 会拿 schema 实测对齐。
+SEQUENCE_TABLES = (
+    "audit_log",
+    "summaries",
+    "kb_docs",
+    "cost_records",
+)
+
+# SQLite 各 store 建出、但 POSTGRES_SCHEMA 里没有的表：
+# 本迁移模块无法把它们迁到 Postgres（目标库无此表），
+# 不是静默跳过——migration_coverage() 与 CLI 逐项报告。
+# 其中 rate_limit_hits / workers 是易失的限流与心跳数据，
+# 丢了只影响限流窗口与 worker 可见性，不影响业务数据；
+# 其余（api_keys、task_reviews、automations、GitHub App、
+# user_gateway_keys 等）丢了就是真丢，迁库前必须另行
+# 处理（重建/重绑/重发 key），不能只跑本迁移就宣称迁完。
+UNMIGRATED_TABLES = (
+    "api_keys",
+    "task_approvals",
+    "task_reviews",
+    "automations",
+    "automation_runs",
+    "github_app_setup_states",
+    "github_app_config",
+    "github_app_installations",
+    "github_fix_review_dedup",
+    "external_identities",
+    "pr_jobs",
+    "rate_limit_hits",
+    "user_gateway_keys",
+    "webhook_deliveries",
+    "workers",
+)
+
+
+def schema_tables():
+    """POSTGRES_SCHEMA 里的表名集合（只认 CREATE TABLE 语句）。"""
+    names = set()
+    for statement in POSTGRES_SCHEMA:
+        m = re.search(r"CREATE TABLE IF NOT EXISTS (\w+)",
+                      statement)
+        if m:
+            names.add(m.group(1))
+    return names
+
+
+def migration_coverage():
+    """迁移覆盖报告：清单、schema、未迁移项三者一次给全。
+
+    migrated：本次迁移模块会迁的表（= MIGRATION_TABLES）；
+    schema_tables：POSTGRES_SCHEMA 里的表；
+    missing_from_migration：schema 里有、清单里没有的表
+        （本批后应为空；非空即对齐回归，调用方应报错
+        而不是继续跑迁移）；
+    unmigrated_no_pg_schema：UNMIGRATED_TABLES（目标库
+        无此表，无法迁移，须另行处理）；
+    skipped_by_design：kb_fts / sqlite_sequence（设计上不迁）。
+    不含任何密钥或行内容，只含表名。
+    """
+    schema = schema_tables()
+    return {
+        "migrated": tuple(MIGRATION_TABLES),
+        "schema_tables": tuple(sorted(schema)),
+        "missing_from_migration": tuple(
+            sorted(schema - set(MIGRATION_TABLES))),
+        "unmigrated_no_pg_schema": tuple(UNMIGRATED_TABLES),
+        "skipped_by_design": tuple(sorted(_SKIP_TABLES)),
+    }
+
 
 def migrate_sqlite_to_postgres(sqlite_path, pg_conn=None, tables=None):
     """迁移并返回 {表名: 本次插入行数}。
@@ -83,6 +181,26 @@ def migrate_sqlite_to_postgres(sqlite_path, pg_conn=None, tables=None):
                 continue
             columns = [d[0] for d in src.execute(
                 'SELECT * FROM "%s" LIMIT 0' % table).description]
+            if table == "task_followups" and "sender_task_id" in columns:
+                # SQLite 侧有、POSTGRES_SCHEMA 里没有的列：
+                # 与 peer_messages.ensure_followups_table 同款，
+                # 先查现有列、缺了再 ALTER 补列，已有列是 no-op。
+                # PRAGMA 经适配层翻译成 information_schema 查询，
+                # 取 row[1] 即列名（与 init_postgres_schema 同款）。
+                # 补列失败也不在这里吞掉迁移：INSERT 时若列仍
+                # 缺失会直接报错（大声失败，绝不静默丢列）。
+                try:
+                    existing = {row[1] for row in pg_conn.execute(
+                        "PRAGMA table_info(task_followups)").fetchall()}
+                except Exception:
+                    existing = set()
+                if "sender_task_id" not in existing:
+                    try:
+                        pg_conn.execute(
+                            "ALTER TABLE task_followups"
+                            " ADD COLUMN sender_task_id TEXT")
+                    except Exception:
+                        pass
             placeholders = ", ".join("?" for _ in columns)
             col_list = ", ".join('"%s"' % c for c in columns)
             inserted = 0
@@ -96,8 +214,7 @@ def migrate_sqlite_to_postgres(sqlite_path, pg_conn=None, tables=None):
                     inserted += cur.rowcount
             # 身份列的序列不会因显式插 id 自动前进：迁完把序列
             # 拨到当前最大 id，否则后续省略 id 的 INSERT 会撞主键。
-            if table in ("audit_log", "summaries", "kb_docs",
-                         "cost_records") and "id" in columns:
+            if table in SEQUENCE_TABLES and "id" in columns:
                 try:
                     pg_conn.execute(
                         "SELECT setval(pg_get_serial_sequence(?, ?),"
@@ -120,5 +237,9 @@ def migrate_sqlite_to_postgres(sqlite_path, pg_conn=None, tables=None):
 
 __all__ = [
     "MIGRATION_TABLES",
+    "SEQUENCE_TABLES",
+    "UNMIGRATED_TABLES",
+    "migration_coverage",
     "migrate_sqlite_to_postgres",
+    "schema_tables",
 ]
diff --git a/cloud/tests/test_p2_migrations.py b/cloud/tests/test_p2_migrations.py
new file mode 100644
index 0000000..56ddc38
--- /dev/null
+++ b/cloud/tests/test_p2_migrations.py
@@ -0,0 +1,203 @@
+"""P2-ops 迁移对齐：迁移清单必须等于 Postgres schema 表集合，
+未迁移项必须被显式报告，绝不能静默跳过。
+
+不用真实 Postgres：迁移本体用一个假 PG 连接跑（只记录
+INSERT / ALTER，不连库）；schema 对齐直接读 POSTGRES_SCHEMA。
+真实 PG 上的幂等性仍由 test_postgres_integration.py 覆盖
+（有 CLOUD_TEST_PG_DSN 时才跑）。
+"""
+import os
+import re
+import sqlite3
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.db import POSTGRES_SCHEMA
+    from cloud.migrations import (
+        MIGRATION_TABLES,
+        SEQUENCE_TABLES,
+        UNMIGRATED_TABLES,
+        migrate_sqlite_to_postgres,
+        migration_coverage,
+        schema_tables,
+    )
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.db import POSTGRES_SCHEMA
+    from cloud.migrations import (
+        MIGRATION_TABLES,
+        SEQUENCE_TABLES,
+        UNMIGRATED_TABLES,
+        migrate_sqlite_to_postgres,
+        migration_coverage,
+        schema_tables,
+    )
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+class FakeCursor:
+    def __init__(self, rows=None, rowcount=1):
+        self._rows = rows or []
+        self.rowcount = rowcount
+
+    def fetchall(self):
+        return self._rows
+
+
+class FakePgConn:
+    """只记录 SQL 的假 PG 连接：init_postgres_schema 的 DDL
+    直接忽略；PRAGMA 按构造时给的列集合回答（row[1] 为列名，
+    与 init_postgres_schema / 迁移补列代码的取法一致）。"""
+
+    def __init__(self, pragma_columns=None):
+        # pragma_columns: {表名: {列名}}；缺省表示“列齐全”。
+        self.pragma_columns = pragma_columns or {}
+        self.statements = []
+        self.inserts = []  # (表名, 列名元组)
+        self.alters = []
+        self.commits = 0
+
+    def execute(self, sql, params=()):
+        self.statements.append(sql)
+        if sql.startswith("PRAGMA table_info("):
+            table = sql[len("PRAGMA table_info("):].rstrip(")")
+            cols = self.pragma_columns.get(table)
+            if cols is None:
+                # 未指定时假装列齐全：projects 含 repo_url，
+                # task_followups 含 sender_task_id。
+                cols = {"repo_url"} if table == "projects" \
+                    else {"sender_task_id"}
+            rows = [(i, c, None) for i, c in enumerate(sorted(cols))]
+            return FakeCursor(rows=rows)
+        m = re.search(r'INSERT INTO "(\w+)" \(([^)]*)\)', sql)
+        if m:
+            cols = tuple(c.strip().strip('"')
+                         for c in m.group(2).split(","))
+            self.inserts.append((m.group(1), cols))
+            return FakeCursor(rowcount=1)
+        if sql.startswith("ALTER TABLE"):
+            self.alters.append(sql)
+            return FakeCursor(rowcount=0)
+        return FakeCursor(rowcount=0)
+
+    def commit(self):
+        self.commits += 1
+
+
+def _make_sqlite(path):
+    conn = sqlite3.connect(path)
+    conn.execute("CREATE TABLE uploads(id TEXT PRIMARY KEY, created_by TEXT,"
+                 " created_at REAL, status TEXT)")
+    conn.execute("CREATE TABLE task_attachments(task_id TEXT, bundle_id TEXT,"
+                 " rel_path TEXT, size INTEGER,"
+                 " PRIMARY KEY(task_id, rel_path))")
+    conn.execute("CREATE TABLE task_followups(id TEXT PRIMARY KEY,"
+                 " task_id TEXT, project_id TEXT, body TEXT, mode TEXT,"
+                 " status TEXT, created_by TEXT, sender_task_id TEXT,"
+                 " created_at REAL, updated_at REAL)")
+    conn.execute("CREATE TABLE environments(id TEXT PRIMARY KEY,"
+                 " project_id TEXT, name TEXT, branch TEXT, executor TEXT,"
+                 " status TEXT, error_text TEXT, repo_url TEXT,"
+                 " created_by TEXT, created_at REAL, updated_at REAL)")
+    conn.execute("INSERT INTO uploads VALUES(?, ?, ?, ?)",
+                 ("up1", "u1", 1.0, "done"))
+    conn.execute("INSERT INTO task_attachments VALUES(?, ?, ?, ?)",
+                 ("t1", "b1", "a.txt", 3))
+    conn.execute("INSERT INTO task_followups VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                 ("f1", "t1", "p1", "hi", "queued", "pending", "u1",
+                  "sender-task", 1.0, 1.0))
+    conn.execute("INSERT INTO environments VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                 ("e1", "p1", "dev", "main", "docker", "ready", "",
+                  "https://example.invalid/r", "u1", 1.0, 1.0))
+    conn.commit()
+    conn.close()
+
+
+class TestMigrationAlignment(unittest.TestCase):
+    def test_migration_tables_equal_schema_tables(self):
+        # 核心对齐断言：以前漏掉的 4 张表补进后，两边集合必须一致。
+        self.assertEqual(set(MIGRATION_TABLES), schema_tables())
+        for table in ("uploads", "task_attachments",
+                      "task_followups", "environments", "user_quotas"):
+            self.assertIn(table, MIGRATION_TABLES)
+
+    def test_sequence_tables_equal_identity_tables(self):
+        # 序列拨动清单必须等于 schema 里 IDENTITY 列的表集合：
+        # TEXT 主键的新表不需要拨序列，漏拨或多拨都会在这里变红。
+        identity = set()
+        for statement in POSTGRES_SCHEMA:
+            m = re.search(r"CREATE TABLE IF NOT EXISTS (\w+)", statement)
+            if m and "GENERATED BY DEFAULT AS IDENTITY" in statement:
+                identity.add(m.group(1))
+        self.assertEqual(set(SEQUENCE_TABLES), identity)
+
+    def test_unmigrated_are_real_sqlite_tables_not_in_schema(self):
+        # 未迁移清单不得是凭空编的：每项都必须在 cloud/ 源码里
+        # 真有 CREATE TABLE，且确实不在 PG schema 里；同时不得
+        # 与迁移清单重叠（重叠即“既迁又报未迁”，自相矛盾）。
+        sources = ""
+        for name in os.listdir(REPO_ROOT / "cloud"):
+            if name.endswith(".py"):
+                with open(REPO_ROOT / "cloud" / name) as f:
+                    sources += f.read()
+        schema = schema_tables()
+        for table in UNMIGRATED_TABLES:
+            self.assertIn("CREATE TABLE IF NOT EXISTS %s" % table, sources,
+                          "%s 不在任何 SQLite DDL 里，清单已过期" % table)
+            self.assertNotIn(table, schema)
+            self.assertNotIn(table, MIGRATION_TABLES)
+
+    def test_coverage_reports_no_missing(self):
+        coverage = migration_coverage()
+        self.assertEqual(coverage["missing_from_migration"], ())
+        self.assertEqual(coverage["unmigrated_no_pg_schema"],
+                         tuple(UNMIGRATED_TABLES))
+        self.assertEqual(set(coverage["skipped_by_design"]),
+                         {"kb_fts", "sqlite_sequence"})
+
+
+class TestMigrateNewTables(unittest.TestCase):
+    def test_new_tables_migrated_and_sender_column_added(self):
+        # PG 侧 task_followups 尚无 sender_task_id：迁移必须先
+        # ALTER 补列，再把含该列的行插过去，而不是撞未知列失败。
+        with tempfile.TemporaryDirectory() as tmp:
+            src = os.path.join(tmp, "src.db")
+            _make_sqlite(src)
+            fake = FakePgConn(pragma_columns={
+                "projects": {"repo_url"},
+                "task_followups": {"id", "task_id", "project_id", "body",
+                                   "mode", "status", "created_by",
+                                   "created_at", "updated_at"},
+            })
+            result = migrate_sqlite_to_postgres(src, pg_conn=fake)
+        for table in ("uploads", "task_attachments",
+                      "task_followups", "environments"):
+            self.assertEqual(result.get(table), 1, table)
+        self.assertTrue(
+            any("task_followups" in a and "sender_task_id" in a
+                for a in fake.alters),
+            "未发出 sender_task_id 补列 ALTER：%s" % fake.alters)
+        fu_inserts = [cols for t, cols in fake.inserts
+                      if t == "task_followups"]
+        self.assertEqual(len(fu_inserts), 1)
+        self.assertIn("sender_task_id", fu_inserts[0])
+
+    def test_sender_column_present_skips_alter(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            src = os.path.join(tmp, "src.db")
+            _make_sqlite(src)
+            fake = FakePgConn()  # 缺省：列齐全
+            migrate_sqlite_to_postgres(src, pg_conn=fake,
+                                       tables=("task_followups",))
+        # init_postgres_schema 自身会给 projects/tasks 补列，
+        # 这里只断言 task_followups 没有重复补列。
+        self.assertEqual(
+            [a for a in fake.alters if "task_followups" in a], [])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/deploy/README.md b/deploy/README.md
index 796e9ea..15e6213 100644
--- a/deploy/README.md
+++ b/deploy/README.md
@@ -104,6 +104,16 @@ deploy/rollback.sh <host> [user]
 恢复后同样重启并验 health（60 秒），成功/失败都明说；主机上没有备份包
 时直接报错退出，不编造。
 
+### 回滚成功标准：清单里每个端口都通过（P2-ops）
+
+rollback.sh 与 host_deploy.sh 共用同一份 unit 清单约定：server-1 必含，
+server-2 以 `systemctl cat` 探测为准（存在才纳入停/起与健康检查），worker
+是 autoscaler active 时用 autoscaler、否则用 cloud-worker。两脚本改动清单
+时必须同步改，`deploy/tests/test_p2_ops.py` 会双向核对（只改一处就变红）。
+
+回滚成功 = 8765 通过，且 server-2 存在时 8766 也通过。只通 8765 不算成功：
+LB 上游的 server-2 若仍跑故障代码，回滚等于没回完。
+
 ### 回滚边界：只回代码，不回数据库
 
 数据库迁移是只加列的幂等操作（例如 projects.repo_url、tasks 的 6 列），
@@ -128,6 +138,9 @@ sh /opt/codex-app-server/deploy/restore-drill.sh
 audit_log 行数（行数相等即通过，只比行数，不比对具体内容，不打印任何
 行内容）→ 比对完立即删掉测试库（trap 保证中途失败也删）。
 
+范围（P2-ops）：演练 PASS 只证明数据库备份可恢复，不证明 worktree /
+链路 / 上传可恢复——那些不在备份里，见“备份制度”一节。
+
 安全红线：恢复目标永远是新生成的测试库名，绝不等于生产库名；
 `restore_postgres` 本身也会拒绝目标库名等于 `CLOUD_DB_NAME`
 （`allow_production` 绝不在本脚本里设为 True），双重保险。本脚本绝不
@@ -165,14 +178,38 @@ journalctl -u monitor.service -f   # 看每次检查的输出
 
 timer 每 5 分钟跑一次，日志进 journal。
 
-### 监控边界：本轮只有“主机侧检查 + journal 日志”
+### 监控边界：默认只有“检查 + journal 日志”，告警出口可选（P2-ops）
 
-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 已启用。
+默认不配告警出口时，monitor.sh 仍然只做检查 + journal 日志，不会发消息
+给任何人。需要 fail 时通知时，在 monitor.service 的 `Environment=`（或
+monitor.timer 的运行环境）里二选一配置，默认关闭：
+
+```ini
+# 二选一，同时配了只用 webhook：
+Environment=MONITOR_ALERT_URL=https://example.invalid/hook
+# 或：
+Environment=MONITOR_ALERT_MAILTO=ops@example.invalid
+```
+
+- 只在 overall=fail 时发，pass 时不发；故障持续期间 timer 每 5 分钟跑
+  一次就会重发，接收端需自行去重（本机不做去重状态文件，免得状态文件
+  本身成为新的故障点）。
+- 告警内容只有 JSON 检查结果（pass/fail 与 server_mode），不含 health
+  正文、不含任何环境变量、不含密钥；URL 与收件人只走环境变量。
+- 邮件路径依赖本机 `mail` 命令，没有它时会记一条“告警未发出”到 journal，
+  不会静默假装已通知。
+
+### 单机模式：MONITOR_SINGLE_HOST=1（P2-ops）
+
+默认 server 检查要求 server-1 与 server-2 双 active。确认是单机部署时，
+在 monitor 运行环境里设 `MONITOR_SINGLE_HOST=1`，server 检查只查
+server-1；否则单机主机的 server 检查会恒 fail，恒 fail 会掩盖真故障。
+双机主机上绝不要设它（设了就是静默降级）。输出 JSON 的 `server_mode`
+字段写明本次是 single 还是 dual，降级可见。
+
+手工验证步骤：主机上跑一次 monitor.sh 看 JSON 与退出码，再跑一次
+--self-test 看 SELF-TEST PASS，最后 `systemctl list-timers | grep monitor`
+确认 timer 已启用。--self-test 时内层检查会清掉告警出口，不会误发告警。
 
 ## worker 心跳与 health 的 worker 字段
 
@@ -225,17 +262,72 @@ completed 测试任务是直接在 DB 里把测试任务置 completed，不跑
 - 审计日志是 append-only 的：验证过程的审计条目会留在 audit_log 里，
   脚本不删审计表——审计就该留痕，这一点不是清理遗漏。
 
-## 公网入口与 webhook secret（已有文件，不属于本轮新增）
+## 备份制度：什么可恢复、什么不可恢复（P2-ops，二选一里选了“写清区分”）
 
-- `caddy/Caddyfile`：只暴露 `/webhooks/*` 与 `/oauth/*`，任务/项目/管理
-  API 对外一律 403；域名与邮箱占位符需部署时替换。
+不要谎称已做全量备份。本节写清现状；worktree / 链路 / 上传目录纳入
+备份范围是另一单（需先定量：租户目录大小、备份窗口、恢复演练抽查），
+不在本批做。
+
+| 内容 | 在备份里吗 | 恢复手段 |
+| --- | --- | --- |
+| 数据库（Postgres 全库） | 是 | pg-backup 每天一次（留 7 份）+ restore-drill 恢复到测试库比对行数 |
+| 代码（/opt/codex-app-server，不含 work/） | 是 | 每次部署前打备份 tar（留 5 份）+ rollback.sh 恢复 |
+| 任务 worktree（work/、tenants/ 下各任务目录） | 否 | 无。主机丢了就丢了；数据库里的任务行还在，但 worktree 没了，续跑/链路/附件都会缺 |
+| 链路文件（host-traces/）与上传文件（附件包） | 否 | 无。链路是过程记录，丢了只影响可观测性；上传附件丢了则任务附件不可下载 |
+| 运行配置（/etc/codex-app-server/env、unit 文件、nginx/Caddy 配置） | 否（不在代码 tar 与数据库备份里） | 换机时从备份清单手动重贴，见“公网入口”与 new-host-checklist.md |
+
+后果要说透：只靠现有备份，数据库能恢复、代码能回滚，但一台主机彻底丢了，
+在跑任务的工作目录与历史链路是回不来的；恢复后需要把受影响的任务标记为
+失败/重建，而不是假装它们还能续跑。restore-drill.sh 的 PASS 只证明数据库
+备份可恢复，不证明整机可恢复（脚本头注释同样写明）。
+
+## 公网入口与 webhook secret
+
+- `caddy/Caddyfile`（P2-ops：占位符改为 Caddy 原生变量模板）：只暴露
+  `/webhooks/*` 与 `/oauth/*`，任务/项目/管理 API 对外一律 403；自动证书
+  走 Caddy（本批二选一里选了 Caddy，certbot 路径未做成模板、只是备选，
+  不要假装两条路都已就绪）。
 - `setup-webhook-secret.sh`：在主机上生成 webhook secret 并写入 env 文件，
   绝不打印 secret；已设则不动，避免轮换后 GitHub 端旧 secret 失效。
 
+### Caddy 模板部署步骤（新机可复现，不申请真实证书也能备好文件）
+
+```sh
+# 1. 主机上装 Caddy（包管理器或官方二进制均可），建运行目录：
+mkdir -p /etc/caddy
+cp deploy/caddy/Caddyfile /etc/caddy/Caddyfile
+cp deploy/caddy/caddy.service /etc/systemd/system/caddy.service
+# 2. 写变量文件（域名与邮箱不是密钥，但也不要写进仓库）：
+cat > /etc/caddy/env <<'EOF'
+DOMAIN=example.invalid
+ACME_EMAIL=ops@example.invalid
+EOF
+chmod 600 /etc/caddy/env
+# 3. Caddyfile 里的 {$DOMAIN} / {$ACME_EMAIL} 由 Caddy 运行时从环境变量
+#    展开；变量未设时 Caddy 启动即报错（fail-fast，不会静默跑成空域名）。
+systemctl daemon-reload
+systemctl enable --now caddy
+# 4. 验证：curl -v https://$DOMAIN/webhooks/github 应被 Caddy 转发
+#    （控制面本身仍只监听 127.0.0.1:8765，不直接对外）。
+```
+
+### 当前生产现状（换机需重贴，模板不等于已上线）
+
+当前生产走宝塔 nginx（api.diante.me 等 vhost 由 BT 面板管理），不是
+Caddy；BT 若重生成 vhost 会吃掉 location，需从备份贴回（见
+new-host-checklist.md 第 8 节）。仓库内 Caddy 模板是“新机可复现”的选项，
+不是“生产已切 Caddy”的声明。备份清单必须另含：BT vhost 配置备份、
+/etc/caddy/Caddyfile（如已用 Caddy）、caddy.service、unit 文件
+（codex-cloud-server*.service 等不在 deploy tar 里，改动后须单独 scp，
+见 codex-cloud-server-2.service 头注释）——这些都不在代码 tar 与数据库
+备份里，换机时不重贴就等于没备份。
+
 ## 仍未做（不要假装已做）
 
-- 推送告警通道：monitor 失败时通知到 Muse 聊天的通道还没定（见监控一节）。
 - 回滚不回数据库：见回滚边界一节；改类型/删列的迁移需要单独评估。
 - 恢复演练不碰生产库：演练只比行数，不比对具体内容；恢复到生产库的
   真实灾难恢复流程（停服务、恢复、重启、验 health）还没演练过。
+- worktree / 链路 / 上传目录未纳入备份：见“备份制度”一节；主机彻底丢了
+  这些回不来，需要另单定量后做。
+- 生产仍走宝塔 nginx，未切 Caddy：Caddy 模板只是仓库内可部署选项。
 - CloudExecutor、GUI、Slack 执行、推送实发：与本轮无关，仍是 0 或占位。
diff --git a/deploy/caddy/Caddyfile b/deploy/caddy/Caddyfile
index e4805a5..92c818c 100644
--- a/deploy/caddy/Caddyfile
+++ b/deploy/caddy/Caddyfile
@@ -1,13 +1,27 @@
-# 公网入口模板：只暴露 webhook 与 OAuth 回调，任务/项目/管理 API 不对外。
-# 使用前把 <DOMAIN> 换成你自己的域名，把 <EMAIL> 换成 Let's Encrypt 联系邮箱。
-# 部署侧在主机上装 Caddy 后，把本文件放到 /etc/caddy/Caddyfile 并 reload。
+# 公网入口模板（Caddy 自动证书路径，P2-ops 二选一里选了 Caddy，
+# certbot 路径只是备选、未做成模板）。
+#
+# 只暴露 webhook 与 OAuth 回调，任务/项目/管理 API 不对外。
+#
+# 变量是 Caddy 原生环境变量展开（{$DOMAIN} / {$ACME_EMAIL}），
+# 不是部署时 sed 替换：Caddy 启动时从进程环境读这两个变量，
+# 未设时启动即报错（fail-fast，不会静默跑成空域名）。
+# 部署步骤见 deploy/README.md“公网入口”一节；变量文件示例：
+#   /etc/caddy/env（600）：
+#     DOMAIN=example.invalid
+#     ACME_EMAIL=ops@example.invalid
+# caddy.service 模板（同目录）用 EnvironmentFile 指向该文件。
+#
 # 控制面本身仍只监听 127.0.0.1:8765，不直接对外；Caddy 只做反向代理 + TLS。
 # GitHub webhook 与 OAuth 回调都要求 https 且证书有效，Caddy 的自动证书满足这一点；
 # 自签名证书不行，GitHub 会拒绝送达，浏览器也会拦截 OAuth 回调。
+#
+# 当前生产现状：生产走宝塔 nginx，不是 Caddy；本模板是新机可复现的选项，
+# 不是“生产已切 Caddy”的声明（见 deploy/README.md）。
 
-<DOMAIN> {
+{$DOMAIN} {
     encode gzip
-    tls <EMAIL>
+    tls {$ACME_EMAIL}
 
     # GitHub webhook：签名校验在控制面做（GITHUB_WEBHOOK_SECRET），Caddy 只转发。
     @github path /webhooks/github
@@ -18,7 +32,7 @@
     reverse_proxy @slack 127.0.0.1:8765
 
     # OAuth 浏览器登录/回调：回调 URL 必须与 OAuth App 里填的完全一致，
-    # 即 https://<DOMAIN>/oauth/<provider>/callback。
+    # 即 https://{$DOMAIN}/oauth/<provider>/callback。
     @oauth path /oauth/*
     reverse_proxy @oauth 127.0.0.1:8765
 
diff --git a/deploy/caddy/caddy.service b/deploy/caddy/caddy.service
new file mode 100644
index 0000000..ad79e3a
--- /dev/null
+++ b/deploy/caddy/caddy.service
@@ -0,0 +1,24 @@
+# Caddy 前置代理 unit 模板（与 deploy/caddy/Caddyfile 配对）。
+#
+# 注意：unit 文件不在 deploy tar 里（与 codex-cloud-server-2.service
+# 同款约束），改动后须单独放到 /etc/systemd/system/ 再
+# daemon-reload + enable + start；只重启 Caddy 不会生效。
+# EnvironmentFile 指向 /etc/caddy/env（600，部署时按
+# deploy/README.md“公网入口”一节手写 DOMAIN / ACME_EMAIL，
+# 域名与邮箱不要写进仓库）。Caddyfile 里的 {$DOMAIN} /
+# {$ACME_EMAIL} 由 Caddy 运行时从该环境展开。
+[Unit]
+Description=Codex public entry (Caddy reverse proxy + automatic TLS)
+After=network.target
+
+[Service]
+User=caddy
+Group=caddy
+EnvironmentFile=/etc/caddy/env
+ExecStart=/usr/bin/caddy run --config /etc/caddy/Caddyfile --adapter caddyfile
+ExecReload=/usr/bin/caddy reload --config /etc/caddy/Caddyfile --adapter caddyfile
+Restart=on-failure
+RestartSec=10
+
+[Install]
+WantedBy=multi-user.target
diff --git a/deploy/deploy.py b/deploy/deploy.py
index 3a44bcf..512441e 100644
--- a/deploy/deploy.py
+++ b/deploy/deploy.py
@@ -291,8 +291,17 @@ def transfer(host, user, local, remote, expected_sha):
     sys.exit(1)
 
 
+# 部署前默认跑的测试清单：必须用 discover -s <目录>，
+# 绝不能用 deploy.tests 这种包路径——包路径会把 deploy 当包
+# import、盖掉 deploy.py 本体（2026-10-27 教训），导致
+# 13 个假报错。P2-ops：deploy/tests 以前漏在清单外，
+# deploy.py 自身的回归（如 ssh 分发）可能带病上线，
+# 现补进默认清单。
+DEFAULT_TEST_SUITES = ("tests", "cloud/tests", "deploy/tests")
+
+
 def run_tests():
-    for suite in ("tests", "cloud/tests"):
+    for suite in DEFAULT_TEST_SUITES:
         run(["python3", "-m", "unittest", "discover", "-s", suite],
             cwd=REPO_ROOT)
 
diff --git a/deploy/host_deploy.sh b/deploy/host_deploy.sh
index 1c905ec..c70bdea 100644
--- a/deploy/host_deploy.sh
+++ b/deploy/host_deploy.sh
@@ -133,6 +133,8 @@ worker_unit() {
 # server-2（8766）是 HA 阶段 2 加的；unit 存在才纳入停/起与 health
 # 检查，老主机没有它时退回单机行为。2026-10-27 首推教训：脚本只动
 # server-1 时，LB 上游的 server-2 仍跑旧内存代码，新旧混跑。
+# 与 rollback.sh 共享的 unit 清单约定（P2-ops）：改动清单时两脚本
+# 同步改，deploy/tests/test_p2_ops.py 会双向核对。
 SERVER_UNITS="codex-cloud-server"
 HEALTH_PORTS="8765"
 if systemctl cat codex-cloud-server-2 >/dev/null 2>&1; then
diff --git a/deploy/monitor.service b/deploy/monitor.service
index 049f799..c3bcdf7 100644
--- a/deploy/monitor.service
+++ b/deploy/monitor.service
@@ -6,8 +6,10 @@ After=network.target codex-cloud-server.service codex-cloud-worker.service
 Type=oneshot
 User=codexsvc
 WorkingDirectory=/opt/codex-app-server
-# monitor.sh 只做检查、不读 env 文件、不发消息；失败时 systemd 会把
-# 退出码与输出记进 journal，告警通道见 deploy/README.md（另行决定）。
+# monitor.sh 只做检查、不读 env 文件；默认不发消息（告警出口默认关闭），
+# 失败时 systemd 会把退出码与输出记进 journal。可选告警出口
+# （MONITOR_ALERT_URL / MONITOR_ALERT_MAILTO）见 deploy/README.md：
+# 配了才在 fail 时发，告警内容只有 pass/fail，不含密钥。
 ExecStart=/bin/sh /opt/codex-app-server/deploy/monitor.sh
 
 [Install]
diff --git a/deploy/monitor.sh b/deploy/monitor.sh
index d015d28..1da26fa 100755
--- a/deploy/monitor.sh
+++ b/deploy/monitor.sh
@@ -31,19 +31,45 @@
 # JSON），断言 health 那一项为 pass；其它项在非主机机器上 fail 是正常的，
 # self-test 不因此失败。不要把 --self-test 通过误读成“主机健康”。
 #
-# 告警边界：本脚本只做“检查 + journal 日志”（配合 monitor.timer 每 5 分钟
-# 跑一次），它本身不会发消息给用户。推送到 Muse 聊天的告警仍需用户另行
-# 决定通道（SSH key 常驻通道或 webhook URL），见 deploy/README.md。
+# 单机模式（P2-ops）：默认 server 检查要求 server-1 与 server-2
+# 双 active（HA 阶段 2），单机主机上 server-2 不存在会恒 fail，
+# 恒 fail 会掩盖真故障。确认是单机部署时设 MONITOR_SINGLE_HOST=1，
+# server 检查只查 server-1；双机主机上绝不要设它（设了就是
+# 静默降级，冗余丢了也看不见）。输出 JSON 里 server_mode 字段
+# 写明本次用的是 single 还是 dual，降级可见。
+#
+# 告警出口（P2-ops，默认关闭）：只在 overall=fail 时发一次，
+# pass 时不发；故障持续期间 timer 每 5 分钟跑一次就会重发，
+# 接收端需自行去重（本机不做去重状态文件，免得状态文件本身
+# 成为新的故障点）。二选一，同时配了只用 webhook：
+#   MONITOR_ALERT_URL=https://example.invalid/hook  # webhook POST JSON
+#   MONITOR_ALERT_MAILTO=ops@example.invalid        # mail 命令发邮件
+# 告警内容只有 JSON 检查结果（pass/fail 与 server_mode），
+# 不含 health 正文、不含任何环境变量、不含密钥；URL 与收件人
+# 只走环境变量，不写进脚本、不进日志。
 set -eu
 
+SERVER_MODE=dual
+if [ "${MONITOR_SINGLE_HOST:-0}" = "1" ]; then SERVER_MODE=single; fi
+
 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() {
-  # HA 阶段 2：两台 server 都必须 active，任一挂了都算 fail
+  # 默认双机：两台 server 都必须 active，任一挂了都算 fail
   # （单台 active 时 API 还能用，但冗余已丢，必须可见）。
+  # MONITOR_SINGLE_HOST=1 时只查 server-1（单机部署专用，
+  # 双机主机上设它等于静默降级，见文件头注释）。
+  if [ "$SERVER_MODE" = "single" ]; then
+    if systemctl is-active codex-cloud-server >/dev/null 2>&1; then
+      echo pass
+    else
+      echo fail
+    fi
+    return
+  fi
   if systemctl is-active codex-cloud-server >/dev/null 2>&1 \
     && systemctl is-active codex-cloud-server-2 >/dev/null 2>&1; then
     echo pass
@@ -136,11 +162,36 @@ run_checks() {
   for v in "$S" "$W" "$H" "$P" "$D" "$K" "$A" "$B"; 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", "backup": "%s"}}\n' \
-    "$OVERALL" "$S" "$W" "$H" "$P" "$D" "$K" "$A" "$B"
+  LAST_JSON=$(printf '{"overall": "%s", "server_mode": "%s", "checks": {"server": "%s", "worker": "%s", "health": "%s", "postgres": "%s", "docker_postgres": "%s", "disk": "%s", "app_dir": "%s", "backup": "%s"}}' \
+    "$OVERALL" "$SERVER_MODE" "$S" "$W" "$H" "$P" "$D" "$K" "$A" "$B")
+  printf '%s\n' "$LAST_JSON"
   if [ "$OVERALL" = "pass" ]; then return 0; else return 1; fi
 }
 
+send_alert() {
+  # 只在 fail 时由主流程调用。内容就是 LAST_JSON（pass/fail
+  # 与 server_mode），绝不拼 health 正文、环境变量或密钥。
+  if [ -n "${MONITOR_ALERT_URL:-}" ]; then
+    if [ -n "${MONITOR_ALERT_MAILTO:-}" ]; then
+      echo "monitor: webhook 与邮件同时配置，只用 webhook" >&2
+    fi
+    curl -s --max-time 10 -X POST -H 'Content-Type: application/json' \
+      -d "$LAST_JSON" "$MONITOR_ALERT_URL" >/dev/null 2>&1 \
+      || echo "monitor: webhook 告警发送失败" >&2
+    return 0
+  fi
+  if [ -n "${MONITOR_ALERT_MAILTO:-}" ]; then
+    if command -v mail >/dev/null 2>&1; then
+      printf '%s\n' "$LAST_JSON" \
+        | mail -s "codex monitor FAIL" "$MONITOR_ALERT_MAILTO" \
+        2>/dev/null \
+        || echo "monitor: 邮件告警发送失败" >&2
+    else
+      echo "monitor: 已配 MAILTO 但本机无 mail 命令，告警未发出" >&2
+    fi
+  fi
+}
+
 if [ "${1:-}" = "--self-test" ]; then
   # 起一个 canned health server：只返回固定健康 JSON，不读任何真实服务。
   SELF_PORT=18999
@@ -173,8 +224,11 @@ PY
   SERVER_PID=$!
   trap 'kill $SERVER_PID 2>/dev/null || true' EXIT
   sleep 1
+  # 内层跑完整检查时清掉告警出口：self-test 在非主机上
+  # 其它项必然 fail，不清会在配了告警的环境里误发一条。
   MONITOR_HEALTH_URL="http://127.0.0.1:$SELF_PORT/health" \
     HEALTH_URL="http://127.0.0.1:$SELF_PORT/health" \
+    MONITOR_ALERT_URL= MONITOR_ALERT_MAILTO= \
     sh "$0" > /tmp/monitor-self-test.json 2>/dev/null || true
   # 注意：上面那行是跑完整检查，health 以外项在非主机上 fail 是正常的；
   # self-test 只断言两件事：输出是合法 JSON，且 health 解析为 pass。
@@ -195,4 +249,10 @@ sys.exit(0 if d["checks"]["health"] == "pass" else 1)
   fi
 fi
 
-run_checks
+if run_checks; then
+  exit 0
+fi
+# 只有 fail 走到这里：默认不配出口时 send_alert 是空操作，
+# 不会因为没配告警就多报错；配了出口则发一条 fail 告警。
+send_alert
+exit 1
diff --git a/deploy/restore-drill.sh b/deploy/restore-drill.sh
index 311ac67..c153be3 100755
--- a/deploy/restore-drill.sh
+++ b/deploy/restore-drill.sh
@@ -4,6 +4,12 @@
 # 用法（主机上）：sh /opt/codex-app-server/deploy/restore-drill.sh
 #   可选环境变量：BACKUP_DIR（默认 /var/lib/codex-app-server/backups）。
 #
+# 演练范围（P2-ops，诚实边界）：本演练只证明“数据库备份可恢复”
+# （行数比对通过）。它不证明 worktree / 链路 / 上传文件可恢复，
+# 因为那些根本不在备份里（见 deploy/README.md“备份制度”一节）：
+# 代码备份 tar 明确排除 work/，pg-backup 只备数据库。不要把
+# “恢复演练 PASS”读成“整机可恢复”。
+#
 # 步骤：
 #   1. 跑一次 backup_postgres 生成新鲜备份（.sql）。
 #   2. 新建一个测试库 codex_control_restore_test_<时间戳>，用
diff --git a/deploy/rollback.sh b/deploy/rollback.sh
index 8010539..f294dc3 100755
--- a/deploy/rollback.sh
+++ b/deploy/rollback.sh
@@ -10,9 +10,23 @@
 #
 # 回滚边界：只回代码，不回数据库（理由见 deploy.sh 头注释与
 # deploy/README.md：数据库迁移是只加列的幂等操作，老代码仍能读新列）。
-# 备份包来自 deploy.sh 每次部署前在主机上打的
+# 备份包来自每次部署前在主机上打的
 # /opt/codex-app-server-backups/codex-app-server-<时间戳>.tar.gz，
 # 本脚本恢复其中最新的一份；主机上没有备份包时直接报错退出，不编造。
+#
+# 与 host_deploy.sh 共享的约定（P2-ops，两脚本改动时必须同步，
+# deploy/tests/test_p2_ops.py 会双向核对，见文件尾注释）：
+#   - server unit 清单：codex-cloud-server 必含，codex-cloud-server-2
+#     以 systemctl cat 探测为准（存在才纳入停/起与健康检查）；
+#   - worker：autoscaler active 时停/起 autoscaler，否则 cloud-worker
+#     （与 host_deploy.sh 的 worker_unit() 同款）；
+#   - 成功标准：清单里每个 server 端口的 health 都通过（8765 必查，
+#     server-2 存在时 8766 也必查），只通 8765 不算回滚成功——
+#     否则 LB 上游的 server-2 仍跑故障代码，回滚等于没回完。
+#
+# 仿真覆盖（仅测试用，生产用默认值）：CODEX_ROLLBACK_APP_DIR、
+# CODEX_ROLLBACK_BACKUP_DIR、CODEX_ROLLBACK_HEALTH_ATTEMPTS、
+# CODEX_ROLLBACK_HEALTH_INTERVAL。默认值与旧行为完全一致。
 set -eu
 
 HOST=${1:-}
@@ -33,12 +47,30 @@ ssh_cmd() {
 
 ssh_cmd "$USER_ARG@$HOST" 'sh -s' <<'HOST_SCRIPT'
 set -eu
-APP_DIR=/opt/codex-app-server
-BACKUP_DIR=/opt/codex-app-server-backups
+APP_DIR=${CODEX_ROLLBACK_APP_DIR:-/opt/codex-app-server}
+BACKUP_DIR=${CODEX_ROLLBACK_BACKUP_DIR:-/opt/codex-app-server-backups}
+HEALTH_ATTEMPTS=${CODEX_ROLLBACK_HEALTH_ATTEMPTS:-12}
+HEALTH_INTERVAL=${CODEX_ROLLBACK_HEALTH_INTERVAL:-5}
+
+# 与 host_deploy.sh 共享的 unit 清单约定（见本文件头注释）：
+# server-1 必含，server-2 以 unit 存在为准；改动清单时两脚本同步改。
+SERVER_UNITS="codex-cloud-server"
+HEALTH_PORTS="8765"
+if systemctl cat codex-cloud-server-2 >/dev/null 2>&1; then
+  SERVER_UNITS="$SERVER_UNITS codex-cloud-server-2"
+  HEALTH_PORTS="$HEALTH_PORTS 8766"
+fi
+if systemctl is-active -q codex-autoscaler 2>/dev/null; then
+  WORKER_UNIT=codex-autoscaler
+else
+  WORKER_UNIT=codex-cloud-worker
+fi
 
 check_health() {
-  OUT=$(curl -s --max-time 5 http://127.0.0.1:8765/health || true)
-  [ -n "$OUT" ] && printf '%s' "$OUT" | python3 -c '
+  for port in $HEALTH_PORTS; do
+    OUT=$(curl -s --max-time 5 "http://127.0.0.1:$port/health" || true)
+    [ -n "$OUT" ] || return 1
+    if ! printf '%s' "$OUT" | python3 -c '
 import json, sys
 try:
     d = json.load(sys.stdin)
@@ -46,14 +78,18 @@ 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)
-'
+'; then
+      return 1
+    fi
+  done
+  return 0
 }
 
 wait_health() {
   i=0
-  while [ "$i" -lt 12 ]; do
+  while [ "$i" -lt "$HEALTH_ATTEMPTS" ]; do
     if check_health; then return 0; fi
-    sleep 5
+    sleep "$HEALTH_INTERVAL"
     i=$((i + 1))
   done
   return 1
@@ -66,15 +102,18 @@ if [ -z "$BACKUP_FILE" ]; then
   exit 1
 fi
 echo "将恢复备份包：$BACKUP_FILE"
-systemctl stop codex-cloud-worker codex-cloud-server || true
+echo "停服务：worker_unit=$WORKER_UNIT server_units=$SERVER_UNITS"
+systemctl stop "$WORKER_UNIT" $SERVER_UNITS || true
 find "$APP_DIR" -mindepth 1 -maxdepth 1 ! -name work -exec rm -rf {} +
-tar -xzf "$BACKUP_FILE" -C /opt
-chown -R codexsvc:codexsvc "$APP_DIR"
-systemctl start codex-cloud-server codex-cloud-worker
+tar -xzf "$BACKUP_FILE" -C "$(dirname "$APP_DIR")"
+if id codexsvc >/dev/null 2>&1; then
+  chown -R codexsvc:codexsvc "$APP_DIR"
+fi
+systemctl start $SERVER_UNITS "$WORKER_UNIT"
 if wait_health; then
-  echo "回滚成功，health 通过，当前代码为备份包：$BACKUP_FILE"
+  echo "回滚成功，全部端口 health 通过（$HEALTH_PORTS），当前代码为备份包：$BACKUP_FILE"
   exit 0
 fi
-echo "回滚后 health 仍失败，需人工介入" >&2
+echo "回滚后 health 仍失败（需全部端口通过），需人工介入" >&2
 exit 1
 HOST_SCRIPT
diff --git a/deploy/tests/test_deploy.py b/deploy/tests/test_deploy.py
index 950b5b8..4ba850a 100644
--- a/deploy/tests/test_deploy.py
+++ b/deploy/tests/test_deploy.py
@@ -82,6 +82,33 @@ class TestDeployPure(unittest.TestCase):
         self.assertIsNone(deploy_mod.live_commit("Warning: something\n"))
 
 
+class TestDefaultTestSuites(unittest.TestCase):
+    """P2-ops 回归：部署前默认清单必须含 deploy/tests，
+    且 run_tests() 确实按 discover -s 逐套跑（不是包路径）。"""
+
+    def test_default_suites_include_deploy_tests(self):
+        self.assertIn("deploy/tests", deploy_mod.DEFAULT_TEST_SUITES)
+        self.assertIn("tests", deploy_mod.DEFAULT_TEST_SUITES)
+        self.assertIn("cloud/tests", deploy_mod.DEFAULT_TEST_SUITES)
+
+    def test_run_tests_discovers_every_suite(self):
+        calls = []
+
+        def fake_run(cmd, cwd=None, check=True):
+            calls.append(cmd)
+            return subprocess.CompletedProcess(args=cmd, returncode=0,
+                                               stdout="", stderr="")
+
+        with mock.patch.object(deploy_mod, "run", side_effect=fake_run):
+            deploy_mod.run_tests()
+        discovered = [c[c.index("-s") + 1] for c in calls
+                      if "discover" in c]
+        self.assertEqual(discovered, list(deploy_mod.DEFAULT_TEST_SUITES))
+        # 没有任何一套走包路径（deploy.tests 会盖掉 deploy.py）。
+        for c in calls:
+            self.assertNotIn("deploy.tests", c)
+
+
 class TestBuildRelease(unittest.TestCase):
     def test_build_from_temp_git_repo(self):
         with tempfile.TemporaryDirectory() as tmp:
diff --git a/deploy/tests/test_p2_ops.py b/deploy/tests/test_p2_ops.py
new file mode 100644
index 0000000..3485cf2
--- /dev/null
+++ b/deploy/tests/test_p2_ops.py
@@ -0,0 +1,296 @@
+"""P2-ops 回归：monitor 单机开关与告警出口、rollback 与
+host_deploy 的 unit/健康清单同步。
+
+全部用假 systemctl / curl / ssh / mail，不碰真实主机与真实服务。
+告警测试用的 webhook 接收器只监听 127.0.0.1 临时端口。
+"""
+import http.server
+import json
+import os
+import shutil
+import subprocess
+import tarfile
+import tempfile
+import threading
+import unittest
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
+DEPLOY_DIR = os.path.join(REPO_ROOT, "deploy")
+MONITOR_SCRIPT = os.path.join(DEPLOY_DIR, "monitor.sh")
+ROLLBACK_SCRIPT = os.path.join(DEPLOY_DIR, "rollback.sh")
+HOST_DEPLOY_SCRIPT = os.path.join(DEPLOY_DIR, "host_deploy.sh")
+REAL_CURL = shutil.which("curl")
+
+
+def _write_exe(path, body):
+    with open(path, "w") as f:
+        f.write(body)
+    os.chmod(path, 0o755)
+
+
+class MonitorSim(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        base = self.tmp.name
+        self.app_dir = os.path.join(base, "app")
+        os.makedirs(self.app_dir)
+        self.backup_dir = os.path.join(base, "backups")
+        os.makedirs(self.backup_dir)
+        with open(os.path.join(self.backup_dir,
+                               "cloud-backup-fresh.sql"), "w") as f:
+            f.write("-- fresh\n")
+        self.health_path = os.path.join(base, "health.json")
+        # 健康正文里埋一个哨兵串：告警正文绝不能把它带出去
+        # （告警只能发 LAST_JSON 的 pass/fail，不发 health 正文）。
+        with open(self.health_path, "w") as f:
+            json.dump({"ok": True, "backend": "postgres", "db_ok": True,
+                       "worker_ok": True,
+                       "note": "sentinel-health-secret"}, f)
+        fake_bin = os.path.join(base, "fakebin")
+        os.makedirs(fake_bin)
+        _write_exe(os.path.join(fake_bin, "systemctl"),
+                   "#!/bin/sh\n"
+                   "if [ \"$1\" = \"is-active\" ]; then\n"
+                   "  case \"$2\" in\n"
+                   "    codex-cloud-server) exit \"$FAKE_SERVER1\" ;;\n"
+                   "    codex-cloud-server-2) exit \"$FAKE_SERVER2\" ;;\n"
+                   "    *) exit 1 ;;\n"
+                   "  esac\n"
+                   "fi\n"
+                   "exit 0\n")
+        real_curl = REAL_CURL or "/usr/bin/curl"
+        self.curl_log = os.path.join(base, "curl.log")
+        open(self.curl_log, "w").close()
+        _write_exe(os.path.join(fake_bin, "curl"),
+                   "#!/bin/sh\n"
+                   "echo \"$@\" >> \"$FAKE_CURL_LOG\"\n"
+                   "for a in \"$@\"; do\n"
+                   "  case \"$a\" in\n"
+                   "    *\"/health\") cat \"$FAKE_HEALTH_JSON\"; exit 0 ;;\n"
+                   "  esac\n"
+                   "done\n"
+                   "exec \"%s\" \"$@\"\n" % real_curl)
+        self.fake_bin = fake_bin
+        self.env = dict(os.environ)
+        self.env["PATH"] = fake_bin + os.pathsep + os.environ["PATH"]
+        self.env["MONITOR_HEALTH_URL"] = "http://127.0.0.1:8765/health"
+        self.env["MONITOR_APP_DIR"] = self.app_dir
+        self.env["MONITOR_DISK_PATH"] = base
+        self.env["MONITOR_BACKUP_DIR"] = self.backup_dir
+        self.env["MONITOR_ALERT_URL"] = ""
+        self.env["MONITOR_ALERT_MAILTO"] = ""
+        self.env["FAKE_HEALTH_JSON"] = self.health_path
+        self.env["FAKE_CURL_LOG"] = self.curl_log
+        self.env["FAKE_SERVER1"] = "0"  # server-1 active
+        self.env["FAKE_SERVER2"] = "1"  # server-2 inactive
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def run_monitor(self, extra_env=None):
+        env = dict(self.env)
+        if extra_env:
+            env.update(extra_env)
+        r = subprocess.run(["sh", MONITOR_SCRIPT],
+                           capture_output=True, text=True, env=env)
+        return r, json.loads(r.stdout.strip())
+
+    def test_single_host_passes_server_check(self):
+        _, out = self.run_monitor({"MONITOR_SINGLE_HOST": "1"})
+        self.assertEqual(out["server_mode"], "single")
+        self.assertEqual(out["checks"]["server"], "pass")
+
+    def test_dual_host_fails_server_check_when_server2_down(self):
+        # 默认双机：server-2 不 active 即 fail，不能静默降级。
+        _, out = self.run_monitor()
+        self.assertEqual(out["server_mode"], "dual")
+        self.assertEqual(out["checks"]["server"], "fail")
+
+    def test_webhook_alert_on_fail_only(self):
+        if REAL_CURL is None:
+            self.skipTest("本机无 curl，跳过 webhook 告警测试")
+        received = []
+
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def do_POST(self):
+                length = int(self.headers.get("Content-Length", 0))
+                received.append(self.rfile.read(length))
+                self.send_response(200)
+                self.end_headers()
+
+            def log_message(self, *a):
+                pass
+
+        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            url = "http://127.0.0.1:%d/hook" % server.server_address[1]
+            r, out = self.run_monitor({"MONITOR_ALERT_URL": url})
+            self.assertNotEqual(r.returncode, 0)  # 其它项 fail，整体 fail
+            self.assertEqual(len(received), 1)
+            body = json.loads(received[0].decode())
+            self.assertEqual(body["overall"], "fail")
+            self.assertNotIn("sentinel-health-secret",
+                             received[0].decode())
+        finally:
+            server.shutdown()
+            server.server_close()
+
+    def test_alert_default_off(self):
+        # 不配出口：即使整体 fail，monitor 也不应发出任何告警请求。
+        # 行为证据：假 curl 记录了全部调用，其中没有任何 POST
+        # （webhook 告警是唯一的 POST 调用；health 检查是 GET）。
+        r, out = self.run_monitor()
+        self.assertNotEqual(r.returncode, 0)
+        self.assertEqual(out["overall"], "fail")
+        with open(self.curl_log) as f:
+            calls = f.read()
+        self.assertNotIn("POST", calls)
+        self.assertIn("/health", calls)
+
+    def test_mail_alert_uses_mail_command(self):
+        mail_log = os.path.join(self.tmp.name, "mail.log")
+        _write_exe(os.path.join(self.fake_bin, "mail"),
+                   "#!/bin/sh\n"
+                   "echo \"ARGS: $@\" >> \"%s\"\n"
+                   "cat >> \"%s\"\n" % (mail_log, mail_log))
+        r, _ = self.run_monitor(
+            {"MONITOR_ALERT_MAILTO": "ops@example.invalid"})
+        self.assertNotEqual(r.returncode, 0)
+        with open(mail_log) as f:
+            content = f.read()
+        self.assertIn("codex monitor FAIL", content)
+        self.assertIn('"overall": "fail"', content)
+        self.assertNotIn("sentinel-health-secret", content)
+
+
+class RollbackSim(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        base = self.tmp.name
+        self.app = os.path.join(base, "codex-app-server")
+        os.makedirs(os.path.join(self.app, "work"))
+        with open(os.path.join(self.app, "work", "keep.txt"), "w") as f:
+            f.write("keep\n")
+        with open(os.path.join(self.app, "broken.py"), "w") as f:
+            f.write("x = 0\n")
+        self.backup_dir = os.path.join(base, "backups")
+        os.makedirs(self.backup_dir)
+        # 备份包：顶层目录名必须等于 APP_DIR 的 basename，
+        # rollback 解包到 dirname(APP_DIR) 后正好落回 APP_DIR。
+        src = os.path.join(base, "src", "codex-app-server")
+        os.makedirs(src)
+        with open(os.path.join(src, "restored.py"), "w") as f:
+            f.write("x = 1\n")
+        with open(os.path.join(src, "DEPLOYED_COMMIT"), "w") as f:
+            f.write("backupcommit\n")
+        tarball = os.path.join(self.backup_dir,
+                               "codex-app-server-20260101.tar.gz")
+        with tarfile.open(tarball, "w:gz") as tf:
+            for root, dirs, files in os.walk(src):
+                dirs.sort()
+                for name in sorted(files):
+                    full = os.path.join(root, name)
+                    tf.add(full, arcname=os.path.relpath(full, os.path.join(
+                        base, "src")))
+        self.health_dir = os.path.join(base, "health")
+        os.makedirs(self.health_dir)
+        fake_bin = os.path.join(base, "fakebin")
+        os.makedirs(fake_bin)
+        self.log_path = os.path.join(base, "systemctl.log")
+        open(self.log_path, "w").close()
+        _write_exe(os.path.join(fake_bin, "ssh"),
+                   "#!/bin/sh\n"
+                   "# 假 ssh：忽略目标参数，直接把 heredoc 脚本在本地跑。\n"
+                   "exec sh -s\n")
+        _write_exe(os.path.join(fake_bin, "systemctl"),
+                   "#!/bin/sh\n"
+                   "echo \"$@\" >> \"$FAKE_SYSTEMCTL_LOG\"\n"
+                   "if [ \"$1\" = \"cat\" ] && [ \"$2\" = \"codex-cloud-server-2\" ]; then\n"
+                   "  if [ \"$FAKE_SERVER2_PRESENT\" = \"0\" ]; then exit 1; fi\n"
+                   "  exit 0\n"
+                   "fi\n"
+                   "if [ \"$1\" = \"is-active\" ]; then exit 1; fi\n"
+                   "exit 0\n")
+        _write_exe(os.path.join(fake_bin, "curl"),
+                   "#!/bin/sh\n"
+                   "for a in \"$@\"; do\n"
+                   "  case \"$a\" in\n"
+                   "    *\":8765/health\") cat \"$FAKE_HEALTH_DIR/8765.json\"; exit 0 ;;\n"
+                   "    *\":8766/health\") cat \"$FAKE_HEALTH_DIR/8766.json\"; exit 0 ;;\n"
+                   "  esac\n"
+                   "done\n"
+                   "exit 1\n")
+        self.env = dict(os.environ)
+        self.env["PATH"] = fake_bin + os.pathsep + os.environ["PATH"]
+        self.env["CODEX_ROLLBACK_APP_DIR"] = self.app
+        self.env["CODEX_ROLLBACK_BACKUP_DIR"] = self.backup_dir
+        self.env["CODEX_ROLLBACK_HEALTH_ATTEMPTS"] = "2"
+        self.env["CODEX_ROLLBACK_HEALTH_INTERVAL"] = "0.1"
+        self.env["FAKE_SYSTEMCTL_LOG"] = self.log_path
+        self.env["FAKE_HEALTH_DIR"] = self.health_dir
+        self.env["FAKE_SERVER2_PRESENT"] = "1"
+        self.env.pop("SSH_PROXY_CMD", None)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def set_health(self, port, ok):
+        with open(os.path.join(self.health_dir, "%s.json" % port), "w") as f:
+            json.dump({"ok": ok, "backend": "postgres", "db_ok": ok}, f)
+
+    def run_rollback(self):
+        return subprocess.run(["sh", ROLLBACK_SCRIPT, "testhost"],
+                              capture_output=True, text=True, env=self.env)
+
+    def systemctl_log(self):
+        with open(self.log_path) as f:
+            return f.read()
+
+    def test_success_requires_both_ports(self):
+        self.set_health(8765, True)
+        self.set_health(8766, True)
+        r = self.run_rollback()
+        self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
+        self.assertIn("全部端口 health 通过", r.stdout)
+        log = self.systemctl_log()
+        self.assertIn(
+            "stop codex-cloud-worker codex-cloud-server codex-cloud-server-2",
+            log)
+        self.assertIn(
+            "start codex-cloud-server codex-cloud-server-2 codex-cloud-worker",
+            log)
+        self.assertTrue(os.path.exists(os.path.join(self.app, "restored.py")))
+        self.assertFalse(os.path.exists(os.path.join(self.app, "broken.py")))
+        with open(os.path.join(self.app, "work", "keep.txt")) as f:
+            self.assertEqual(f.read().strip(), "keep")
+
+    def test_server2_health_failure_fails_rollback(self):
+        # 8765 通、8766 不通：只通一端口不算回滚成功。
+        self.set_health(8765, True)
+        self.set_health(8766, False)
+        r = self.run_rollback()
+        self.assertNotEqual(r.returncode, 0, r.stdout + r.stderr)
+        self.assertIn("health 仍失败", r.stdout + r.stderr)
+
+
+class TestUnitListSync(unittest.TestCase):
+    """两脚本清单同步核对：任一脚本改了清单、另一脚本没跟进，
+    这里就变红，而不是等到回滚时才发现 server-2 没被停/验。"""
+
+    def test_both_scripts_share_server_units_and_ports(self):
+        with open(ROLLBACK_SCRIPT) as f:
+            rollback = f.read()
+        with open(HOST_DEPLOY_SCRIPT) as f:
+            host_deploy = f.read()
+        for script, name in ((rollback, "rollback.sh"),
+                             (host_deploy, "host_deploy.sh")):
+            self.assertIn("codex-cloud-server-2", script, name)
+            self.assertIn("8766", script, name)
+            self.assertIn("共享的 unit 清单", script, name)
+
+
+if __name__ == "__main__":
+    unittest.main()