diff --git a/core/wren/src/wren/context.py b/core/wren/src/wren/context.py
index a35829d055..e062d04533 100644
--- a/core/wren/src/wren/context.py
+++ b/core/wren/src/wren/context.py
@@ -1493,6 +1493,86 @@ class UpgradeError(Exception):
     """Raised when a project upgrade cannot proceed."""
 
 
+class _UniqueKeySafeLoader(yaml.SafeLoader):
+    """SafeLoader variant that rejects explicit duplicate mapping keys."""
+
+    def construct_mapping(self, node, deep=False):
+        """Reject explicit duplicates without breaking YAML merge-key semantics."""
+        seen = set()
+        for key_node, _ in node.value:
+            # SafeConstructor.flatten_mapping() handles << later. Checking the
+            # merge node itself here would try to construct the special merge
+            # tag too early and reject YAML that yaml.safe_load() accepts.
+            if key_node.tag == "tag:yaml.org,2002:merge":
+                continue
+            key = self.construct_object(key_node, deep=deep)
+            try:
+                if key in seen:
+                    raise yaml.constructor.ConstructorError(
+                        "while constructing a mapping",
+                        node.start_mark,
+                        f"found duplicate YAML key {key!r}",
+                        key_node.start_mark,
+                    )
+                seen.add(key)
+            except TypeError as exc:
+                raise yaml.constructor.ConstructorError(
+                    "while constructing a mapping",
+                    node.start_mark,
+                    f"found unhashable YAML key {key!r}",
+                    key_node.start_mark,
+                ) from exc
+        return super().construct_mapping(node, deep=deep)
+
+
+def _collect_yaml_node_ids(node, seen=None) -> set[int]:
+    """Collect node identities below a composed YAML node, guarding cycles."""
+    if seen is None:
+        seen = set()
+    node_id = id(node)
+    if node_id in seen:
+        return seen
+    seen.add(node_id)
+
+    if isinstance(node, yaml.nodes.MappingNode):
+        for key_node, value_node in node.value:
+            _collect_yaml_node_ids(key_node, seen)
+            _collect_yaml_node_ids(value_node, seen)
+    elif isinstance(node, yaml.nodes.SequenceNode):
+        for item in node.value:
+            _collect_yaml_node_ids(item, seen)
+    return seen
+
+
+def _v1_views_anchor_root_keys(text: str) -> set[str]:
+    """Return extra root keys whose value node is aliased inside views."""
+    root = yaml.compose(text, Loader=yaml.SafeLoader)
+    if not isinstance(root, yaml.nodes.MappingNode):
+        return set()
+
+    views_node = None
+    for key_node, value_node in root.value:
+        if (
+            isinstance(key_node, yaml.nodes.ScalarNode)
+            and key_node.value == "views"
+        ):
+            views_node = value_node
+            break
+    if views_node is None:
+        return set()
+
+    referenced_ids = _collect_yaml_node_ids(views_node)
+    anchor_keys: set[str] = set()
+    for key_node, value_node in root.value:
+        if (
+            isinstance(key_node, yaml.nodes.ScalarNode)
+            and key_node.value != "views"
+            and id(value_node) in referenced_ids
+        ):
+            anchor_keys.add(key_node.value)
+    return anchor_keys
+
+
 def _resolve_upgrade_directory(
     project_path: Path,
     collection: str,
@@ -1556,6 +1636,19 @@ def _resolve_upgrade_file(target_directory: Path, filename: str) -> Path:
     return resolved_file
 
 
+def _record_unique_upgrade_target(
+    seen: dict[str, str], target: str, source_kind: str
+) -> None:
+    """Reject exact and case-insensitive v1→v2 destination collisions."""
+    key = target.casefold()
+    previous = seen.get(key)
+    if previous is not None:
+        detail = f"'{target}'" if previous == target else f"'{previous}' and '{target}'"
+        raise UpgradeError(
+            f"Cannot upgrade: multiple legacy {source_kind} map to {detail}"
+        )
+    seen[key] = target
+
 def _validate_upgrade_view_statement(name: str, statement: Any) -> str | None:
     """Reject malformed legacy view statements before migration."""
     if statement is not None and not isinstance(statement, str):
@@ -1693,6 +1786,63 @@ def _reject_malformed_v1_model_columns(project_path: Path) -> None:
         )
 
 
