fix: harden cleanup safety and snapshot concurrency

.claude/agents/consistency-checker.md

@@ -1,7 +1,7 @@
 ---
 name: consistency-checker
 description: 设定一致性检查，输出结构化报告供润色步骤参考
-tools: Read, Grep
+tools: Read, Grep, Bash
 ---
 
 # consistency-checker (设定一致性检查器)

.claude/agents/pacing-checker.md

@@ -1,7 +1,7 @@
 ---
 name: pacing-checker
 description: Strand Weave 节奏检查，输出结构化报告供润色步骤参考
-tools: Read, Grep
+tools: Read, Grep, Bash
 ---
 
 # pacing-checker (节奏检查器)

.claude/references/context-contract-v2.md

@@ -32,7 +32,7 @@
   - 聚合审查趋势与低分区间（`review_trend` / `low_score_ranges`）
 - `genre_profile`
   - 基于 `state.json -> project.genre` 自动选取题材策略片段
-  - 引用 `.claude/references/genre-profiles.md` 与 `reading-power-taxonomy.md`
+  - 引用 `.claude/references/genre-profiles.md` 与 `.claude/references/reading-power-taxonomy.md`
   - 输出 `reference_hints` 供 Writer 快速执行
 
 ## Phase C 扩展段

.claude/scripts/data_modules/snapshot_manager.py

@@ -8,11 +8,19 @@ from __future__ import annotations
 import json
 from dataclasses import dataclass
 from datetime import datetime, timezone
+from filelock import FileLock
 from pathlib import Path
 from typing import Any, Dict, Optional
 
 from .config import get_config
 
+try:
+    # 当 scripts 目录在 sys.path 中
+    from security_utils import atomic_write_json
+except ImportError:  # pragma: no cover
+    # 当以 python -m scripts.data_modules... 形式运行
+    from scripts.security_utils import atomic_write_json
+
 SNAPSHOT_VERSION = "1.1"
 
 
@@ -40,6 +48,9 @@ class SnapshotManager:
     def _snapshot_path(self, chapter: int) -> Path:
         return self.snapshot_dir / f"ch{chapter:04d}.json"
 
