feat: guard frozen split products against silent overwrite (--force)

app/harness/build_split.py

@@ -12,6 +12,7 @@ test，再以视频组为原子切出诊断 / 验证池，原子冻结 pools.jso
 from __future__ import annotations
 
 import hashlib
+import json
 import sqlite3
 from collections import Counter, defaultdict
 from dataclasses import asdict, dataclass, field
@@ -104,6 +105,38 @@ class SplitBuildResult:
         return getattr(self, key)
 
 
+def _guard_frozen_products(out_path: Path, manifest_path: Path, *, force: bool) -> None:
+    """冻结前的覆盖保护：产物已存在时按 force 决定报错或备份。
+
+    参数:
+        out_path: 目标 pools.json 路径。
+        manifest_path: 目标 split_manifest.json 路径。
+        force: False 时已存在即 FileExistsError；True 时把旧产物重命名为
+            .bak.<旧 pools_sha256 前 8 位或 'prev'> 再放行。
+
+    异常:
+        FileExistsError: force=False 且产物已存在（防静默覆盖冻结锚点）。
+    """
+    if not out_path.exists() and not manifest_path.exists():
+        return
+    if not force:
+        raise FileExistsError(
+            f"已存在冻结产物 {out_path}（或其 manifest）。重跑切分会覆盖训练依赖的"
+            "冻结锚点——确认要替换请加 --force（旧产物将备份为 .bak.*）。"
+        )
+    # 备份后缀取旧 manifest 的 pools_sha256 前 8 位，无则用 'prev'
+    suffix = "prev"
+    if manifest_path.exists():
+        try:
+            old = json.loads(manifest_path.read_text(encoding="utf-8"))
+            suffix = str(old.get("pools_sha256", "prev"))[:8] or "prev"
+        except (json.JSONDecodeError, OSError):
+            suffix = "prev"
+    for p in (out_path, manifest_path):
+        if p.exists():
+            p.rename(p.with_name(f"{p.name}.bak.{suffix}"))
+
+
 def build_split(
     *,
     db_path: Path,
@@ -115,6 +148,7 @@ def build_split(
     out_path: Path,
     manifest_path: Path,
     generated_at: str,
+    force: bool = False,
 ) -> SplitBuildResult:
     """顶层编排结果驱动视频级切分，冻结 pools.json + manifest 并跑防御断言。
 
@@ -137,6 +171,9 @@ def build_split(
         out_path: 冻结 pools.json 目标路径（原子写）。
         manifest_path: 溯源 manifest 目标路径（原子写）。
         generated_at: 生成时间戳（ISO 字符串），由调用方传入以保证可复现。
+        force: 覆盖保护开关。False（默认）时若 out_path/manifest_path 已存在即
+            FileExistsError（防静默覆盖训练依赖的冻结锚点）；True 时先把旧产物备份为
+            .bak.* 再放行覆盖。
 
     返回:
         SplitBuildResult，含 pools / manifest / assignment，支持字典式访问。
@@ -199,6 +236,7 @@ def build_split(
         val_wrong_min=config.val_wrong_min,
         wrong_tier_by_video=dict(wrong_tier_by_video),
     )
+    _guard_frozen_products(out_path, manifest_path, force=force)
     save_pools(pools, out_path)
 
     # Phase 4: 溯源 manifest（pools_sha256 锚定冻结内容）。

app/harness/video_split_cli.py

@@ -519,6 +519,7 @@ async def run_pipeline(
     questions_dir: Path,
     out_dir: Path,
     generated_at: str,
+    force: bool = False,
 ) -> SplitBuildResult:
     """内联三阶段：Phase 0 INFRA T0 补录 → Phase 1 诊断 → Phase 2 冻结切分 → McNemar 护栏。
 
@@ -534,6 +535,7 @@ async def run_pipeline(
         questions_dir: benchmark 题库目录（Phase 2 加载题库切池）。
         out_dir: 冻结产物目录（pools.json + split_manifest.json）。
         generated_at: 生成时间戳（ISO 字符串，由调用方传入；见模块 C-2 复现锚点约定）。
+        force: 覆盖已存在冻结产物开关，透传给 build_split（False 时已存在即报错）。
 
     返回:
         SplitBuildResult（冻结三池 + manifest + assignment）。
@@ -577,6 +579,7 @@ async def run_pipeline(
         out_path=out_dir / "pools.json",
         manifest_path=out_dir / "split_manifest.json",
         generated_at=generated_at,
+        force=force,
     )
 
     # McNemar 功效护栏（build_split 契约外的 capstone 层校验）。
@@ -650,6 +653,7 @@ def _execute_real(config: VideoSplitConfig, fingerprint: str, args: argparse.Nam
                 questions_dir=questions_dir,
                 out_dir=out_dir,
                 generated_at=generated_at,
+                force=args.force,
             )
         )
     finally:
@@ -773,6 +777,11 @@ def build_arg_parser() -> argparse.ArgumentParser:
             "对 manifest 做字节级复现比对。"
         ),
     )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="覆盖已存在的冻结 pools.json/manifest（旧产物备份为 .bak.*）",
+    )
     return parser
 
 

tests/unit/test_video_split_cli.py

@@ -303,3 +303,29 @@ def test_git_short_sha_nonempty():
     sha = cli.git_short_sha()
     assert sha
     assert len(sha) >= 4
+
+
+def test_build_split_refuses_overwrite_without_force(tmp_path):
+    """已存在指纹不同的 pools.json 时，force=False 必须报错不覆盖。"""
+    from app.harness.build_split import _guard_frozen_products
+
+    out_path = tmp_path / "pools.json"
+    out_path.write_text('{"split_mode":"global"}', encoding="utf-8")
+    manifest_path = tmp_path / "split_manifest.json"
+
+    with pytest.raises(FileExistsError, match="已存在冻结产物"):
+        _guard_frozen_products(out_path, manifest_path, force=False)
+
+
+def test_build_split_force_backs_up_old(tmp_path):
+    """force=True 时旧产物被备份为 .bak.* 再允许覆盖。"""
+    from app.harness.build_split import _guard_frozen_products
+
+    out_path = tmp_path / "pools.json"
+    out_path.write_text('{"old":1}', encoding="utf-8")
+    manifest_path = tmp_path / "split_manifest.json"
+    manifest_path.write_text('{"pools_sha256":"deadbeef00000000"}', encoding="utf-8")
+
+    _guard_frozen_products(out_path, manifest_path, force=True)
+    baks = list(tmp_path.glob("pools.json.bak.*"))
+    assert len(baks) == 1, f"未备份旧产物: {list(tmp_path.iterdir())}"