fix: make StepRegistry atomic and raise on corruption

src/specify_cli/workflows/step/_helpers.py

@@ -47,3 +47,18 @@ def _resolve_steps_base_dir_or_exit(project_root: cli.Path) -> cli.Path:
     except StepInstallError as exc:
         cli.console.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}")
         raise cli.typer.Exit(1) from exc
+
+
+def _load_step_registry_or_exit(project_root: cli.Path):
+    """Build the step registry, turning an OSError into a clean CLI error.
+
+    The registry fails loudly when the file is symlinked or corrupted, so
+    surface that as a printed error rather than an unhandled traceback.
+    """
+    from .catalog import StepRegistry
+
+    try:
+        return StepRegistry(project_root)
+    except OSError as exc:
+        cli.console.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}")
+        raise cli.typer.Exit(1) from exc

src/specify_cli/workflows/step/catalog/_domain.py

@@ -100,29 +100,45 @@ def _has_symlinked_parent(self) -> bool:
         return False
 
     def _load(self) -> dict[str, Any]:
-        """Load registry from disk or create default."""
+        """Load registry from disk or create default.
+
+        Raises ``OSError`` when the registry exists but cannot be read or is
+        not a valid registry document, instead of silently handing back an
+        empty registry and hiding the corruption from the caller.
+        """
         default_registry: dict[str, Any] = {"schema_version": self.SCHEMA_VERSION, "steps": {}}
         # Defense-in-depth: refuse to read the registry if any parent directory
         # under .specify/workflows/steps is a symlink, which could redirect the
-        # read outside the project root.
-        if self._has_symlinked_parent():
-            return default_registry
-        # Defense-in-depth: also refuse to read a symlinked registry file,
-        # which could redirect the read outside the project root.
-        if self.registry_path.is_symlink():
-            return default_registry
+        # read outside the project root, or if the registry file itself is one.
+        if self._has_symlinked_parent() or self.registry_path.is_symlink():
+            raise OSError(
+                f"Refusing to read step registry at {self.registry_path}: "
+                "a parent directory or the registry file itself is a symlink"
+            )
         if self.registry_path.exists():
             try:
                 with open(self.registry_path, encoding="utf-8") as f:
                     data = json.load(f)
-                # Validate shape: must be a dict with a dict "steps" field
-                if not isinstance(data, dict):
-                    return default_registry
-                if not isinstance(data.get("steps"), dict):
-                    data["steps"] = {}
-                return data
-            except (json.JSONDecodeError, ValueError, OSError, UnicodeError):
-                return default_registry
+            except OSError as exc:
+                raise OSError(
+                    f"Failed to read step registry at {self.registry_path}: {exc}"
+                ) from exc
+            except (json.JSONDecodeError, ValueError, UnicodeError) as exc:
+                raise OSError(
+                    f"Step registry at {self.registry_path} is corrupted: {exc}"
+                ) from exc
+            # Validate shape: must be a dict with a dict "steps" field
+            if not isinstance(data, dict):
+                raise OSError(
+                    f"Step registry at {self.registry_path} is corrupted: "
+                    "top-level value must be an object"
+                )
+            if not isinstance(data.get("steps"), dict):
+                raise OSError(
+                    f"Step registry at {self.registry_path} is corrupted: "
+                    "'steps' must be an object"
+                )
+            return data
         return default_registry
 
     def save(self) -> None:

src/specify_cli/workflows/step/command_info.py

@@ -35,12 +35,13 @@ def workflow_step_info(
 ):
     """Show details for a step type."""
     from .. import STEP_REGISTRY
-    from .catalog import StepCatalog, StepCatalogError, StepRegistry
+    from . import _helpers as step_helpers
+    from .catalog import StepCatalog, StepCatalogError
 
     project_root = cli._require_specify_project()
     safe_step_id = cli._escape_markup(str(step_id))
 
-    registry = StepRegistry(project_root)
+    registry = step_helpers._load_step_registry_or_exit(project_root)
     installed_meta = registry.get(step_id)
 
     # Check if it's a built-in

src/specify_cli/workflows/step/command_list.py

@@ -10,15 +10,15 @@
 def workflow_step_list():
     """List installed step types (built-in and custom)."""
     from .. import STEP_REGISTRY
-    from .catalog import StepRegistry
+    from . import _helpers as step_helpers
 
     project_root = cli._require_specify_project()
     specify_dir = project_root / ".specify"
 
-    # Read installed custom steps from registry only — no dynamic imports
+    # Read installed custom steps from registry only �?" no dynamic imports
     installed: dict = {}
     if specify_dir.exists():
-        registry = StepRegistry(project_root)
+        registry = step_helpers._load_step_registry_or_exit(project_root)
         installed = registry.list()
 
     cli.console.print("\n[bold cyan]Installed Step Types:[/bold cyan]\n")

src/specify_cli/workflows/step/command_remove.py

