fix: migration never prunes state on partial failure

docs/superpowers/plans/2026-06-10-audit-fix-plan.md

@@ -95,9 +95,9 @@ def rollback(self, chapter: int) -> bool:
 - Modify: `webnovel-writer/scripts/data_modules/migrate_state_to_sqlite.py:235-258`
 - Test: `webnovel-writer/scripts/data_modules/tests/test_migrate_state_to_sqlite.py`
 
-- [ ] **Step 1: 写测试**：构造一条会迁移失败的实体（如非法类型触发 `stats["errors"] += 1`），跑迁移，断言 state.json 中 `entities_v3` 字段仍在、CLI 退出码非 0。
-- [ ] **Step 2: 实现**：`if stats["errors"]: 跳过步骤5精简，输出"存在迁移错误，已保留原字段"`；步骤 5 的裸 `open('w')+json.dump` 改为 `security_utils.atomic_write_json(state_path, state, use_lock=True)`。
-- [ ] **Step 3: 提交** `fix: migration never prunes state on partial failure`。
+- [x] **Step 1: 写测试**：构造一条会迁移失败的实体（如非法类型触发 `stats["errors"] += 1`），跑迁移，断言 state.json 中 `entities_v3` 字段仍在、CLI 退出码非 0。
+- [x] **Step 2: 实现**：`if stats["errors"]: 跳过步骤5精简，输出"存在迁移错误，已保留原字段"`；步骤 5 的裸 `open('w')+json.dump` 改为 `security_utils.atomic_write_json(state_path, state, use_lock=True)`。
+- [x] **Step 3: 提交** `fix: migration never prunes state on partial failure`。
 
 ### Task 6: archive_manager 原子写 + 恢复顺序反转
 

webnovel-writer/scripts/data_modules/migrate_state_to_sqlite.py

@@ -34,6 +34,7 @@ from typing import Dict, Any, List
 
 from .config import get_config, DataModulesConfig
 from .sql_state_manager import SQLStateManager, EntityData
+from security_utils import atomic_write_json
 
 
 def migrate_state_to_sqlite(
@@ -233,7 +234,10 @@ def migrate_state_to_sqlite(
         print(f"  ✅ 关系: {stats['relationships']} 条")
 
     # 5. 精简 state.json（移除已迁移字段）
-    if not dry_run:
+    if stats["errors"]:
+        if verbose:
+            print("\n⚠️ 存在迁移错误，已保留原字段")
+    elif not dry_run:
         if verbose:
             print(f"\n🔄 精简 state.json...")
 
@@ -254,8 +258,7 @@ def migrate_state_to_sqlite(
             "_migration_timestamp": datetime.now().isoformat()
         }
 
-        with open(state_file, 'w', encoding='utf-8') as f:
-            json.dump(slim_state, f, ensure_ascii=False, indent=2)
+        atomic_write_json(state_file, slim_state, use_lock=True)
 
         new_size = state_file.stat().st_size / 1024
         if verbose:

webnovel-writer/scripts/data_modules/tests/test_migrate_state_to_sqlite.py

@@ -172,7 +172,7 @@ def test_migrate_state_backup_and_skips(temp_project):
     assert backups
 
 
-def test_migrate_state_error_branches(tmp_path, monkeypatch):
+def test_migrate_state_error_branches(tmp_path, monkeypatch, capsys):
     cfg = DataModulesConfig.from_project_root(tmp_path)
     cfg.ensure_dirs()
     state = {
@@ -210,5 +210,51 @@ def test_migrate_state_error_branches(tmp_path, monkeypatch):
 
     monkeypatch.setattr(migrate_module, "SQLStateManager", BoomSQL)
 
-    stats = migrate_state_to_sqlite(cfg, dry_run=False, backup=False, verbose=False)
+    stats = migrate_state_to_sqlite(cfg, dry_run=False, backup=False, verbose=True)
+    output = capsys.readouterr().out
     assert stats["errors"] >= 4
+    assert "存在迁移错误，已保留原字段" in output
+
+
+def test_migrate_cli_preserves_state_fields_on_partial_failure(tmp_path, monkeypatch, capsys):
+    cfg = DataModulesConfig.from_project_root(tmp_path)
+    cfg.ensure_dirs()
+    state = {
+        "entities_v3": {"角色": {"boom": {"canonical_name": "爆"}}},
+        "alias_index": {},
+        "state_changes": [],
+        "structured_relationships": [],
+        "relationships": {},
+        "world_settings": {},
+        "plot_threads": {},
+        "review_checkpoints": [],
+        "project_info": {},
+    }
+    cfg.state_file.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    class BoomSQL:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def upsert_entity(self, *args, **kwargs):
+            raise RuntimeError("boom")
+
+    monkeypatch.setattr(migrate_module, "SQLStateManager", BoomSQL)
+    monkeypatch.setattr(
+        "sys.argv",
+        [
+            "migrate_state_to_sqlite",
+            "--project-root",
+            str(tmp_path),
+            "--no-backup",
+        ],
+    )
+
+    with pytest.raises(SystemExit) as exc:
+        migrate_module.main()
+
+    assert exc.value.code == 1
+    output = json.loads(capsys.readouterr().out)
+    assert output.get("status") == "error"
+    saved = json.loads(cfg.state_file.read_text(encoding="utf-8"))
+    assert "entities_v3" in saved