+def _reject_malformed_v1_views(project_path: Path) -> None:
+    """Abort v1→v2 upgrade if ``views.yml`` contains content the loader drops.
+
+    Loader normalisation is correct for validation/runtime, but migration
+    deletes ``views.yml`` after writing only the surviving entries. Inspect
+    the raw source before migration so malformed or nameless views are never
+    silently discarded.
+    """
+    views_file = project_path / "views.yml"
+    if not views_file.exists():
+        return
+
+    source_text = views_file.read_text(encoding="utf-8")
+    try:
+        anchor_root_keys = _v1_views_anchor_root_keys(source_text)
+        raw = yaml.load(source_text, Loader=_UniqueKeySafeLoader)
+    except (TypeError, yaml.YAMLError) as exc:
+        raise UpgradeError(f"Cannot upgrade: invalid views.yml: {exc}") from exc
+
+    problems: list[str] = []
+
+    if raw is not None and not isinstance(raw, dict):
+        problems.append(
+            f"views.yml: must be a mapping with a 'views' key, got {type(raw).__name__}"
+        )
+    else:
+        raw_views = None
+        if isinstance(raw, dict):
+            unexpected_keys = set(raw) - {"views"} - anchor_root_keys
+            if unexpected_keys:
+                problems.append(
+                    "views.yml: unsupported root keys are not preserved by migration: "
+                    + ", ".join(sorted(str(key) for key in unexpected_keys))
+                )
+            raw_views = raw.get("views")
+
+        if raw_views is not None and not isinstance(raw_views, list):
+            problems.append(
+                f"views.yml > views: must be a list, got {type(raw_views).__name__}"
+            )
+        elif isinstance(raw_views, list):
+            for i, view in enumerate(raw_views):
+                if not isinstance(view, dict):
+                    problems.append(
+                        f"views.yml > views[{i}]: view entry must be a mapping, "
+                        f"got {type(view).__name__}"
+                    )
+                elif not view.get("name"):
+                    problems.append(f"views.yml > views[{i}]: view missing 'name'")
+
+    if problems:
+        detail = "; ".join(problems)
+        raise UpgradeError(
+            "Cannot upgrade: malformed views would be discarded — " + detail
+        )
+
+
 def _plan_v1_to_v2(project_path: Path) -> tuple[list[str], list[str]]:
     """Plan the v1→v2 file restructuring. Returns (files_created, files_deleted)."""
     created: list[str] = []
@@ -1703,8 +1853,10 @@ def _plan_v1_to_v2(project_path: Path) -> tuple[list[str], list[str]]:
     # (drops non-list / non-dict entries), and _apply_v1_to_v2 then unlinks the
     # v1 source. Abort in preflight so upgrade cannot silently discard data.
     _reject_malformed_v1_model_columns(project_path)
+    _reject_malformed_v1_views(project_path)
 
     # Models: flat files → directories
+    seen_model_targets: dict[str, str] = {}
     models = _load_models_v1(project_path)
     for model in models:
         source_dir = model.pop("_source_dir", None)
@@ -1717,12 +1869,17 @@ def _plan_v1_to_v2(project_path: Path) -> tuple[list[str], list[str]]:
             created.append(ref_sql_file.relative_to(project_root).as_posix())
 
         metadata_file = _resolve_upgrade_file(model_dir, "metadata.yml")
-        created.append(metadata_file.relative_to(project_root).as_posix())
+        metadata_target = metadata_file.relative_to(project_root).as_posix()
+        _record_unique_upgrade_target(
+            seen_model_targets, metadata_target, "model files"
+        )
+        created.append(metadata_target)
 
         if source_dir:
             deleted.append(f"models/{source_dir}.yml")
 
     # Views: single file → directories
+    seen_view_targets: dict[str, str] = {}
     views = _load_views_v1(project_path)
     for view in views:
         name = view.get("name")
@@ -1736,14 +1893,16 @@ def _plan_v1_to_v2(project_path: Path) -> tuple[list[str], list[str]]:
             created.append(sql_file.relative_to(project_root).as_posix())
 
         metadata_file = _resolve_upgrade_file(view_dir, "metadata.yml")
-        created.append(metadata_file.relative_to(project_root).as_posix())
+        metadata_target = metadata_file.relative_to(project_root).as_posix()
+        _record_unique_upgrade_target(seen_view_targets, metadata_target, "views")
+        created.append(metadata_target)
 
     views_file = project_path / "views.yml"
     if views_file.exists():
         deleted.append("views.yml")
 
     # Cubes: flat files → directories
