fix(workflows): StepRegistry.add tolerates a corrupted non-dict existing entry (#3630)

StepRegistry.add read existing = self.data['steps'].get(step_id, {}) then called
existing.get('installed_at', ...). A corrupted-but-parseable registry holding a
non-dict entry (e.g. {'steps': {'foo': 'corrupted'}}) — which _load() accepts,
since it validates only the top-level dict and that 'steps' is a dict — made
add() raise AttributeError. WorkflowRegistry.add was hardened for exactly this
(#3419); mirror its isinstance guard so a non-dict existing entry is treated as
absent.

Test copies the WorkflowRegistry sibling test for StepRegistry (fails before:
AttributeError on existing.get()).

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: jawwad-ali

src/specify_cli/workflows/catalog.py

@@ -894,7 +894,11 @@ def add(self, step_id: str, metadata: dict[str, Any]) -> None:
         import copy
         from datetime import datetime, timezone
 
-        existing = self.data["steps"].get(step_id, {})
+        raw_existing = self.data["steps"].get(step_id)
+        # Corrupted-but-parseable registries may hold non-dict entries; treat
+        # them as absent rather than crashing on existing.get() (mirrors
+        # WorkflowRegistry.add).
+        existing = raw_existing if isinstance(raw_existing, dict) else {}
         metadata_to_store = copy.deepcopy(metadata)
         metadata_to_store["installed_at"] = existing.get(
             "installed_at", datetime.now(timezone.utc).isoformat()

tests/test_workflows.py

@@ -10015,6 +10015,17 @@ def test_registry_add_survives_non_dict_existing_entry(self, project_dir):
         registry.add("align-wf", {"version": "1.0.0", "source": "catalog"})
         assert registry.get("align-wf")["version"] == "1.0.0"
 
+    def test_step_registry_add_survives_non_dict_existing_entry(self, project_dir):
+        """StepRegistry.add must treat a corrupted non-dict existing entry as
+        absent rather than crash on existing.get() (parity with
+        WorkflowRegistry.add)."""
+        from specify_cli.workflows.catalog import StepRegistry
+
+        registry = StepRegistry(project_dir)
+        registry.data["steps"]["my-step"] = "corrupted"
+        registry.add("my-step", {"version": "1.0.0"})
+        assert registry.get("my-step")["version"] == "1.0.0"
+
     @pytest.mark.parametrize(
         "contents",
         [