fix(workflows): harden catalog.py against mis-shaped registry & non-string fields (#3375)

* fix(workflows): harden catalog.py against mis-shaped registry & non-string fields

Two robustness gaps where WorkflowRegistry/WorkflowCatalog diverged from
their StepRegistry/StepCatalog siblings, which already guard both:

- WorkflowRegistry._load returned json.load() verbatim, so a JSON-valid
  but mis-shaped registry (a list root, or a dict lacking a 'workflows'
  mapping) made is_installed/get/list/remove/add crash with
  TypeError/KeyError. Mirror StepRegistry._load: validate the shape and
  reset to default, and widen the except tuple to OSError/UnicodeError.
- WorkflowCatalog.search joined name/description/id without coercion, so a
  null or non-string field raised TypeError. Coerce with str(... or '')
  exactly as StepCatalog.search does.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* test(workflows): tighten mis-shaped-registry assertions

Per review: WorkflowRegistry.list() always returns a dict, so assert
'== {}' directly (the previous '== {} or == []' called list() twice and
admitted a shape it never returns), and reference
WorkflowRegistry.SCHEMA_VERSION instead of hard-coding '1.0'.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/specify_cli/workflows/catalog.py

@@ -76,8 +76,16 @@ class WorkflowRegistry:
         if self.registry_path.exists():
             try:
                 with open(self.registry_path, encoding="utf-8") as f:
-                    return json.load(f)
-            except (json.JSONDecodeError, ValueError):
+                    data = json.load(f)
+                # Validate shape: must be a dict with a dict "workflows" field,
+                # otherwise every method that indexes data["workflows"] crashes.
+                # Mirrors StepRegistry._load.
+                if not isinstance(data, dict):
+                    return {"schema_version": self.SCHEMA_VERSION, "workflows": {}}
+                if not isinstance(data.get("workflows"), dict):
+                    data["workflows"] = {}
+                return data
+            except (json.JSONDecodeError, ValueError, OSError, UnicodeError):
                 # Corrupted registry file — reset to default
                 return {"schema_version": self.SCHEMA_VERSION, "workflows": {}}
         return {"schema_version": self.SCHEMA_VERSION, "workflows": {}}
@@ -438,9 +446,9 @@ class WorkflowCatalog:
                 q = query.lower()
                 searchable = " ".join(
                     [
-                        wf_data.get("name", ""),
-                        wf_data.get("description", ""),
-                        wf_data.get("id", ""),
+                        str(wf_data.get("name") or ""),
+                        str(wf_data.get("description") or ""),
+                        str(wf_data.get("id") or ""),
                     ]
                 ).lower()
                 if q not in searchable:

tests/test_workflows.py

@@ -4746,12 +4746,59 @@ class TestWorkflowRegistry:
         registry2 = WorkflowRegistry(project_dir)
         assert registry2.is_installed("test-wf")
 
+    @pytest.mark.parametrize("bad_content", ["[]", '{"schema_version": "1.0"}'])
+    def test_load_tolerates_misshaped_registry(self, project_dir, bad_content):
+        """A JSON-valid but mis-shaped registry file must not crash every method.
+
+        A list root, or a dict lacking a 'workflows' mapping, previously made
+        is_installed/get/list/remove/add raise TypeError/KeyError. Mirrors the
+        shape guard StepRegistry._load already has.
+        """
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        reg_path = project_dir / ".specify" / "workflows" / "workflow-registry.json"
+        reg_path.parent.mkdir(parents=True, exist_ok=True)
+        reg_path.write_text(bad_content, encoding="utf-8")
+
+        registry = WorkflowRegistry(project_dir)
+        assert registry.data == {
+            "schema_version": WorkflowRegistry.SCHEMA_VERSION,
+            "workflows": {},
+        }
+        # None of these should raise on the recovered-default shape.
+        assert registry.is_installed("x") is False
+        assert registry.get("x") is None
+        assert registry.list() == {}  # list() always returns a dict
+        registry.remove("x")
+        registry.add("x", {"name": "X"})
+        assert registry.is_installed("x")
+
 
 # ===== Workflow Catalog Tests =====
 
 class TestWorkflowCatalog:
     """Test WorkflowCatalog catalog resolution."""
 
+    def test_search_with_non_string_fields(self, project_dir, monkeypatch):
+        """Non-string workflow fields (null/int name/description) must not
+        raise TypeError in search — StepCatalog.search already coerces these."""
+        from specify_cli.workflows.catalog import WorkflowCatalog
+
+        catalog = WorkflowCatalog(project_dir)
+        monkeypatch.setattr(catalog, "_get_merged_workflows", lambda **kw: {
+            "42": {
+                "id": 42,
+                "name": None,
+                "description": 99,
+                "_catalog_name": "test",
+                "_install_allowed": True,
+            },
+        })
+
+        assert len(catalog.search()) == 1
+        assert len(catalog.search(query="42")) == 1
+        assert len(catalog.search(query="missing")) == 0
+
     def test_default_catalogs(self, project_dir, monkeypatch):
         from specify_cli.workflows.catalog import WorkflowCatalog