+    def _snapshot_lock_path(self, chapter: int) -> Path:
+        return self._snapshot_path(chapter).with_suffix(".json.lock")
+
     def save_snapshot(self, chapter: int, payload: Dict[str, Any], meta: Optional[Dict[str, Any]] = None) -> Path:
         data: Dict[str, Any] = {
             "version": self.version,
@@ -51,14 +62,18 @@ class SnapshotManager:
             data["meta"] = meta
 
         path = self._snapshot_path(chapter)
-        path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+        lock = FileLock(str(self._snapshot_lock_path(chapter)), timeout=10)
+        with lock:
+            atomic_write_json(path, data, use_lock=False, backup=False)
         return path
 
     def load_snapshot(self, chapter: int) -> Optional[Dict[str, Any]]:
         path = self._snapshot_path(chapter)
-        if not path.exists():
-            return None
-        data = json.loads(path.read_text(encoding="utf-8"))
+        lock = FileLock(str(self._snapshot_lock_path(chapter)), timeout=10)
+        with lock:
+            if not path.exists():
+                return None
+            data = json.loads(path.read_text(encoding="utf-8"))
         version = str(data.get("version", ""))
         if version != self.version:
             raise SnapshotVersionMismatch(self.version, version)
@@ -66,9 +81,11 @@ class SnapshotManager:
 
     def delete_snapshot(self, chapter: int) -> bool:
         path = self._snapshot_path(chapter)
-        if path.exists():
-            path.unlink()
-            return True
+        lock = FileLock(str(self._snapshot_lock_path(chapter)), timeout=10)
+        with lock:
+            if path.exists():
+                path.unlink()
+                return True
         return False
 
     def list_snapshots(self) -> list[str]:

.claude/scripts/data_modules/tests/test_context_manager.py

@@ -44,6 +44,14 @@ def test_snapshot_version_mismatch(temp_project):
         other.load_snapshot(1)
 
 
+def test_snapshot_delete_roundtrip(temp_project):
+    manager = SnapshotManager(temp_project)
+    manager.save_snapshot(2, {"x": 1})
+
+    assert manager.delete_snapshot(2) is True
+    assert manager.load_snapshot(2) is None
+
+
 def test_context_manager_build_and_filter(temp_project):
     state = {
         "protagonist_state": {"name": "萧炎", "location": {"current": "天云宗"}},

.claude/scripts/data_modules/tests/test_workflow_manager.py

@@ -5,6 +5,7 @@ import json
 import logging
 import sys
 from pathlib import Path
+from types import SimpleNamespace
 
 
 def _load_module():
@@ -134,3 +135,61 @@ def test_workflow_reentry_does_not_duplicate_history(tmp_path, monkeypatch):
 
     task = state.get("current_task") or {}
     assert int(task.get("retry_count", 0)) >= 2
+
+
+def test_cleanup_artifacts_requires_confirm(tmp_path, monkeypatch):
+    module = _load_module()
+    monkeypatch.setattr(module, "find_project_root", lambda: tmp_path)
+
+    webnovel_dir = tmp_path / ".webnovel"
+    webnovel_dir.mkdir(parents=True, exist_ok=True)
+
+    draft_path = module.default_chapter_draft_path(tmp_path, 7)
+    draft_path.parent.mkdir(parents=True, exist_ok=True)
+    draft_path.write_text("draft", encoding="utf-8")
+
+    git_called = {"count": 0}
+
+    def _fake_run(*args, **kwargs):
+        git_called["count"] += 1
+        return SimpleNamespace(returncode=0, stderr="", stdout="")
+
+    monkeypatch.setattr(module.subprocess, "run", _fake_run)
+
+    preview = module.cleanup_artifacts(7, confirm=False)
+
+    assert draft_path.exists()
+    assert git_called["count"] == 0
+    assert any(item.startswith("[预览]") for item in preview)
+
+
+def test_cleanup_artifacts_confirm_deletes_with_backup(tmp_path, monkeypatch):
+    module = _load_module()
+    monkeypatch.setattr(module, "find_project_root", lambda: tmp_path)
+
+    webnovel_dir = tmp_path / ".webnovel"
+    webnovel_dir.mkdir(parents=True, exist_ok=True)
+
+    draft_path = module.default_chapter_draft_path(tmp_path, 8)
+    draft_path.parent.mkdir(parents=True, exist_ok=True)
+    draft_path.write_text("draft", encoding="utf-8")
+
+    git_called = {"count": 0, "cmd": None}
+
+    def _fake_run(cmd, **kwargs):
+        git_called["count"] += 1
+        git_called["cmd"] = cmd
+        return SimpleNamespace(returncode=0, stderr="", stdout="")
+
+    monkeypatch.setattr(module.subprocess, "run", _fake_run)
+
+    cleaned = module.cleanup_artifacts(8, confirm=True)
+
+    assert not draft_path.exists()
+    assert git_called["count"] == 1
+    assert git_called["cmd"] == ["git", "reset", "HEAD", "."]
+    assert any("Git 暂存区已清理" in item for item in cleaned)
+
+    backup_dir = tmp_path / ".webnovel" / "recovery_backups"
+    backups = list(backup_dir.glob("ch0008-*"))
+    assert backups

.claude/scripts/requirements.txt

@@ -9,3 +9,5 @@ pydantic>=2.0.0         # Schema 校验
 # 可选依赖（开发/测试）
 pytest>=7.0.0           # 单元测试
 pytest-cov>=4.1.0       # 覆盖率统计
+pytest-asyncio>=0.23.0  # 异步测试支持
+pytest-timeout>=2.3.0   # 测试超时保护

.claude/scripts/workflow_manager.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 import json
 import logging
 import os
+import shutil
 import subprocess
 import sys
 from datetime import datetime
@@ -545,9 +546,22 @@ def analyze_recovery_options(interrupt_info):
     ]
 
 
-def cleanup_artifacts(chapter_num):
+def _backup_chapter_for_cleanup(project_root: Path, chapter_num: int, chapter_path: Path) -> Path:
+    """Backup chapter file before destructive cleanup."""
+    backup_dir = project_root / ".webnovel" / "recovery_backups"
+    create_secure_directory(str(backup_dir))
+
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    backup_name = f"ch{chapter_num:04d}-{chapter_path.name}.{timestamp}.bak"
+    backup_path = backup_dir / backup_name
+    shutil.copy2(chapter_path, backup_path)
+    return backup_path
+
+
+def cleanup_artifacts(chapter_num, *, confirm: bool = False):
     """Cleanup partial artifacts."""
     artifacts_cleaned = []
+    planned_actions = []
 
     project_root = find_project_root()
 
@@ -558,22 +572,60 @@ def cleanup_artifacts(chapter_num):
             chapter_path = draft_path
 
     if chapter_path and chapter_path.exists():
+        planned_actions.append(f"删除章节文件: {chapter_path.relative_to(project_root)}")
+
+    planned_actions.append("重置 Git 暂存区: git reset HEAD .")
+
+    if not confirm:
+        preview_items = [f"[预览] {action}" for action in planned_actions]
+        safe_append_call_trace(
+            "artifacts_cleanup_preview",
+            {
+                "chapter": chapter_num,
+                "planned_actions": planned_actions,
+                "confirmed": False,
+            },
+        )
+        print("⚠️ 检测到高风险清理操作，当前仅预览。若确认执行，请追加 --confirm。")
+        return preview_items or ["[预览] 无可清理项"]
+
+    if chapter_path and chapter_path.exists():
+        try:
+            backup_path = _backup_chapter_for_cleanup(project_root, chapter_num, chapter_path)
+        except OSError as exc:
+            error_msg = f"❌ 章节备份失败，已取消删除: {exc}"
+            safe_append_call_trace(
+                "artifacts_cleanup_backup_failed",
+                {
+                    "chapter": chapter_num,
+                    "chapter_file": str(chapter_path),
+                    "error": str(exc),
+                },
+            )
+            return [error_msg]
+
         chapter_path.unlink()
         artifacts_cleaned.append(str(chapter_path.relative_to(project_root)))
+        artifacts_cleaned.append(f"章节备份已保存: {backup_path.relative_to(project_root)}")
 
     result = subprocess.run(["git", "reset", "HEAD", "."], cwd=project_root, capture_output=True, text=True)
     if result.returncode == 0:
         artifacts_cleaned.append("Git 暂存区已清理（project）")
+    else:
+        git_error = (result.stderr or "").strip() or "unknown error"
+        artifacts_cleaned.append(f"⚠️ Git 暂存区清理失败: {git_error}")
 
     safe_append_call_trace(
         "artifacts_cleaned",
         {
             "chapter": chapter_num,
             "items": artifacts_cleaned,
+            "planned_actions": planned_actions,
+            "confirmed": True,
             "git_reset_ok": result.returncode == 0,
         },
     )
-    return artifacts_cleaned
+    return artifacts_cleaned or ["无可清理项"]
 
 
 def clear_current_task():
@@ -689,6 +741,7 @@ if __name__ == "__main__":
 
     p_cleanup = subparsers.add_parser("cleanup", help="清理 artifacts")
     p_cleanup.add_argument("--chapter", type=int, required=True, help="章节号")
+    p_cleanup.add_argument("--confirm", action="store_true", help="确认执行删除与 Git 重置（高风险）")
 
     subparsers.add_parser("clear", help="清除中断任务")
 
@@ -715,8 +768,13 @@ if __name__ == "__main__":
         else:
             print("✅ 无中断任务")
     elif args.action == "cleanup":
-        cleaned = cleanup_artifacts(args.chapter)
-        print(f"✅ 已清理: {', '.join(cleaned)}")
+        cleaned = cleanup_artifacts(args.chapter, confirm=args.confirm)
+        if args.confirm:
+            print(f"✅ 已清理: {', '.join(cleaned)}")
+        else:
+            for item in cleaned:
+                print(item)
+            print("⚠️ 以上为预览，未执行实际清理。")
     elif args.action == "clear":
         clear_current_task()
     else:

.claude/skills/webnovel-init/references/system-data-flow.md

@@ -39,6 +39,6 @@ cat "${CLAUDE_PLUGIN_ROOT}/skills/webnovel-query/references/system-data-flow.md"
 - **双 Agent 架构**: Context Agent (读) + Data Agent (写)
 - **无 XML 标签**: 纯正文写作，Data Agent AI 自动提取实体
 - **SQLite 存储**: entities/aliases/state_changes 迁移到 index.db
-- **state.json 精简**: 保持 < 5KB，仅存 protagonist_state 和 plot_threads
+- **state.json 精简**: 保持 < 5KB，主要包含 progress/protagonist_state/strand_tracker/disambiguation
 
 </instructions>

.claude/skills/webnovel-resume/SKILL.md

@@ -126,7 +126,7 @@ B) 回滚到Ch6，放弃Ch7所有进度
 
 **选项 A - 删除重来**（推荐）：
 ```bash
-python "${CLAUDE_PLUGIN_ROOT}/scripts/workflow_manager.py" cleanup --chapter {N}
+python "${CLAUDE_PLUGIN_ROOT}/scripts/workflow_manager.py" cleanup --chapter {N} --confirm
 python "${CLAUDE_PLUGIN_ROOT}/scripts/workflow_manager.py" clear
 ```
 

.claude/skills/webnovel-resume/references/workflow-resume.md

@@ -44,7 +44,7 @@ python "${CLAUDE_PLUGIN_ROOT}/scripts/workflow_manager.py" detect
 
 **选项 A（推荐）**: 删除半成品重新开始
 ```bash
-python "${CLAUDE_PLUGIN_ROOT}/scripts/workflow_manager.py" cleanup --chapter {N}
+python "${CLAUDE_PLUGIN_ROOT}/scripts/workflow_manager.py" cleanup --chapter {N} --confirm
 python "${CLAUDE_PLUGIN_ROOT}/scripts/workflow_manager.py" clear
 /webnovel-write {N}
 ```