@@ -17,13 +17,15 @@ def _remove_step_locked(project_root: cli.Path, step_id: str) -> None:
     """
     import shutil
 
-    from .catalog import StepRegistry, StepValidationError
+    from .catalog import StepValidationError
 
     safe_step_id = cli._escape_markup(step_id)
-    registry = StepRegistry(project_root)
+    # Directory-level symlink refusal runs first so it keeps its specific
+    # message; registry construction then surfaces read errors cleanly.
+    steps_base_dir = step_helpers._resolve_steps_base_dir_or_exit(project_root)
+    registry = step_helpers._load_step_registry_or_exit(project_root)
     in_registry = registry.is_installed(step_id)
 
-    steps_base_dir = step_helpers._resolve_steps_base_dir_or_exit(project_root)
     step_dir = (steps_base_dir / step_id).resolve()
     # Defense-in-depth: even though step_helpers._validate_step_id_or_exit rejects path
     # separators, ensure that the resolved directory is a single child of

src/specify_cli/workflows/step/installer.py

@@ -164,6 +164,21 @@ def resolve_steps_base_dir(project_root: Path) -> Path:
     return steps_base_dir
 
 
+def _load_step_registry(project_root: Path):
+    """Load the step registry, mapping read failures onto ``StepInstallError``.
+
+    ``StepRegistry`` raises ``OSError`` for a symlinked or corrupted registry
+    file, while callers of this module only handle ``StepInstallError``, so
+    translate the error instead of letting it escape as a traceback.
+    """
+    from .catalog import StepRegistry
+
+    try:
+        return StepRegistry(project_root)
+    except OSError as exc:
+        raise StepInstallError(str(exc)) from exc
+
+
 def _resolve_step_dir(steps_base_dir: Path, step_id: str) -> Path:
     """Return the canonical destination directory for ``step_id``."""
     step_dir = steps_base_dir / step_id
@@ -474,14 +489,12 @@ def check_installable(project_root: Path, step_id: str, *, force: bool = False)
     uses this to reject before a download; :func:`install_step_package`
     re-runs the same checks as defense-in-depth.
     """
-    from .catalog import StepRegistry
-
     validate_step_id(step_id)
     steps_base_dir = resolve_steps_base_dir(project_root)
     step_dir = _resolve_step_dir(steps_base_dir, step_id)
     _reject_unsafe_destination(step_dir)
     _reject_builtin_collision(step_id)
-    registry = StepRegistry(project_root)
+    registry = _load_step_registry(project_root)
     _check_duplicate(registry, step_id, step_dir, force=force)
     return step_dir
 