-    seen_cube_targets: set[str] = set()
+    seen_cube_targets: dict[str, str] = {}
     cubes = _load_cubes_v1(project_path)
     for cube in cubes:
         source_file = cube.pop("_source_file", None)
@@ -1751,11 +1910,9 @@ def _plan_v1_to_v2(project_path: Path) -> tuple[list[str], list[str]]:
         cube_dir = _resolve_upgrade_directory(project_path, "cubes", "cube", name)
         metadata_file = _resolve_upgrade_file(cube_dir, "metadata.yml")
         target = metadata_file.relative_to(project_root).as_posix()
-        if target in seen_cube_targets:
-            raise UpgradeError(
-                f"Cannot upgrade: multiple legacy cube files map to '{target}'"
-            )
-        seen_cube_targets.add(target)
+        _record_unique_upgrade_target(
+            seen_cube_targets, target, "cube files"
+        )
 
         created.append(target)
 
diff --git a/core/wren/tests/unit/test_context.py b/core/wren/tests/unit/test_context.py
index 9dcf68a97d..ce548776cd 100644
--- a/core/wren/tests/unit/test_context.py
+++ b/core/wren/tests/unit/test_context.py
@@ -1347,6 +1347,15 @@ def _assert_v1_sources_unchanged(
     assert get_schema_version(project_path) == 1
 
 
+def _assert_no_v2_entity_directories(project_path: Path) -> None:
+    """Failed v1→v2 apply must not leave any migrated entity directories."""
+    for collection in ("models", "views", "cubes"):
+        root = project_path / collection
+        if not root.is_dir():
+            continue
+        assert not [entry for entry in root.iterdir() if entry.is_dir()]
+
+
 def test_plan_upgrade_v1_to_v2(tmp_path):
     _make_v1_project(tmp_path)
     result = plan_upgrade(tmp_path, target_version=2)
@@ -1594,13 +1603,304 @@ def test_validate_project_reports_non_list_v1_views_container(
     # Not additionally reported once per character/key of the container.
     assert not [e for e in hard if "must be a mapping" in e.message]
 
-    # Every consumer degrades to "no views" rather than raising.
+    # Runtime/build consumers still normalise to "no views". Migration is
+    # stricter because deleting the source after normalisation would lose data.
     assert build_manifest(tmp_path)["views"] == []
     assert build_json(tmp_path)["views"] == []
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="malformed views"):
+        plan_upgrade(tmp_path, target_version=2)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+
+
+@pytest.mark.parametrize("views_yml", ["[]\n", "false\n", "0\n", '""\n'])
+def test_plan_upgrade_v1_to_v2_rejects_falsey_non_mapping_views_root(
+    tmp_path, views_yml
+):
+    """Falsey YAML roots are still malformed unless the document is empty."""
+    _make_v1_project(tmp_path)
+    views_file = tmp_path / "views.yml"
+    views_file.write_text(views_yml, encoding="utf-8")
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="malformed views"):
+        plan_upgrade(tmp_path, target_version=2)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+
+
+def test_plan_upgrade_v1_to_v2_allows_empty_views_document(tmp_path):
+    """An empty YAML document has no views to lose and remains upgradeable."""
+    _make_v1_project(tmp_path)
+    (tmp_path / "views.yml").write_text("", encoding="utf-8")
+
     plan = plan_upgrade(tmp_path, target_version=2)
+
+    assert "views.yml" in plan.files_deleted
     assert not [f for f in plan.files_created if f.startswith("views/")]
+
+
+@pytest.mark.parametrize(
+    "views_yml",
+    [
+        "legacy_setting: value\n",
+        "views: []\nlegacy_setting: value\n",
+    ],
+)
+def test_plan_upgrade_v1_to_v2_rejects_unpreserved_views_root_keys(
+    tmp_path, views_yml
+):
+    """Root fields migration cannot carry forward must block source deletion."""
+    _make_v1_project(tmp_path)
+    views_file = tmp_path / "views.yml"
+    views_file.write_text(views_yml, encoding="utf-8")
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="unsupported root keys"):
+        plan_upgrade(tmp_path, target_version=2)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+
+
+def test_plan_upgrade_v1_to_v2_allows_anchor_root_used_by_merge_key(tmp_path):
+    """A root anchor consumed by a view is preserved through YAML expansion."""
+    _make_v1_project(tmp_path)
+    (tmp_path / "views.yml").write_text(
+        "base: &base\n"
+        "  statement: SELECT 1\n"
+        "views:\n"
+        "  - name: anchored\n"
+        "    <<: *base\n",
+        encoding="utf-8",
+    )
+
+    plan = plan_upgrade(tmp_path, target_version=2)
+
+    assert "views/anchored/metadata.yml" in plan.files_created
+    assert "views.yml" in plan.files_deleted
+
+
+def test_apply_upgrade_v1_to_v2_preserves_merge_override(tmp_path):
+    """Explicit keys may override merged defaults without looking duplicated."""
+    _make_v1_project(tmp_path)
+    (tmp_path / "views.yml").write_text(
+        "base: &base\n"
+        "  statement: SELECT 1\n"
+        "views:\n"
+        "  - <<: *base\n"
+        "    name: anchored\n"
+        "    statement: SELECT 2\n",
+        encoding="utf-8",
+    )
+    plan = plan_upgrade(tmp_path, target_version=2)
+
     apply_upgrade(tmp_path, plan)
