diff --git a/app/service/profile_assembly_service.py b/app/service/profile_assembly_service.py
index ce68ecc..09b739b 100644
--- a/app/service/profile_assembly_service.py
+++ b/app/service/profile_assembly_service.py
@@ -226,8 +226,16 @@ class ProfileAssemblyService:
     ) -> None:
         """写入新版本快照并把旧版本置为非当前。
 
-        唯一键 `uk_profile_snapshot_current` 建立在生成列 `current_customer_id` 上，
+        唯一键 `uk_profile_snapshot_current` 建立在 `current_customer_id` 上，
         保证「每个客户最多一条 current」；因此必须先清旧再写新，顺序不能反。
+
+        ⚠️ **该列不是生成列**（订正 2026-09-12）：它由 `alembic/baseline_generated.sql`
+        与真实库确认为**普通可空列 + 唯一键**，`app/model/profile.py` 的模块 docstring
+        第 2 条已写明"当前版本必须由写入方**显式写入**客户 ID（历史版本写 NULL）"。
+        此处原先按"生成列"理解而两处都不赋值，后果实测到：
+        旧版本 `is_current=0` 却仍留着 `current_customer_id`，新版本 `is_current=1`
+        却是 NULL —— **唯一键形同虚设**（多个 NULL 不冲突），
+        而且旧值一旦残留、新值再补写就会直接撞键。
         """
         previous = list(await self.session.scalars(
             select(ProfileSnapshot).where(
@@ -236,6 +244,8 @@ class ProfileAssemblyService:
         ))
         for row in previous:
             row.is_current = False
+            # 必须同时清空，否则与即将插入的新当前版本撞唯一键（见 docstring）。
+            row.current_customer_id = None
             row.updated_at = now
         await self.session.flush()
 
@@ -252,6 +262,8 @@ class ProfileAssemblyService:
             generation_basis=basis,
             snapshot_hash=sha256(payload.encode("utf-8")).hexdigest(),
             is_current=True,
+            # 当前版本必须显式写入客户 ID（历史版本为 NULL），见 docstring。
+            current_customer_id=customer_id,
             generated_at=now,
             created_at=now,
             updated_at=now,
diff --git a/tools/fix_profile_snapshot_current.py b/tools/fix_profile_snapshot_current.py
new file mode 100644
index 0000000..287241b
--- /dev/null
+++ b/tools/fix_profile_snapshot_current.py
@@ -0,0 +1,88 @@
+"""数据订正：`profile_snapshots.current_customer_id` 与 `is_current` 对齐。
+
+**默认 dry-run**，加 `--apply` 才提交。幂等，可重复运行。
+
+### 背景
+
+`current_customer_id` 是**普通可空列 + 唯一键** `uk_profile_snapshot_current`（不是生成列），
+契约是「当前版本写客户 ID、历史版本写 NULL」。但写入方曾把它误当生成列，
+两处路径都未赋值（`ProfileAssemblyService._write_snapshot`、
+`CustomerProfileCandidateService._write_profile_snapshot`），造成两类脏数据：
+
+1. 历史版本 `is_current=0` 却**仍留着** `current_customer_id`；
+2. 当前版本 `is_current=1` 却是 NULL。
+
+后果：唯一键形同虚设（多个 NULL 不冲突）⇒「每个客户最多一条 current」失效；
+而且 (1) 的值一旦与 (2) 补上的值相同就会**撞唯一键**。
+
+### 顺序要紧
+
+先清 (1) 再补 (2)：反过来的话，第 2 步写入时旧值还在，会直接撞键。
+
+用法：
+    .\\.venv\\Scripts\\python.exe tools\\fix_profile_snapshot_current.py           # 先看
+    .\\.venv\\Scripts\\python.exe tools\\fix_profile_snapshot_current.py --apply   # 再改
+"""
+
+import asyncio
+import sys
+
+from sqlalchemy import text
+
+from app.infrastructure.db import engine
+
+#: (1) 历史版本不该持有该列 → 清空。必须先做。
+_CLEAR_STALE = text("""
+    UPDATE profile_snapshots
+       SET current_customer_id = NULL
+     WHERE is_current = 0 AND current_customer_id IS NOT NULL
+""")
+
+#: (2) 当前版本必须持有该列 → 补上。放在 (1) 之后，避免撞唯一键。
+_FILL_CURRENT = text("""
+    UPDATE profile_snapshots
+       SET current_customer_id = customer_id
+     WHERE is_current = 1 AND current_customer_id IS NULL
+""")
+
+_COUNT_STALE = text(
+    "SELECT COUNT(*) FROM profile_snapshots "
+    "WHERE is_current = 0 AND current_customer_id IS NOT NULL"
+)
+_COUNT_MISSING = text(
+    "SELECT COUNT(*) FROM profile_snapshots "
+    "WHERE is_current = 1 AND current_customer_id IS NULL"
+)
+
+
+async def main(apply: bool) -> int:
+    async with engine.connect() as conn:
+        stale = await conn.scalar(_COUNT_STALE)
+        missing = await conn.scalar(_COUNT_MISSING)
+        print(f"待清空（is_current=0 却留着值）: {stale}")
+        print(f"待补写（is_current=1 却是 NULL）: {missing}")
+
+        if not apply:
+            print("dry-run：未提交。确认无误后加 --apply 再运行。")
+            await conn.rollback()
+            return 0
+
+        cleared = (await conn.execute(_CLEAR_STALE)).rowcount
+        filled = (await conn.execute(_FILL_CURRENT)).rowcount
+        await conn.commit()
+        print(f"已清空 {cleared} 行、已补写 {filled} 行")
+
+        # 复核：两类异常都应归零
+        print("复核 待清空:", await conn.scalar(_COUNT_STALE))
+        print("复核 待补写:", await conn.scalar(_COUNT_MISSING))
+        print("\n各客户的当前快照：")
+        for row in (await conn.execute(text(
+            "SELECT customer_id, version, is_current, current_customer_id "
+            "FROM profile_snapshots ORDER BY customer_id, version"
+        ))).mappings().all():
+            print("  ", dict(row))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main("--apply" in sys.argv)))