@@ -742,8 +755,6 @@ def install_step_package(
     ``source`` is exactly ``"catalog"``, ``"local"``, or ``"url"``. Returns the
     registry entry that was persisted.
     """
-    from .catalog import StepRegistry
-
     package_dir = Path(package_dir)
     if package_dir.is_symlink():
         raise StepInstallError(
@@ -774,7 +785,7 @@ def install_step_package(
         )
 
     _reject_builtin_collision(step_id)
-    registry = StepRegistry(project_root)
+    registry = _load_step_registry(project_root)
     _check_duplicate(registry, step_id, step_dir, force=force)
 
     # Validate source and all caller-controlled metadata before creating any
@@ -821,7 +832,7 @@ def install_step_package(
             step_dir = _resolve_step_dir(locked_base_dir, step_id)
             _reject_unsafe_destination(step_dir)
             _reject_builtin_collision(step_id)
-            registry = StepRegistry(project_root)
+            registry = _load_step_registry(project_root)
             _check_duplicate(registry, step_id, step_dir, force=force)
             _replace_install(
                 step_dir,

tests/test_workflows.py

@@ -9876,19 +9876,20 @@ def test_persistence(self, project_dir):
         registry2 = StepRegistry(project_dir)
         assert registry2.is_installed("deploy")
 
-    def test_corrupted_registry_resets(self, project_dir):
+    def test_corrupted_registry_raises(self, project_dir):
         from specify_cli.workflows.step.catalog import StepRegistry
 
         registry = StepRegistry(project_dir)
         registry.steps_dir.mkdir(parents=True, exist_ok=True)
         registry.registry_path.write_text("not json", encoding="utf-8")
 
-        # Loading again should reset
-        registry2 = StepRegistry(project_dir)
-        assert registry2.list() == {}
+        # Loading again must fail loudly instead of handing back an empty
+        # registry that hides the corruption.
+        with pytest.raises(OSError, match="corrupted"):
+            StepRegistry(project_dir)
 
-    def test_registry_missing_steps_key_resets(self, project_dir):
-        """Valid JSON but missing 'steps' key should not crash add/get."""
+    def test_registry_invalid_steps_key_raises(self, project_dir):
+        """Valid JSON but a non-dict 'steps' must fail loudly, not repair."""
         from specify_cli.workflows.step.catalog import StepRegistry
         import json as _json
 
@@ -9900,15 +9901,12 @@ def test_registry_missing_steps_key_resets(self, project_dir):
             encoding="utf-8",
         )
 
-        registry2 = StepRegistry(project_dir)
-        # Should be safe to call add/get without KeyError
-        assert registry2.list() == {}
-        registry2.add("deploy", {"name": "Deploy", "type_key": "deploy"})
-        assert registry2.is_installed("deploy")
+        with pytest.raises(OSError, match="'steps' must be an object"):
+            StepRegistry(project_dir)
 
     @pytest.mark.skipif(sys.platform == "win32", reason="chmod not reliable on Windows")
-    def test_registry_unreadable_file_resets(self, project_dir):
-        """OSError reading the registry file should fall back to default."""
+    def test_registry_unreadable_file_raises(self, project_dir):
+        """An OSError reading the registry file must surface to the caller."""
         from specify_cli.workflows.step.catalog import StepRegistry
         import json as _json
 
@@ -9922,12 +9920,13 @@ def test_registry_unreadable_file_resets(self, project_dir):
         # Make it unreadable
         registry.registry_path.chmod(0o000)
         try:
-            registry2 = StepRegistry(project_dir)
-            assert registry2.list() == {}
+            with pytest.raises(OSError, match="Failed to read"):
+                StepRegistry(project_dir)
         finally:
             registry.registry_path.chmod(0o644)
 
         # After restoring permissions the registry is fully functional
+        registry2 = StepRegistry(project_dir)
         registry2.add("deploy", {"name": "Deploy", "type_key": "deploy"})
         assert registry2.is_installed("deploy")
 
@@ -9946,20 +9945,23 @@ def test_registry_load_refuses_symlinked_steps_dir(self, project_dir):
         steps_link = project_dir / ".specify" / "workflows" / "steps"
         steps_link.symlink_to(outside, target_is_directory=True)
 
-        registry = StepRegistry(project_dir)
-        assert registry.list() == {}
+        with pytest.raises(OSError, match="symlink"):
+            StepRegistry(project_dir)
 
     @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
     def test_registry_save_refuses_symlinked_steps_dir(self, project_dir):
         """save() must refuse symlinked registry paths (defense-in-depth)."""
         from specify_cli.workflows.step.catalog import StepRegistry, StepValidationError
 
+        # Build the registry before swapping in the symlink so the refusal
+        # comes from save() itself rather than the constructor's read check.
+        registry = StepRegistry(project_dir)
+
         outside = project_dir.parent / "outside-steps-save"
         outside.mkdir(parents=True, exist_ok=True)
         steps_link = project_dir / ".specify" / "workflows" / "steps"
         steps_link.symlink_to(outside, target_is_directory=True)
 
-        registry = StepRegistry(project_dir)
         with pytest.raises(StepValidationError, match="symlinked path"):
             registry.save()
 
@@ -11412,3 +11414,49 @@ def boom(*args, **kwargs):
         assert fresh.get("align-wf")["version"] == "1.0.0"
         assert stat.S_IMODE(registry.registry_path.stat().st_mode) == 0o644
         assert not list(registry.workflows_dir.glob("*.tmp"))
+
+
+class TestStepRegistryAtomicSaveRegression:
+    """StepRegistry must fail loudly on corruption and save atomically."""
+
+    @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
+    def test_load_raises_on_symlinked_file(self, tmp_path):
+        from specify_cli.workflows.step.catalog import StepRegistry
+
+        steps_dir = tmp_path / ".specify" / "workflows" / "steps"
+        steps_dir.mkdir(parents=True)
+        registry_path = steps_dir / StepRegistry.REGISTRY_FILE
+        real_file = tmp_path / "real_registry.json"
+        real_file.write_text(
+            json.dumps({"schema_version": "1.0", "steps": {}}), encoding="utf-8"
+        )
+        registry_path.symlink_to(real_file)
+
+        with pytest.raises(OSError, match="symlink"):
+            StepRegistry(tmp_path)
+
+    def test_load_raises_on_corrupted_json(self, tmp_path):
+        from specify_cli.workflows.step.catalog import StepRegistry
+
+        steps_dir = tmp_path / ".specify" / "workflows" / "steps"
+        steps_dir.mkdir(parents=True)
+        registry_path = steps_dir / StepRegistry.REGISTRY_FILE
+        registry_path.write_text("not valid json {{{", encoding="utf-8")
+
+        with pytest.raises(OSError, match="corrupted"):
+            StepRegistry(tmp_path)
+
+    def test_save_uses_atomic_write(self, tmp_path):
+        from specify_cli.workflows.step.catalog import StepRegistry
+
+        registry = StepRegistry(tmp_path)
+        registry.add("test-step", {"type": "command", "config": {}})
+        registry.save()
+
+        registry_path = (
+            tmp_path / ".specify" / "workflows" / "steps" / StepRegistry.REGISTRY_FILE
+        )
+        assert registry_path.exists()
+        data = json.loads(registry_path.read_text(encoding="utf-8"))
+        assert "test-step" in data.get("steps", {})
+        assert not list(registry_path.parent.glob(".*.tmp"))