-    assert not (tmp_path / "views").exists()
+
+    migrated = yaml.safe_load(
+        (tmp_path / "views" / "anchored" / "metadata.yml").read_text(
+            encoding="utf-8"
+        )
+    )
+    assert migrated["name"] == "anchored"
+    assert migrated["statement"] == "SELECT 2"
+    assert not (tmp_path / "views.yml").exists()
+
+
+@pytest.mark.parametrize(
+    ("first_name", "second_name"),
+    [
+        ("revenue", "revenue"),
+        ("Revenue", "revenue"),
+    ],
+)
+def test_plan_upgrade_v1_to_v2_rejects_colliding_view_targets(
+    tmp_path, first_name, second_name
+):
+    """Legacy views must map to distinct portable v2 target directories."""
+    _make_v1_project(tmp_path)
+    views_file = tmp_path / "views.yml"
+    views_file.write_text(
+        "views:\n"
+        f"  - name: {first_name}\n"
+        "    statement: SELECT 1\n"
+        f"  - name: {second_name}\n"
+        "    statement: SELECT 2\n",
+        encoding="utf-8",
+    )
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="multiple legacy views map"):
+        plan_upgrade(tmp_path, target_version=2)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+    _assert_no_v2_entity_directories(tmp_path)
+
+
+def test_apply_upgrade_v1_to_v2_rechecks_view_target_collisions_before_writing(
+    tmp_path,
+):
+    """A collision introduced after planning must abort before any v2 write."""
+    _make_v1_project(tmp_path)
+    plan = plan_upgrade(tmp_path, target_version=2)
+    (tmp_path / "views.yml").write_text(
+        "views:\n"
+        "  - name: Revenue\n"
+        "    statement: SELECT 1\n"
+        "  - name: revenue\n"
+        "    statement: SELECT 2\n",
+        encoding="utf-8",
+    )
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="multiple legacy views map"):
+        apply_upgrade(tmp_path, plan)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+    _assert_no_v2_entity_directories(tmp_path)
+
+
+@pytest.mark.parametrize("second_name", ["orders", "Orders"])
+def test_plan_upgrade_v1_to_v2_rejects_colliding_model_targets(
+    tmp_path, second_name
+):
+    """Different legacy model files cannot collapse into one v2 directory."""
+    _make_v1_project(tmp_path)
+    revenue_file = tmp_path / "models" / "revenue.yml"
+    revenue_file.write_text(
+        revenue_file.read_text(encoding="utf-8").replace(
+            "name: revenue\n", f"name: {second_name}\n", 1
+        ),
+        encoding="utf-8",
+    )
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="multiple legacy model files map"):
+        plan_upgrade(tmp_path, target_version=2)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+    _assert_no_v2_entity_directories(tmp_path)
+
+
+@pytest.mark.parametrize(
+    "views_yml",
+    [
+        "views:\n  - name: first\nviews:\n  - name: second\n",
+        "views:\n  - name: first\n    statement: SELECT 1\n    statement: SELECT 2\n",
+    ],
+)
+def test_plan_upgrade_v1_to_v2_rejects_duplicate_yaml_keys_without_data_loss(
+    tmp_path, views_yml
+):
+    """Duplicate YAML keys must never be collapsed before migration."""
+    _make_v1_project(tmp_path)
+    views_file = tmp_path / "views.yml"
+    views_file.write_text(views_yml, encoding="utf-8")
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="duplicate YAML key"):
+        plan_upgrade(tmp_path, target_version=2)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+
+
+def test_apply_upgrade_v1_to_v2_rechecks_duplicate_yaml_keys_before_writing(tmp_path):
+    """A duplicate introduced after planning must still block all writes."""
+    _make_v1_project(tmp_path)
+    plan = plan_upgrade(tmp_path, target_version=2)
+    views_file = tmp_path / "views.yml"
+    views_file.write_text(
+        "views:\n  - name: first\nviews:\n  - name: second\n",
+        encoding="utf-8",
+    )
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="duplicate YAML key"):
+        apply_upgrade(tmp_path, plan)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+    _assert_no_v2_entity_directories(tmp_path)
+
+
+@pytest.mark.parametrize(
+    "views_yml",
+    [
+        "views: [\n",
+        "views:\n  - ? [name]\n    : value\n",
+    ],
+)
+def test_plan_upgrade_v1_to_v2_wraps_yaml_loader_errors_without_data_loss(
+    tmp_path, views_yml
+):
+    """Malformed YAML must fail as UpgradeError without changing source files."""
+    _make_v1_project(tmp_path)
+    views_file = tmp_path / "views.yml"
+    views_file.write_text(views_yml, encoding="utf-8")
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="invalid views.yml"):
+        plan_upgrade(tmp_path, target_version=2)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+
+
+def test_apply_upgrade_v1_to_v2_rechecks_yaml_loader_errors_before_writing(tmp_path):
+    """Malformed YAML introduced after planning must block all writes."""
+    _make_v1_project(tmp_path)
+    plan = plan_upgrade(tmp_path, target_version=2)
+    views_file = tmp_path / "views.yml"
+    views_file.write_text("views: [\n", encoding="utf-8")
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="invalid views.yml"):
+        apply_upgrade(tmp_path, plan)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+    _assert_no_v2_entity_directories(tmp_path)
+
+
+def test_plan_upgrade_v1_to_v2_rejects_nameless_view_without_data_loss(tmp_path):
+    _make_v1_project(tmp_path)
+    views_file = tmp_path / "views.yml"
+    views_file.write_text(
+        "views:\n"
+        "  - name: kept\n"
+        "    statement: SELECT 1\n"
+        "  - statement: SELECT id FROM orders\n"
+        "  - just_a_bare_string\n",
+        encoding="utf-8",
+    )
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="malformed views"):
+        plan_upgrade(tmp_path, target_version=2)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
+    assert "SELECT id FROM orders" in views_file.read_text(encoding="utf-8")
+    assert "just_a_bare_string" in views_file.read_text(encoding="utf-8")
 
 
 # ── Regression: model columns non-list / non-dict entries ─────────────────
