diff --git a/src/specify_cli/bundler/models/manifest.py b/src/specify_cli/bundler/models/manifest.py
index 39684b2327..4d52757c43 100644
--- a/src/specify_cli/bundler/models/manifest.py
+++ b/src/specify_cli/bundler/models/manifest.py
@@ -192,9 +192,17 @@ def structural_errors(self) -> list[str]:
                 "(lowercase letters, digits, '.', '_', '-'; no path separators)."
             )
 
+        seen_components: set[tuple[str, str]] = set()
         for ref in self.components:
             if not ref.id:
                 errors.append(f"A {ref.kind[:-1]} entry is missing its 'id'.")
+            key = (ref.kind, ref.id)
+            if ref.id and key in seen_components:
+                errors.append(
+                    f"Duplicate {ref.kind[:-1]} '{ref.id}' in "
+                    f"'provides.{ref.kind}'."
+                )
+            seen_components.add(key)
             if ref.kind != "steps" and not ref.version:
                 errors.append(
                     f"{ref.kind[:-1]} '{ref.id or '<unknown>'}' must be pinned to a 'version'."
diff --git a/src/specify_cli/bundler/services/adapters.py b/src/specify_cli/bundler/services/adapters.py
index ca39a2489b..a0cc03970e 100644
--- a/src/specify_cli/bundler/services/adapters.py
+++ b/src/specify_cli/bundler/services/adapters.py
@@ -223,6 +223,12 @@ def is_installed(self, project_root: Path, component: ComponentRef) -> bool:
         manager = self._manager_for(component, project_root)
         return manager.is_installed(component)
 
+    def snapshot(
+        self, project_root: Path, component: ComponentRef
+    ) -> ComponentRef | None:
+        manager = self._manager_for(component, project_root)
+        return manager.snapshot(component)
+
     def install(self, project_root: Path, component: ComponentRef) -> None:
         manager = self._manager_for(component, project_root)
         manager.install(component)
diff --git a/src/specify_cli/bundler/services/installer.py b/src/specify_cli/bundler/services/installer.py
index 58e220638d..6858a678b3 100644
--- a/src/specify_cli/bundler/services/installer.py
+++ b/src/specify_cli/bundler/services/installer.py
@@ -35,6 +35,10 @@ class PrimitiveInstaller(Protocol):
 
     def is_installed(self, project_root: Path, component: ComponentRef) -> bool: ...
 
+    def snapshot(
+        self, project_root: Path, component: ComponentRef
+    ) -> ComponentRef | None: ...
+
     def install(self, project_root: Path, component: ComponentRef) -> None: ...
 
     def remove(self, project_root: Path, component: ComponentRef) -> None: ...
@@ -65,11 +69,9 @@ def install_bundle(
 ) -> InstallResult:
     """Execute *plan*, recording provenance. Idempotent, with bounded rollback.
 
-    Atomicity is scoped, not global: on failure only the components newly
-    installed during *this* call are rolled back, and the provenance record is
-    written solely on full success (a failure records nothing). Components that
-    were already installed beforehand — including those re-applied when *refresh*
-    is True — are never rolled back.
+    Atomicity is scoped, not global: completed component mutations are reversed
+    on failure, and the provenance record is written solely on full success.
+    Rollback is best-effort because primitive restoration can itself fail.
 
     When *refresh* is True (used by ``specify bundle update``), components that
     are already installed are re-applied through the primitive machinery so they
@@ -108,9 +110,8 @@ def install_bundle(
         if r.bundle_id != plan.bundle_id
         for c in r.contributed_components
     }
-
     contributed: list[ComponentRef] = []
-    done: list[ComponentRef] = []
+    rollback_actions: list[tuple[str, ComponentRef]] = []
     try:
         for component in plan.components:
             key = (component.kind, component.id)
@@ -122,7 +123,11 @@ def install_bundle(
                 # does not own (FR-022).
                 owned = key in prior_ours or key in other_tracked
                 if refresh and owned:
+                    prior_component = _snapshot_component(
+                        project_root, installer, component
+                    )
                     _refresh_component(project_root, installer, component)
+                    rollback_actions.append(("refresh", prior_component))
                     result.refreshed.append(component)
                 else:
                     result.skipped.append(component)
@@ -130,7 +135,7 @@ def install_bundle(
                     contributed.append(component)
                 continue
             installer.install(project_root, component)
-            done.append(component)
+            rollback_actions.append(("remove", component))
             result.installed.append(component)
             contributed.append(component)
 
@@ -153,27 +158,41 @@ def install_bundle(
                 if key in still_needed:
                     continue
                 if installer.is_installed(project_root, component):
+                    prior_component = _snapshot_component(
+                        project_root, installer, component
+                    )
                     installer.remove(project_root, component)
+                    rollback_actions.append(("install", prior_component))
                     result.uninstalled.append(component)
-    except BundlerError:
-        _rollback(project_root, installer, done)
-        raise
+
+        record = InstalledBundleRecord.create(
+            bundle_id=plan.bundle_id,
+            version=plan.version,
+            components=contributed,
+            # Preserve the original install time across refresh/update so
+            # ``bundle list`` keeps reporting when the bundle was first installed.
+            installed_at=existing.installed_at if existing is not None else None,
+        )
+        save_records(project_root, upsert_record(records, record))
     except Exception as exc:  # noqa: BLE001
-        _rollback(project_root, installer, done)
-        raise BundlerError(
-            f"Failed to install bundle '{plan.bundle_id}': {exc}. "
-            "No changes were recorded."
-        ) from exc
+        rollback_complete = _rollback(
+            project_root, installer, rollback_actions
+        )
+        if isinstance(exc, BundlerError) and rollback_complete:
+            raise
+        detail = (
+            "Completed changes were rolled back and no provenance record was written."
+            if rollback_complete
+            else "Rollback was incomplete and no provenance record was written; "
+            "the project may be inconsistent."
+        )
+        message = (
+            str(exc)
+            if isinstance(exc, BundlerError)
+            else f"Failed to install bundle '{plan.bundle_id}': {exc}"
+        )
+        raise BundlerError(f"{message}. {detail}") from exc
 
-    record = InstalledBundleRecord.create(
-        bundle_id=plan.bundle_id,
-        version=plan.version,
-        components=contributed,
-        # Preserve the original install time across refresh/update so
-        # ``bundle list`` keeps reporting when the bundle was first installed.
-        installed_at=existing.installed_at if existing is not None else None,
-    )
-    save_records(project_root, upsert_record(records, record))
     return result
 
 
@@ -245,13 +264,54 @@ def _refresh_component(
         installer.install(project_root, component)
 
 
+def _snapshot_component(
+    project_root: Path,
+    installer: PrimitiveInstaller,
+    component: ComponentRef,
+) -> ComponentRef:
+    """Capture actual installed metadata before a destructive update."""
+    try:
+        snapshot = installer.snapshot(project_root, component)
+    except BundlerError:
+        raise
+    except Exception as exc:  # noqa: BLE001
+        raise BundlerError(
+            f"Cannot safely update {component.label()}: failed to snapshot "
+            f"the installed component: {exc}"
+        ) from exc
+
+    if snapshot is None:
+        raise BundlerError(
+            f"Cannot safely update {component.label()}: installed state "
+            "could not be snapshotted."
+        )
+    if (snapshot.kind, snapshot.id) != (component.kind, component.id):
+        raise BundlerError(
+            f"Cannot safely update {component.label()}: snapshot returned "
+            f"the wrong component ({snapshot.label()})."
+        )
+    if not isinstance(snapshot.version, str) or not snapshot.version.strip():
+        raise BundlerError(
+            f"Cannot safely update {component.label()}: installed version "
+            "could not be determined for rollback."
+        )
+    return snapshot
+
+
 def _rollback(
     project_root: Path,
     installer: PrimitiveInstaller,
-    done: list[ComponentRef],
-) -> None:
-    for component in reversed(done):
+    actions: list[tuple[str, ComponentRef]],
+) -> bool:
+    complete = True
+    for operation, component in reversed(actions):
         try:
-            installer.remove(project_root, component)
+            if operation == "remove":
+                installer.remove(project_root, component)
+            elif operation == "install":
+                installer.install(project_root, component)
+            else:
+                _refresh_component(project_root, installer, component)
         except Exception:  # noqa: BLE001 - best-effort rollback
-            continue
+            complete = False
+    return complete
diff --git a/src/specify_cli/bundler/services/primitives.py b/src/specify_cli/bundler/services/primitives.py
index 68a9c47632..3ab961ca18 100644
--- a/src/specify_cli/bundler/services/primitives.py
+++ b/src/specify_cli/bundler/services/primitives.py
@@ -88,6 +88,9 @@ class _KindManager(Protocol):
     def is_installed(self, component: ComponentRef) -> bool:
         pass
 
+    def snapshot(self, component: ComponentRef) -> ComponentRef | None:
+        pass
+
     def install(self, component: ComponentRef) -> None:
         pass
 
@@ -142,6 +145,29 @@ def _delegate_command(action: str, label: str, call) -> None:
             raise BundlerError(f"Failed to {action} {label}.") from exc
 
 
+def _snapshot_ref(
+    component: ComponentRef,
+    *,
+    version: object,
+    metadata: dict[str, object] | None = None,
+) -> ComponentRef:
+    metadata = metadata or {}
+    actual_version = version.strip() if isinstance(version, str) else None
+    source = metadata.get("source")
+    priority = metadata.get("priority")
+    return ComponentRef(
+        kind=component.kind,
+        id=component.id,
+        version=actual_version or None,
+        source=source if isinstance(source, str) else None,
+        priority=(
+            priority
+            if isinstance(priority, int) and not isinstance(priority, bool)
+            else None
+        ),
+    )
+
+
 class _PresetKindManager:
     def __init__(self, project_root: Path, allow_network: bool) -> None:
         from ...presets import PresetManager
@@ -156,6 +182,21 @@ def is_installed(self, component: ComponentRef) -> bool:
         except Exception:  # noqa: BLE001
             return False
 
+    def snapshot(self, component: ComponentRef) -> ComponentRef | None:
+        metadata = self._manager.registry.get(component.id)
+        manifest = self._manager.get_pack(component.id)
+        if metadata is None and manifest is None:
+            return None
+        return _snapshot_ref(
+            component,
+            version=(
+                metadata.get("version")
+                if metadata is not None
+                else manifest.version
+            ),
+            metadata=metadata,
+        )
+
     def install(self, component: ComponentRef) -> None:
         self._do_install(component, force=False)
 
@@ -239,6 +280,16 @@ def is_installed(self, component: ComponentRef) -> bool:
         except Exception:  # noqa: BLE001
             return False
 
+    def snapshot(self, component: ComponentRef) -> ComponentRef | None:
+        metadata = self._manager.registry.get(component.id)
+        if metadata is None:
+            return None
+        return _snapshot_ref(
+            component,
+            version=metadata.get("version"),
+            metadata=metadata,
+        )
+
     def install(self, component: ComponentRef) -> None:
         self._do_install(component, force=False)
 
@@ -326,6 +377,14 @@ def is_installed(self, component: ComponentRef) -> bool:
         except Exception:  # noqa: BLE001
             return False
 
+    def snapshot(self, component: ComponentRef) -> ComponentRef | None:
+        metadata = self._registry.get(component.id)
+        if metadata is None:
+            return None
+        return _snapshot_ref(
+            component, version=metadata.get("version"), metadata=metadata
+        )
+
     def install(self, component: ComponentRef) -> None:
         if not self._allow_network and not self._is_bundled(component.id):
             raise BundlerError(
@@ -392,6 +451,14 @@ def is_installed(self, component: ComponentRef) -> bool:
         except Exception:  # noqa: BLE001
             return False
 
+    def snapshot(self, component: ComponentRef) -> ComponentRef | None:
+        metadata = self._registry.get(component.id)
+        if metadata is None:
+            return None
+        return _snapshot_ref(
+            component, version=metadata.get("version"), metadata=metadata
+        )
+
     def install(self, component: ComponentRef) -> None:
         if not self._allow_network:
             raise BundlerError(
@@ -399,6 +466,7 @@ def install(self, component: ComponentRef) -> None:
                 f"is disabled; re-run without --offline or install it first with "
                 f"'specify workflow step add {component.id}'."
             )
+        self._assert_pinned_version(component)
         from ... import workflow_step_add
 
         with _chdir(self._root):
@@ -458,6 +526,32 @@ def refresh(self, component: ComponentRef) -> None:
         finally:
             shutil.rmtree(backup_dir.parent, ignore_errors=True)
 
+    def _assert_pinned_version(self, component: ComponentRef) -> None:
+        if not component.version:
+            return
+        from ...workflows.catalog import StepCatalog, StepCatalogError
+
+        try:
+            info = StepCatalog(self._root).get_step_info(component.id)
+        except StepCatalogError as exc:
+            raise BundlerError(
+                f"Cannot verify pinned version for step '{component.id}': {exc}"
+            ) from exc
+        if not info:
+            raise BundlerError(
+                f"Cannot verify pinned version for step '{component.id}': "
+                "the step was not found in the catalog."
+            )
+        advertised = info.get("version")
+        if advertised is None or not str(advertised).strip():
+            raise BundlerError(
+                f"Cannot verify pinned version for step '{component.id}': "
+                "the catalog does not advertise a version."
+            )
+        _assert_pinned_version(
+            "Step", component.id, component.version, advertised
+        )
+
     def remove(self, component: ComponentRef) -> None:
         from ... import workflow_step_remove
 
diff --git a/src/specify_cli/bundler/services/references.py b/src/specify_cli/bundler/services/references.py
index b5419237d5..71a28daefe 100644
--- a/src/specify_cli/bundler/services/references.py
+++ b/src/specify_cli/bundler/services/references.py
@@ -48,11 +48,10 @@ def _resolved_locally(root: Path, component: ComponentRef) -> bool:
             # ``_locate_bundled_step`` to mirror the three lookups above.
             # ``BUILTIN_STEP_TYPES`` is the bundled-with-Spec-Kit check for this
             # kind. Deliberately NOT ``STEP_REGISTRY``: ``load_custom_steps``
-            # adds project-installed ids to that process-global mapping and
-            # never removes them, so in a long-lived process a community step
-            # loaded for one project would be accepted as "bundled" when
-            # validating another. Without any bundled check at all, every
-            # built-in step type looked unresolved.
+            # adds the most recently scanned project's ids to that process-global
+            # mapping, so a community step could be accepted as "bundled" when
+            # validating a different root. Without any bundled check at all,
+            # every built-in step type looked unresolved.
             if component.id in BUILTIN_STEP_TYPES:
                 return True
             return StepRegistry(root).is_installed(component.id)
diff --git a/src/specify_cli/workflows/__init__.py b/src/specify_cli/workflows/__init__.py
index 1e608ca168..fd7baf685b 100644
--- a/src/specify_cli/workflows/__init__.py
+++ b/src/specify_cli/workflows/__init__.py
@@ -74,11 +74,8 @@ def _register_builtin_steps() -> None:
 _register_builtin_steps()
 
 # The step types Spec Kit ships, snapshotted before any community step can be
-# loaded. ``load_custom_steps`` adds project-installed ids to the process-global
-# ``STEP_REGISTRY`` and never removes them, so ``STEP_REGISTRY`` cannot answer
-# "is this bundled with Spec Kit?" in a long-lived process: a step loaded for one
-# project would look built-in for the next. Callers that need the immutable set
-# (e.g. the bundler's reference checker) must use this instead.
+# loaded. Callers that need the immutable set (e.g. the bundler's reference
+# checker) must use this instead of the project-scoped entries in STEP_REGISTRY.
 BUILTIN_STEP_TYPES: frozenset[str] = frozenset(STEP_REGISTRY)
 
 
@@ -93,12 +90,24 @@ def load_custom_steps(project_root: Path) -> list[str]:
     Silently skips packages that fail to import or validate.
     """
     import hashlib as _hashlib
+    import importlib as _importlib
     import importlib.util as _importlib_util
     import re as _re
+    import shutil as _shutil
     import sys as _sys
 
     steps_dir = Path(project_root) / ".specify" / "workflows" / "steps"
 
+    # Custom steps are project-scoped even though the registry and Python module
+    # cache are process-global. Clear the previous project's classes and package
+    # modules before every scan so removed or updated code cannot remain active.
+    for _type_key in tuple(STEP_REGISTRY):
+        if _type_key not in BUILTIN_STEP_TYPES:
+            STEP_REGISTRY.pop(_type_key, None)
+    _module_prefix = "_speckit_custom_step_"
+    for _mod_key in [k for k in _sys.modules if k.startswith(_module_prefix)]:
+        _sys.modules.pop(_mod_key, None)
+
     # Defense-in-depth: refuse to execute step code from a symlinked
     # parent directory under .specify/workflows/steps, which could redirect
     # the import outside the project root and bypass the install-time
@@ -151,6 +160,15 @@ def load_custom_steps(project_root: Path) -> list[str]:
             key_hash = _hashlib.sha256(type_key.encode()).hexdigest()[:8]
             module_name = f"_speckit_custom_step_{safe_key}_{key_hash}"
 
+            # Removing sys.modules entries alone is insufficient for same-path
+            # reloads: Python may reuse a same-size, same-mtime .pyc file.
+            # Custom packages are small and source-controlled by the project,
+            # so discard only their generated bytecode before importing.
+            for cache_dir in step_dir.rglob("__pycache__"):
+                if cache_dir.is_dir() and not cache_dir.is_symlink():
+                    _shutil.rmtree(cache_dir, ignore_errors=True)
+            _importlib.invalidate_caches()
+
             # Treat the step directory as a proper package so that relative
             # imports inside the step (e.g. ``from .helpers import …``) work.
             spec = _importlib_util.spec_from_file_location(
diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py
index f275bfe09a..f334dfa3f2 100644
--- a/src/specify_cli/workflows/_commands.py
+++ b/src/specify_cli/workflows/_commands.py
@@ -3408,6 +3408,33 @@ def _safe_fetch(url: str) -> bytes:
             )
             raise typer.Exit(1)
 
+        catalog_version = info.get("version")
+        downloaded_version = step_meta.get("version")
+        if "version" in info:
+            from packaging import version as pkg_version
+
+            versions_match = False
+            if (
+                isinstance(downloaded_version, str)
+                and downloaded_version.strip()
+                and isinstance(catalog_version, str)
+                and catalog_version.strip()
+            ):
+                try:
+                    versions_match = pkg_version.Version(
+                        downloaded_version
+                    ) == pkg_version.Version(catalog_version)
+                except pkg_version.InvalidVersion:
+                    versions_match = downloaded_version == catalog_version
+            if not versions_match:
+                console.print(
+                    f"[red]Error:[/red] step.yml version "
+                    f"({_escape_markup(repr(downloaded_version))}) does not match "
+                    f"the catalog version ({_escape_markup(repr(catalog_version))}). "
+                    "The catalog entry may be stale or misconfigured."
+                )
+                raise typer.Exit(1)
+
         # Write the two required files.
         try:
             (tmp_path / "step.yml").write_bytes(step_yml_content)
diff --git a/tests/bundler_helpers.py b/tests/bundler_helpers.py
index 0ebaf2f1c7..77e8f82f6a 100644
--- a/tests/bundler_helpers.py
+++ b/tests/bundler_helpers.py
@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import json
+from dataclasses import replace
 from pathlib import Path
 
 import yaml
@@ -97,6 +98,7 @@ class FakeInstaller:
 
     def __init__(self, *, fail_on: str | None = None) -> None:
         self.installed: set[tuple[str, str]] = set()
+        self.components: dict[tuple[str, str], ComponentRef] = {}
         self.install_calls: list[tuple[str, str]] = []
         self.remove_calls: list[tuple[str, str]] = []
         self.refresh_calls: list[tuple[str, str]] = []
@@ -115,11 +117,23 @@ def install(self, project_root: Path, component: ComponentRef) -> None:
         if self._fail_on is not None and component.id == self._fail_on:
             raise BundlerError(f"Simulated failure installing {component.id}")
         self.installed.add(self._key(component))
+        self.components[self._key(component)] = replace(
+            component, version=component.version or "test-installed"
+        )
 
     def remove(self, project_root: Path, component: ComponentRef) -> None:
         self.remove_calls.append(self._key(component))
         self.installed.discard(self._key(component))
+        self.components.pop(self._key(component), None)
 
     def refresh(self, project_root: Path, component: ComponentRef) -> None:
         self.refresh_calls.append(self._key(component))
         self.installed.add(self._key(component))
+        self.components[self._key(component)] = replace(
+            component, version=component.version or "test-installed"
+        )
+
+    def snapshot(
+        self, project_root: Path, component: ComponentRef
+    ) -> ComponentRef | None:
+        return self.components.get(self._key(component))
diff --git a/tests/contract/test_manifest_schema.py b/tests/contract/test_manifest_schema.py
index 4784bdf462..6006540495 100644
--- a/tests/contract/test_manifest_schema.py
+++ b/tests/contract/test_manifest_schema.py
@@ -127,6 +127,29 @@ def test_components_property_orders_by_kind():
     assert kinds == ["extensions", "presets", "steps", "workflows"]
 
 
+def test_duplicate_component_in_same_kind_is_rejected():
+    data = valid_manifest_dict()
+    data["provides"]["extensions"].append(
+        {"id": "ext-a", "version": "9.9.9"}
+    )
+
+    errors = BundleManifest.from_dict(data).structural_errors()
+
+    assert any(
+        "duplicate extension 'ext-a'" in error.lower()
+        for error in errors
+    )
+
+
+def test_same_component_id_in_different_kinds_is_allowed():
+    data = valid_manifest_dict()
+    data["provides"]["steps"].append({"id": "ext-a"})
+
+    errors = BundleManifest.from_dict(data).structural_errors()
+
+    assert not any("duplicate" in error.lower() for error in errors)
+
+
 def test_string_tags_rejected_not_split_per_character():
     # A bare string would otherwise be iterated character-by-character; the
     # schema requires a list of strings.
diff --git a/tests/integration/test_bundler_install_flow.py b/tests/integration/test_bundler_install_flow.py
index 0966008a74..cc920d2f4b 100644
--- a/tests/integration/test_bundler_install_flow.py
+++ b/tests/integration/test_bundler_install_flow.py
@@ -64,6 +64,25 @@ def test_partial_failure_rolls_back_and_records_nothing(tmp_path: Path):
     assert load_records(tmp_path) == []
 
 
+def test_record_save_failure_rolls_back_new_components(tmp_path: Path, monkeypatch):
+    make_project(tmp_path)
+    manifest = BundleManifest.from_dict(valid_manifest_dict())
+    installer = FakeInstaller()
+
+    def fail_save(*_args, **_kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr(
+        "specify_cli.bundler.services.installer.save_records", fail_save
+    )
+
+    with pytest.raises(BundlerError, match="disk full"):
+        install_bundle(tmp_path, _plan(manifest), installer, manifest=manifest)
+
+    assert installer.installed == set()
+    assert load_records(tmp_path) == []
+
+
 def test_remove_is_non_collateral(tmp_path: Path):
     make_project(tmp_path)
     installer = FakeInstaller()
@@ -493,6 +512,140 @@ def test_update_keeps_component_still_needed_by_sibling_bundle(tmp_path: Path):
     }
 
 
+def test_update_record_save_failure_restores_refreshed_and_dropped_components(
+    tmp_path: Path, monkeypatch
+):
+    make_project(tmp_path)
+
+    class VersionedInstaller(FakeInstaller):
+        def __init__(self):
+            super().__init__()
+            self.versions: dict[tuple[str, str], str | None] = {}
+
+        def install(self, project_root, component):
+            super().install(project_root, component)
+            self.versions[self._key(component)] = component.version
+
+        def refresh(self, project_root, component):
+            super().refresh(project_root, component)
+            self.versions[self._key(component)] = component.version
+
+        def remove(self, project_root, component):
+            super().remove(project_root, component)
+            self.versions.pop(self._key(component), None)
+
+    installer = VersionedInstaller()
+    man_v1 = _bundle("demo", ["ext-a", "ext-b"])
+    install_bundle(tmp_path, _plan(man_v1), installer, manifest=man_v1)
+    original_record = records_path(tmp_path).read_bytes()
+
+    man_v2 = _bundle("demo", ["ext-a"], version="2.0.0")
+
+    def fail_save(*_args, **_kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr(
+        "specify_cli.bundler.services.installer.save_records", fail_save
+    )
+
+    with pytest.raises(BundlerError, match="disk full"):
+        install_bundle(
+            tmp_path,
+            _plan(man_v2),
+            installer,
+            manifest=man_v2,
+            refresh=True,
+        )
+
+    assert installer.installed == {
+        ("extensions", "ext-a"),
+        ("extensions", "ext-b"),
+    }
+    assert installer.versions == {
+        ("extensions", "ext-a"): "1.0.0",
+        ("extensions", "ext-b"): "1.0.0",
+    }
+    assert records_path(tmp_path).read_bytes() == original_record
+
+
+def test_update_rollback_uses_installed_snapshot_not_shared_bundle_pin(
+    tmp_path: Path, monkeypatch
+):
+    make_project(tmp_path)
+    installer = FakeInstaller()
+
+    man_a = _bundle("a", ["ext-a"], version="1.0.0")
+    install_bundle(tmp_path, _plan(man_a), installer, manifest=man_a)
+
+    man_b_v2 = _bundle("b", ["ext-a", "ext-b"], version="2.0.0")
+    install_bundle(tmp_path, _plan(man_b_v2), installer, manifest=man_b_v2)
+    original_record = records_path(tmp_path).read_bytes()
+
+    assert installer.components[("extensions", "ext-a")].version == "1.0.0"
+    assert next(
+        record for record in load_records(tmp_path) if record.bundle_id == "b"
+    ).contributed_components[0].version == "2.0.0"
+
+    man_b_v3 = _bundle("b", ["ext-a"], version="3.0.0")
+
+    def fail_save(*_args, **_kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr(
+        "specify_cli.bundler.services.installer.save_records", fail_save
+    )
+
+    with pytest.raises(BundlerError, match="disk full"):
+        install_bundle(
+            tmp_path,
+            _plan(man_b_v3),
+            installer,
+            manifest=man_b_v3,
+            refresh=True,
+        )
+
+    assert installer.components[("extensions", "ext-a")].version == "1.0.0"
+    assert installer.components[("extensions", "ext-b")].version == "2.0.0"
+    assert records_path(tmp_path).read_bytes() == original_record
+
+
+def test_bundler_error_reports_incomplete_rollback(tmp_path: Path, monkeypatch):
+    make_project(tmp_path)
+
+    class FailingRollbackInstaller(FakeInstaller):
+        def refresh(self, project_root, component):
+            if component.version == "1.0.0":
+                raise BundlerError("old artifact unavailable")
+            super().refresh(project_root, component)
+
+    installer = FailingRollbackInstaller()
+    man_v1 = _bundle("demo", ["ext-a"], version="1.0.0")
+    install_bundle(tmp_path, _plan(man_v1), installer, manifest=man_v1)
+
+    def fail_save(*_args, **_kwargs):
+        raise BundlerError("record write failed")
+
+    monkeypatch.setattr(
+        "specify_cli.bundler.services.installer.save_records", fail_save
+    )
+
+    man_v2 = _bundle("demo", ["ext-a"], version="2.0.0")
+    with pytest.raises(
+        BundlerError,
+        match=(
+            "record write failed.*Rollback was incomplete.*"
+            "project may be inconsistent"
+        ),
+    ):
+        install_bundle(
+            tmp_path,
+            _plan(man_v2),
+            installer,
+            manifest=man_v2,
+            refresh=True,
+        )
+
+
 def test_install_result_changed_reports_uninstalled():
     # A `bundle update` that only DROPS components (new manifest reduces
     # provides) populates uninstalled with nothing installed/refreshed; that is
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index 2c7141e954..0cb912d385 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -9698,6 +9698,113 @@ def test_get_step_info_returns_entry_or_none(self, project_dir, monkeypatch):
 class TestLoadCustomSteps:
     """Test dynamic loading of custom step types from the filesystem."""
 
+    def test_loading_another_project_replaces_custom_step_modules(self, tmp_path):
+        import hashlib
+        import sys
+
+        from specify_cli.workflows import STEP_REGISTRY, load_custom_steps
+
+        type_key = "project-scoped-step"
+        key_hash = hashlib.sha256(type_key.encode()).hexdigest()[:8]
+        module_name = f"_speckit_custom_step_project_scoped_step_{key_hash}"
+
+        def write_step(project_root, marker):
+            step_dir = (
+                project_root
+                / ".specify"
+                / "workflows"
+                / "steps"
+                / type_key
+            )
+            step_dir.mkdir(parents=True)
+            (step_dir / "step.yml").write_text(
+                f"step:\n  type_key: {type_key}\n", encoding="utf-8"
+            )
+            (step_dir / "helper.py").write_text(
+                f"MARKER = {marker!r}\n", encoding="utf-8"
+            )
+            (step_dir / "__init__.py").write_text(
+                f"""
+from specify_cli.workflows.base import StepBase, StepResult
+from .helper import MARKER
+
+class ProjectScopedStep(StepBase):
+    type_key = {type_key!r}
+    marker = MARKER
+
+    def execute(self, config, context):
+        return StepResult()
+""",
+                encoding="utf-8",
+            )
+
+        project_a = tmp_path / "project-a"
+        project_b = tmp_path / "project-b"
+        write_step(project_a, "project-a")
+        write_step(project_b, "project-b")
+
+        try:
+            assert load_custom_steps(project_a) == [type_key]
+            assert STEP_REGISTRY[type_key].marker == "project-a"
+
+            assert load_custom_steps(project_b) == [type_key]
+            assert STEP_REGISTRY[type_key].marker == "project-b"
+        finally:
+            STEP_REGISTRY.pop(type_key, None)
+            sys.modules.pop(module_name, None)
+            sys.modules.pop(f"{module_name}.helper", None)
+
+    def test_reloading_same_project_ignores_stale_bytecode(self, tmp_path):
+        import hashlib
+        import os
+        import sys
+
+        from specify_cli.workflows import STEP_REGISTRY, load_custom_steps
+
+        type_key = "reload-step"
+        key_hash = hashlib.sha256(type_key.encode()).hexdigest()[:8]
+        module_name = f"_speckit_custom_step_reload_step_{key_hash}"
+        step_dir = (
+            tmp_path / ".specify" / "workflows" / "steps" / type_key
+        )
+        step_dir.mkdir(parents=True)
+        (step_dir / "step.yml").write_text(
+            f"step:\n  type_key: {type_key}\n", encoding="utf-8"
+        )
+        helper = step_dir / "helper.py"
+        helper.write_text("MARKER = 'version-a'\n", encoding="utf-8")
+        (step_dir / "__init__.py").write_text(
+            f"""
+from specify_cli.workflows.base import StepBase, StepResult
+from .helper import MARKER
+
+class ReloadStep(StepBase):
+    type_key = {type_key!r}
+    marker = MARKER
+
+    def execute(self, config, context):
+        return StepResult()
+""",
+            encoding="utf-8",
+        )
+
+        try:
+            assert load_custom_steps(tmp_path) == [type_key]
+            assert STEP_REGISTRY[type_key].marker == "version-a"
+            original_stat = helper.stat()
+            helper.write_text("MARKER = 'version-b'\n", encoding="utf-8")
+            os.utime(
+                helper,
+                ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns),
+            )
+
+            assert load_custom_steps(tmp_path) == [type_key]
+            assert STEP_REGISTRY[type_key].marker == "version-b"
+        finally:
+            STEP_REGISTRY.pop(type_key, None)
+            sys.modules.pop(module_name, None)
+            sys.modules.pop(f"{module_name}.helper", None)
+
     def test_empty_steps_dir(self, project_dir):
         from specify_cli.workflows import load_custom_steps
 
@@ -10771,6 +10878,142 @@ def test_list_escapes_installed_metadata(
 
 
 class TestWorkflowStepAddCLI:
+    @staticmethod
+    def _invoke_step_add(
+        project_dir,
+        monkeypatch,
+        *,
+        catalog_version,
+        downloaded_version,
+        include_downloaded_version=True,
+    ):
+        from typer.testing import CliRunner
+
+        from specify_cli import app
+        from specify_cli.authentication import http as auth_http
+        from specify_cli.workflows.catalog import StepCatalog
+
+        monkeypatch.chdir(project_dir)
+        monkeypatch.setattr(
+            StepCatalog,
+            "get_step_info",
+            lambda self, step_id: {
+                "id": step_id,
+                "name": "Test Step",
+                "version": catalog_version,
+                "url": "https://example.com/step.yml",
+                "init_url": "https://example.com/__init__.py",
+                "_install_allowed": True,
+            },
+        )
+        step_metadata = {"type_key": "my-step"}
+        if include_downloaded_version:
+            step_metadata["version"] = downloaded_version
+        bodies = {
+            "https://example.com/step.yml": yaml.safe_dump(
+                {"step": step_metadata}
+            ).encode(),
+            "https://example.com/__init__.py": b"# custom step\n",
+        }
+
+        class _FakeResponse:
+            def __init__(self, url):
+                self.url = url
+                self.body = bodies[url]
+                self.offset = 0
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+            def getheader(self, name):
+                return None
+
+            def geturl(self):
+                return self.url
+
+            def read(self, size=-1):
+                if size < 0:
+                    size = len(self.body) - self.offset
+                chunk = self.body[self.offset : self.offset + size]
+                self.offset += len(chunk)
+                return chunk
+
+        monkeypatch.setattr(
+            auth_http,
+            "open_url",
+            lambda url, timeout=30, redirect_validator=None: _FakeResponse(url),
+        )
+
+        return CliRunner().invoke(
+            app, ["workflow", "step", "add", "my-step"]
+        )
+
+    @pytest.mark.parametrize(
+        ("catalog_version", "downloaded_version", "include_downloaded_version"),
+        [
+            ("1.0.0", "2.0.0", True),
+            ("1.0.0", 0, True),
+            ("1.0.0", False, True),
+            ("1.0.0", "", True),
+            ("1.0.0", None, True),
+            ("release-a", " release-a ", True),
+            ("1.0.0", None, False),
+            ("None", None, False),
+        ],
+    )
+    def test_add_rejects_step_yml_version_mismatch(
+        self,
+        project_dir,
+        monkeypatch,
+        catalog_version,
+        downloaded_version,
+        include_downloaded_version,
+    ):
+        from specify_cli.workflows.catalog import StepRegistry
+
+        result = self._invoke_step_add(
+            project_dir,
+            monkeypatch,
+            catalog_version=catalog_version,
+            downloaded_version=downloaded_version,
+            include_downloaded_version=include_downloaded_version,
+        )
+
+        assert result.exit_code != 0
+        assert "does not match the catalog version" in result.output
+        assert not StepRegistry(project_dir).is_installed("my-step")
+        assert not (
+            project_dir / ".specify" / "workflows" / "steps" / "my-step"
+        ).exists()
+
+    @pytest.mark.parametrize(
+        ("catalog_version", "downloaded_version"),
+        [
+            ("1.0.0", "1.0.0"),
+            ("1.0.0", "v1.0.0"),
+        ],
+    )
+    def test_add_accepts_matching_step_yml_version(
+        self, project_dir, monkeypatch, catalog_version, downloaded_version
+    ):
+        from specify_cli.workflows.catalog import StepRegistry
+
+        result = self._invoke_step_add(
+            project_dir,
+            monkeypatch,
+            catalog_version=catalog_version,
+            downloaded_version=downloaded_version,
+        )
+
+        assert result.exit_code == 0, result.output
+        assert StepRegistry(project_dir).is_installed("my-step")
+        assert (
+            project_dir / ".specify" / "workflows" / "steps" / "my-step"
+        ).is_dir()
+
     @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
     def test_add_rejects_symlinked_steps_base_dir(self, project_dir, monkeypatch):
         from typer.testing import CliRunner
diff --git a/tests/unit/test_bundler_primitives.py b/tests/unit/test_bundler_primitives.py
index a0fcb3c635..70ab92b3dc 100644
--- a/tests/unit/test_bundler_primitives.py
+++ b/tests/unit/test_bundler_primitives.py
@@ -120,6 +120,103 @@ def test_workflow_version_mismatch_refuses(tmp_path: Path, monkeypatch):
         manager.install(component)
 
 
+def test_step_version_mismatch_refuses(tmp_path: Path, monkeypatch):
+    import specify_cli
+    from specify_cli.workflows.catalog import StepCatalog
+
+    monkeypatch.setattr(
+        StepCatalog, "get_step_info", lambda self, sid: {"version": "9.9.9"}
+    )
+    calls: list[str] = []
+    monkeypatch.setattr(
+        specify_cli, "workflow_step_add", lambda sid: calls.append(sid)
+    )
+
+    manager = primitive_manager("steps", tmp_path, allow_network=True)
+    component = ComponentRef(kind="steps", id="step-a", version="0.3.0")
+
+    with pytest.raises(BundlerError, match="pinned to version 0.3.0"):
+        manager.install(component)
+    assert calls == []
+
+
+@pytest.mark.parametrize("catalog_version", ["0.3.0", "v0.3.0"])
+def test_step_version_match_installs(
+    tmp_path: Path, monkeypatch, catalog_version
+):
+    import specify_cli
+    from specify_cli.workflows.catalog import StepCatalog
+
+    monkeypatch.setattr(
+        StepCatalog,
+        "get_step_info",
+        lambda self, sid: {"version": catalog_version},
+    )
+    calls: list[str] = []
+    monkeypatch.setattr(
+        specify_cli, "workflow_step_add", lambda sid: calls.append(sid)
+    )
+
+    manager = primitive_manager("steps", tmp_path, allow_network=True)
+    manager.install(
+        ComponentRef(kind="steps", id="step-a", version="0.3.0")
+    )
+
+    assert calls == ["step-a"]
+
+
+@pytest.mark.parametrize(
+    "catalog_info",
+    [
+        None,
+        {},
+        {"version": None},
+        {"version": ""},
+    ],
+)
+def test_step_pin_requires_catalog_version(
+    tmp_path: Path, monkeypatch, catalog_info
+):
+    import specify_cli
+    from specify_cli.workflows.catalog import StepCatalog
+
+    monkeypatch.setattr(
+        StepCatalog, "get_step_info", lambda self, sid: catalog_info
+    )
+    calls: list[str] = []
+    monkeypatch.setattr(
+        specify_cli, "workflow_step_add", lambda sid: calls.append(sid)
+    )
+
+    manager = primitive_manager("steps", tmp_path, allow_network=True)
+    component = ComponentRef(kind="steps", id="step-a", version="0.3.0")
+
+    with pytest.raises(BundlerError, match="Cannot verify pinned version"):
+        manager.install(component)
+    assert calls == []
+
+
+def test_step_pin_refuses_catalog_lookup_failure(tmp_path: Path, monkeypatch):
+    import specify_cli
+    from specify_cli.workflows.catalog import StepCatalog, StepCatalogError
+
+    def fail_lookup(_self, _step_id):
+        raise StepCatalogError("catalog unavailable")
+
+    monkeypatch.setattr(StepCatalog, "get_step_info", fail_lookup)
+    calls: list[str] = []
+    monkeypatch.setattr(
+        specify_cli, "workflow_step_add", lambda sid: calls.append(sid)
+    )
+
+    manager = primitive_manager("steps", tmp_path, allow_network=True)
+    component = ComponentRef(kind="steps", id="step-a", version="0.3.0")
+
+    with pytest.raises(BundlerError, match="catalog unavailable"):
+        manager.install(component)
+    assert calls == []
+
+
 def test_preset_install_preserves_explicit_zero_priority(tmp_path: Path, monkeypatch):
     import specify_cli._assets as assets
 
@@ -373,6 +470,29 @@ def _fake_install(self, *a, **k):
     assert force_values == [True], "DefaultPrimitiveInstaller.refresh() must use force=True"
 
 
+def test_default_installer_snapshots_installed_step(tmp_path: Path):
+    from specify_cli.workflows.catalog import StepRegistry
+
+    registry = StepRegistry(tmp_path)
+    registry.add(
+        "my-step",
+        {
+            "name": "My Step",
+            "version": "1.2.3",
+            "type_key": "my-step",
+        },
+    )
+
+    installer = DefaultPrimitiveInstaller(allow_network=False)
+    snapshot = installer.snapshot(
+        tmp_path, _component("steps", "my-step")
+    )
+
+    assert snapshot == ComponentRef(
+        kind="steps", id="my-step", version="1.2.3"
+    )
+
+
 def test_refresh_succeeds_and_passes_force_true(tmp_path: Path, monkeypatch):
     """Regression: bundle update (refresh=True) of an already-installed extension
     must succeed and pass force=True to install_from_directory."""
diff --git a/tests/unit/test_bundler_references.py b/tests/unit/test_bundler_references.py
index b910a93e99..262b347478 100644
--- a/tests/unit/test_bundler_references.py
+++ b/tests/unit/test_bundler_references.py
@@ -49,11 +49,11 @@ def test_builtin_step_type_resolves(tmp_path: Path):
 def test_community_step_is_not_treated_as_bundled(tmp_path: Path):
     """A community step loaded for one project must not resolve for another.
 
-    `load_custom_steps` adds project-installed ids to the process-global
-    `STEP_REGISTRY` and never removes them, so checking `STEP_REGISTRY` here
-    would accept project A's community step as "bundled" while validating
-    project B. `BUILTIN_STEP_TYPES` is snapshotted before any custom step can
-    load, which is why the check uses it instead.
+    `load_custom_steps` adds the most recently scanned project's ids to the
+    process-global `STEP_REGISTRY`, so checking `STEP_REGISTRY` here could
+    accept another project's community step as "bundled".
+    `BUILTIN_STEP_TYPES` is snapshotted before any custom step can load, which
+    is why the check uses it instead.
     """
     from specify_cli.workflows import (
         BUILTIN_STEP_TYPES,