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

@@ -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
 
 

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",
 ]

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()