fix(wren): validate import paths before force cleanup (#2580)

Co-authored-by: GiacomoZGW <dmitryprozorova145@gmail.com>

core/wren/src/wren/context.py

@@ -317,6 +317,18 @@ def write_project_files(
         force: If False, raise SystemExit if any target file already exists.
     """
     output_dir = Path(output_dir)
+    root = output_dir.resolve()
+    resolved_files: list[tuple[ProjectFile, Path]] = []
+
+    for file in files:
+        target = (output_dir / file.relative_path).resolve()
+        try:
+            target.relative_to(root)
+        except ValueError:
+            raise SystemExit(f"Error: invalid output path: {file.relative_path!r}")
+        if target == root:
+            raise SystemExit(f"Error: invalid output path: {file.relative_path!r}")
+        resolved_files.append((file, target))
 
     if force and output_dir.exists():
         import shutil  # noqa: PLC0415
@@ -341,7 +353,7 @@ def write_project_files(
 
     if not force:
         conflicts = [
-            f.relative_path for f in files if (output_dir / f.relative_path).exists()
+            file.relative_path for file, target in resolved_files if target.exists()
         ]
         if conflicts:
             names = ", ".join(f"'{Path(p).name}'" for p in conflicts)
@@ -349,15 +361,9 @@ def write_project_files(
                 f"Error: {names} already exists. Use --force to overwrite."
             )
 
-    for f in files:
-        root = output_dir.resolve()
-        path = (output_dir / f.relative_path).resolve()
-        try:
-            path.relative_to(root)
-        except ValueError:
-            raise SystemExit(f"Error: invalid output path: {f.relative_path!r}")
-        path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_text(f.content)
+    for file, target in resolved_files:
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(file.content)
 
 
 # ── Project discovery ─────────────────────────────────────────────────────

core/wren/tests/unit/test_convert_mdl.py

@@ -11,6 +11,7 @@ import yaml
 from wren.context import (
     _AGENTS_MD_TEMPLATE,
     _CAMEL_TO_SNAKE_MAP,
+    ProjectFile,
     _camel_to_snake,
     _snake_to_camel,
     build_json,
@@ -228,6 +229,63 @@ def test_write_project_files_force_overwrites(tmp_path: Path):
     assert project["schema_version"] == 2
 
 
+def test_write_project_files_force_invalid_path_preserves_existing_files(
+    tmp_path: Path,
+):
+    models_dir = tmp_path / "models"
+    models_dir.mkdir()
+    sentinel = models_dir / "keep.txt"
+    sentinel.write_text("important")
+    escaped_path = tmp_path.parent / f"{tmp_path.name}-escape.txt"
+
+    files = [
+        ProjectFile(relative_path=f"../{escaped_path.name}", content="invalid"),
+    ]
+
+    with pytest.raises(SystemExit, match="invalid output path"):
+        write_project_files(files, tmp_path, force=True)
+
+    assert sentinel.read_text() == "important"
+    assert not escaped_path.exists()
+
+
+def test_write_project_files_preflights_all_paths_before_writing(tmp_path: Path):
+    escaped_path = tmp_path.parent / f"{tmp_path.name}-escape.txt"
+    files = [
+        ProjectFile(relative_path="models/orders/metadata.yml", content="valid"),
+        ProjectFile(relative_path=f"../{escaped_path.name}", content="invalid"),
+    ]
+
+    with pytest.raises(SystemExit, match="invalid output path"):
+        write_project_files(files, tmp_path)
+
+    assert not (tmp_path / "models").exists()
+    assert not escaped_path.exists()
+
+
+@pytest.mark.parametrize("relative_path", ["", ".", "models/.."])
+def test_write_project_files_force_root_target_preserves_existing_files(
+    tmp_path: Path,
+    relative_path: str,
+):
+    models_dir = tmp_path / "models"
+    models_dir.mkdir()
+    sentinel = models_dir / "keep.txt"
+    sentinel.write_text("important")
+
+    with pytest.raises(SystemExit) as exc_info:
+        write_project_files(
+            [ProjectFile(relative_path=relative_path, content="invalid")],
+            tmp_path,
+            force=True,
+        )
+
+    assert str(exc_info.value) == f"Error: invalid output path: {relative_path!r}"
+    assert tmp_path.is_dir()
+    assert sentinel.read_text() == "important"
+    assert set(tmp_path.rglob("*")) == {models_dir, sentinel}
+
+
 # ── Round-trip: convert → build ────────────────────────────────────────────