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

@@ -104,29 +104,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
     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,