@@ -1845,23 +2145,31 @@ def test_build_json_does_not_crash_on_v1_views_yml_non_mapping_entries(tmp_path)
     assert [v["name"] for v in manifest["views"]] == ["summary"]
 
 
-def test_plan_upgrade_v1_to_v2_does_not_crash_on_non_mapping_view(tmp_path):
+def test_plan_upgrade_v1_to_v2_rejects_non_mapping_view(tmp_path):
     _make_v1_project(tmp_path)
     _corrupt_v1_views_yml(tmp_path)
-    result = plan_upgrade(tmp_path, target_version=2)
-    view_files = [f for f in result.files_created if f.startswith("views/")]
-    assert view_files == ["views/summary/metadata.yml"]
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="malformed views"):
+        plan_upgrade(tmp_path, target_version=2)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
 
 
-def test_apply_upgrade_v1_to_v2_does_not_crash_on_non_mapping_view(tmp_path):
+def test_apply_upgrade_v1_to_v2_rechecks_non_mapping_view_before_writing(tmp_path):
     _make_v1_project(tmp_path)
-    _corrupt_v1_views_yml(tmp_path)
     result = plan_upgrade(tmp_path, target_version=2)
-    apply_upgrade(tmp_path, result)
-    assert (tmp_path / "views" / "summary" / "metadata.yml").exists()
-    assert not (tmp_path / "views.yml").exists()
-    # Only the well-formed view became a directory — no junk siblings.
-    assert [d.name for d in (tmp_path / "views").iterdir()] == ["summary"]
+    _corrupt_v1_views_yml(tmp_path)
+    source_contents = _snapshot_v1_sources(tmp_path)
+
+    from wren.context import UpgradeError as _UE  # noqa: PLC0415
+
+    with pytest.raises(_UE, match="malformed views"):
+        apply_upgrade(tmp_path, result)
+
+    _assert_v1_sources_unchanged(tmp_path, source_contents)
 
 
 @pytest.mark.parametrize("entity", ["model", "view", "cube"])