test(schema-sync): 迁移链按序重放 ALTER，比对基准由历史并集改为终态

早前实现把全部 V*.sql 的 CREATE 列做并集：某表日后被 ALTER DROP
COLUMN 时并集里旧列名永存，init-dev.sql 必须保留死列才能过测——
守卫会主动背书一份与生产结构漂移的开发库 DDL。按文件序重放
ADD/DROP COLUMN 后比对基准为迁移链终态；列改名（CHANGE）出现时
重放报差异属人工核对时机（改名同时牵动 init-dev 与 ORM）。

Co-Authored-By: Claude Code <noreply@anthropic.com>

backend/tests/integration/test_schema_sync.py

@@ -28,10 +28,14 @@ _CREATE_TABLE = re.compile(r"CREATE TABLE\s+`?(\w+)`?\s*\((.*?)\n\)", re.DOTALL)
 _COLUMN_LINE = re.compile(r"^\s{4}`?(\w+)`?\s", re.MULTILINE)
 # 这些开头的是约束定义，不是列定义
 _NON_COLUMN_KEYWORDS = {"CONSTRAINT", "PRIMARY", "UNIQUE", "FOREIGN", "KEY", "INDEX"}
+# ALTER TABLE 的列级变更（语义重放用）：ADD COLUMN / DROP COLUMN；
+# MODIFY 只改类型不改列名，对列集合无影响，无需处理
+_ALTER_ADD = re.compile(r"ALTER TABLE\s+`?(\w+)`?\s+ADD(?:\s+COLUMN)?\s+`?(\w+)`?", re.IGNORECASE)
+_ALTER_DROP = re.compile(r"ALTER TABLE\s+`?(\w+)`?\s+DROP(?:\s+COLUMN)?\s+`?(\w+)`?", re.IGNORECASE)
 
 
 def _parse_tables(sql: str) -> dict[str, set[str]]:
-    """从 SQL 文本中提取 {表名: {列名集合}}。"""
+    """从 SQL 文本中提取 {表名: {列名集合}}（仅 CREATE TABLE 的静态解析）。"""
     tables: dict[str, set[str]] = {}
     for table_name, body in _CREATE_TABLE.findall(sql):
         tables[table_name] = {
@@ -42,6 +46,27 @@ def _parse_tables(sql: str) -> dict[str, set[str]]:
     return tables
 
 
+def _migration_final_state(paths: list[Path]) -> dict[str, set[str]]:
+    """按文件序重放迁移脚本，得到迁移链**终态**的 {表名: {列名集合}}。
+
+    早前实现把全部脚本的 CREATE 列做并集：某表日后被 ALTER DROP COLUMN
+    时，并集里旧列名永远存在，init-dev.sql 必须保留死列才能过测 —— 守卫
+    会主动背书一份与生产结构漂移的开发库 DDL。按序重放 ADD/DROP 后，
+    比对基准才是终态（CHANGE RENAME 列名出现时重放会报差异，属人工核对
+    的正确时机：改名同时牵动 init-dev 与 ORM，须人来确认两处同步）。
+    """
+    merged: dict[str, set[str]] = {}
+    for path in paths:
+        sql = path.read_text(encoding="utf-8")
+        for table, columns in _parse_tables(sql).items():
+            merged.setdefault(table, set()).update(columns)
+        for table, column in _ALTER_ADD.findall(sql):
+            merged.setdefault(table, set()).add(column)
+        for table, column in _ALTER_DROP.findall(sql):
+            merged.setdefault(table, set()).discard(column)
+    return merged
+
+
 def _alembic_head() -> str:
     """从迁移脚本推断当前 head 版本号。
 
@@ -65,15 +90,10 @@ def _alembic_head() -> str:
 
 @pytest.fixture(scope="module")
 def migration_tables() -> dict[str, set[str]]:
-    """合并解析 db/migration/ 下全部 V*.sql。"""
+    """按文件序重放 db/migration/ 下全部 V*.sql，取迁移链终态。"""
     files = sorted(MIGRATION_DIR.glob("V*.sql"))
     assert files, f"未找到迁移脚本：{MIGRATION_DIR}"
-
-    merged: dict[str, set[str]] = {}
-    for path in files:
-        for table, columns in _parse_tables(path.read_text(encoding="utf-8")).items():
-            merged.setdefault(table, set()).update(columns)
-    return merged
+    return _migration_final_state(files)
 
 
 @pytest.fixture(scope="module")