fix(workflows): preserve disabled state on update, guard corrupted registry entries

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

src/specify_cli/workflows/_commands.py

@@ -353,7 +353,7 @@ def workflow_run(
         from .catalog import WorkflowRegistry
 
         installed_meta = WorkflowRegistry(project_root).get(source)
-        if installed_meta is not None and installed_meta.get("enabled", True) is False:
+        if isinstance(installed_meta, dict) and installed_meta.get("enabled", True) is False:
             err.print(
                 f"[red]Error:[/red] Workflow '{_escape_markup(source)}' is disabled. "
                 f"Enable with: specify workflow enable {_escape_markup(source)}"
@@ -903,14 +903,19 @@ def _install_workflow_from_catalog(
         )
         raise typer.Exit(1)
 
-    registry.add(workflow_id, {
+    entry = {
         "name": definition.name or info.get("name", workflow_id),
         "version": definition.version or info.get("version", "0.0.0"),
         "description": definition.description or info.get("description", ""),
         "source": "catalog",
         "catalog_name": info.get("_catalog_name", ""),
         "url": workflow_url,
-    })
+    }
+    # Preserve a prior disabled state across updates/reinstalls.
+    existing = registry.get(workflow_id)
+    if isinstance(existing, dict) and existing.get("enabled", True) is False:
+        entry["enabled"] = False
+    registry.add(workflow_id, entry)
     console.print(f"[green]✓[/green] Workflow '{info.get('name', workflow_id)}' installed from catalog")
 
 
@@ -1009,7 +1014,10 @@ def workflow_update(
     updates_available: list[dict[str, str]] = []
     for wf_id in targets:
         safe_id = _escape_markup(str(wf_id))
-        metadata = installed.get(wf_id) or {}
+        metadata = installed.get(wf_id)
+        if not isinstance(metadata, dict):
+            console.print(f"⚠  {safe_id}: Registry entry is corrupted (skipping)")
+            continue
         if metadata.get("source") != "catalog":
             console.print(f"⚠  {safe_id}: Installed from a local path or URL — re-add to update (skipping)")
             continue
@@ -1097,6 +1105,11 @@ def workflow_enable(
     if metadata is None:
         console.print(f"[red]Error:[/red] Workflow '{_escape_markup(workflow_id)}' is not installed")
         raise typer.Exit(1)
+    if not isinstance(metadata, dict):
+        console.print(
+            f"[red]Error:[/red] Registry entry for '{_escape_markup(workflow_id)}' is corrupted"
+        )
+        raise typer.Exit(1)
     if metadata.get("enabled", True):
         console.print(f"[yellow]Workflow '{_escape_markup(workflow_id)}' is already enabled[/yellow]")
         raise typer.Exit(0)
@@ -1118,6 +1131,11 @@ def workflow_disable(
     if metadata is None:
         console.print(f"[red]Error:[/red] Workflow '{_escape_markup(workflow_id)}' is not installed")
         raise typer.Exit(1)
+    if not isinstance(metadata, dict):
+        console.print(
+            f"[red]Error:[/red] Registry entry for '{_escape_markup(workflow_id)}' is corrupted"
+        )
+        raise typer.Exit(1)
     if not metadata.get("enabled", True):
         console.print(f"[yellow]Workflow '{_escape_markup(workflow_id)}' is already disabled[/yellow]")
         raise typer.Exit(0)

tests/test_workflows.py

@@ -7435,6 +7435,80 @@ steps:
         assert meta["version"] == "2.0.0"
         assert "2.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8")
 
+    def test_update_preserves_disabled_state(self, project_dir, monkeypatch):
+        from unittest.mock import patch
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry
+
+        monkeypatch.chdir(project_dir)
+        WorkflowRegistry(project_dir).add("align-wf", {
+            "name": "Align Workflow",
+            "version": "1.0.0",
+            "description": "",
+            "source": "catalog",
+            "url": "https://example.com/workflow.yml",
+            "enabled": False,
+        })
+        monkeypatch.setattr(
+            WorkflowCatalog,
+            "get_workflow_info",
+            lambda self, wid: {
+                "id": wid,
+                "version": "2.0.0",
+                "url": "https://example.com/workflow.yml",
+                "_install_allowed": True,
+            },
+        )
+        data = self.WORKFLOW_YAML.format(version="2.0.0").encode()
+        runner = CliRunner()
+        with patch(
+            "specify_cli.authentication.http.open_url",
+            side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url),
+        ):
+            result = runner.invoke(app, ["workflow", "update"], input="y\n")
+        assert result.exit_code == 0, result.output
+        meta = WorkflowRegistry(project_dir).get("align-wf")
+        assert meta["version"] == "2.0.0"
+        assert meta["enabled"] is False
+
+    def test_update_skips_corrupted_registry_entry(self, project_dir, monkeypatch):
+        import json
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        monkeypatch.chdir(project_dir)
+        registry_path = WorkflowRegistry(project_dir).registry_path
+        registry_path.parent.mkdir(parents=True, exist_ok=True)
+        registry_path.write_text(
+            json.dumps({"schema_version": "1.0", "workflows": {"broken": "not-a-dict"}}),
+            encoding="utf-8",
+        )
+        runner = CliRunner()
+        result = runner.invoke(app, ["workflow", "update"])
+        assert result.exit_code == 0, result.output
+        assert "corrupted" in result.output
+
+    def test_enable_disable_corrupted_registry_entry_errors(self, project_dir, monkeypatch):
+        import json
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        monkeypatch.chdir(project_dir)
+        registry_path = WorkflowRegistry(project_dir).registry_path
+        registry_path.parent.mkdir(parents=True, exist_ok=True)
+        registry_path.write_text(
+            json.dumps({"schema_version": "1.0", "workflows": {"broken": "not-a-dict"}}),
+            encoding="utf-8",
+        )
+        runner = CliRunner()
+        for cmd in ("enable", "disable"):
+            result = runner.invoke(app, ["workflow", cmd, "broken"])
+            assert result.exit_code != 0
+            assert "corrupted" in result.output
+
     def test_update_up_to_date_reports_and_exits_zero(self, project_dir, monkeypatch):
         from typer.testing import CliRunner
         from specify_cli import app