fix(workflows): load custom steps before update

src/specify_cli/workflows/command_update.py

@@ -14,9 +14,11 @@ def workflow_update(
     """Update installed workflow(s) to the latest catalog version."""
     from packaging import version as pkg_version
 
+    from . import load_custom_steps
     from .catalog import WorkflowCatalog, WorkflowCatalogError
 
     project_root = cli._require_specify_project()
+    load_custom_steps(project_root)
     registry = cli._open_workflow_registry(project_root)
     workflows_dir = project_root / ".specify" / "workflows"
     cli._reject_unsafe_dir(project_root / ".specify", ".specify")

tests/specify_cli/workflows/test_command_update.py

@@ -5,6 +5,7 @@
 import json
 import os
 import stat
+import sys
 from pathlib import Path
 
 import pytest
@@ -41,6 +42,49 @@ def _install_dev(self, runner, app, project_dir):
         assert result.exit_code == 0, result.output
         return src
 
+    @staticmethod
+    def _write_custom_step(project_dir, type_key):
+        step_dir = project_dir / ".specify" / "workflows" / "steps" / type_key
+        step_dir.mkdir(parents=True)
+        (step_dir / "step.yml").write_text(
+            "\n".join(
+                (
+                    'schema_version: "1.0"',
+                    "step:",
+                    f'  type_key: "{type_key}"',
+                    '  name: "Test Update Step"',
+                    '  version: "1.0.0"',
+                    "",
+                )
+            ),
+            encoding="utf-8",
+        )
+        (step_dir / "__init__.py").write_text(
+            "\n".join(
+                (
+                    "from specify_cli.workflows.base import StepBase, StepResult",
+                    "",
+                    "",
+                    "class TestUpdateStep(StepBase):",
+                    f'    type_key = "{type_key}"',
+                    "",
+                    "    def execute(self, config, context):",
+                    "        return StepResult()",
+                    "",
+                )
+            ),
+            encoding="utf-8",
+        )
+
+    @staticmethod
+    def _custom_step_module_name(type_key):
+        import hashlib
+        import re
+
+        safe_key = re.sub(r"[^A-Za-z0-9_]", "_", type_key)
+        key_hash = hashlib.sha256(type_key.encode()).hexdigest()[:8]
+        return f"_speckit_custom_step_{safe_key}_{key_hash}"
+
     class _FakeResponse:
         def __init__(self, data, url="https://example.com/workflow.yml", headers=None):
             self._data = data
@@ -202,6 +246,132 @@ def test_update_installs_newer_catalog_version(self, project_dir, monkeypatch):
         assert meta["version"] == "2.0.0"
         assert "2.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8")
 
+    def test_update_installs_newer_workflow_with_project_custom_step(
+        self, project_dir, monkeypatch
+    ):
+        """Updates validate against custom steps installed by this project."""
+        from unittest.mock import patch
+
+        from typer.testing import CliRunner
+        from specify_cli import app
+        import specify_cli.workflows as workflows
+        from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry
+
+        custom_type = "test-update-step"
+        module_name = self._custom_step_module_name(custom_type)
+        self._write_custom_step(project_dir, custom_type)
+        monkeypatch.chdir(project_dir)
+        monkeypatch.setattr(workflows, "STEP_REGISTRY", dict(workflows.STEP_REGISTRY))
+        try:
+            registry = WorkflowRegistry(project_dir)
+            registry.add("align-wf", {
+                "name": "Align Workflow",
+                "version": "1.0.0",
+                "description": "CLI alignment test workflow",
+                "source": "catalog",
+                "catalog_name": "test-catalog",
+                "url": "https://example.com/workflow.yml",
+            })
+            wf_dir = project_dir / ".specify" / "workflows" / "align-wf"
+            wf_dir.mkdir(parents=True)
+            (wf_dir / "workflow.yml").write_text(
+                self.WORKFLOW_YAML.format(version="1.0.0"), encoding="utf-8"
+            )
+            monkeypatch.setattr(
+                WorkflowCatalog,
+                "get_workflow_info",
+                lambda self, wid: {
+                    "id": wid,
+                    "name": "Align Workflow",
+                    "version": "2.0.0",
+                    "url": "https://example.com/workflow.yml",
+                    "_install_allowed": True,
+                    "_catalog_name": "test-catalog",
+                },
+            )
+            data = self.WORKFLOW_YAML.format(version="2.0.0").replace(
+                "type: shell", f"type: {custom_type}"
+            ).encode()
+            with patch(
+                "specify_cli.authentication.http.open_url",
+                side_effect=lambda url, timeout=None, extra_headers=None,
+                redirect_validator=None: self._FakeResponse(data, url),
+            ):
+                result = CliRunner().invoke(app, ["workflow", "update"], input="y\n")
+
+            assert result.exit_code == 0, result.output
+            assert custom_type in workflows.STEP_REGISTRY
+            assert WorkflowRegistry(project_dir).get("align-wf")["version"] == "2.0.0"
+            assert (wf_dir / "workflow.yml").read_bytes() == data
+        finally:
+            sys.modules.pop(module_name, None)
+
+    def test_update_rejects_unknown_step_without_changing_installed_workflow(
+        self, project_dir, monkeypatch
+    ):
+        """Loading a project custom step must not admit unrelated types."""
+        from unittest.mock import patch
+
+        from typer.testing import CliRunner
+        from specify_cli import app
+        import specify_cli.workflows as workflows
+        from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry
+
+        custom_type = "test-update-step"
+        module_name = self._custom_step_module_name(custom_type)
+        self._write_custom_step(project_dir, custom_type)
+        monkeypatch.chdir(project_dir)
+        monkeypatch.setattr(workflows, "STEP_REGISTRY", dict(workflows.STEP_REGISTRY))
+        try:
+            registry = WorkflowRegistry(project_dir)
+            registry.add("align-wf", {
+                "name": "Align Workflow",
+                "version": "1.0.0",
+                "description": "CLI alignment test workflow",
+                "source": "catalog",
+                "catalog_name": "test-catalog",
+                "url": "https://example.com/workflow.yml",
+            })
+            wf_dir = project_dir / ".specify" / "workflows" / "align-wf"
+            wf_dir.mkdir(parents=True)
+            workflow_file = wf_dir / "workflow.yml"
+            original_workflow = self.WORKFLOW_YAML.format(version="1.0.0").encode()
+            workflow_file.write_bytes(original_workflow)
+            original_metadata = dict(WorkflowRegistry(project_dir).get("align-wf"))
+            registry_file = registry.registry_path
+            original_registry = registry_file.read_bytes()
+
+            monkeypatch.setattr(
+                WorkflowCatalog,
+                "get_workflow_info",
+                lambda self, wid: {
+                    "id": wid,
+                    "name": "Align Workflow",
+                    "version": "2.0.0",
+                    "url": "https://example.com/workflow.yml",
+                    "_install_allowed": True,
+                    "_catalog_name": "test-catalog",
+                },
+            )
+            data = self.WORKFLOW_YAML.format(version="2.0.0").replace(
+                "type: shell", "type: unknown-update-step"
+            ).encode()
+            with patch(
+                "specify_cli.authentication.http.open_url",
+                side_effect=lambda url, timeout=None, extra_headers=None,
+                redirect_validator=None: self._FakeResponse(data, url),
+            ):
+                result = CliRunner().invoke(app, ["workflow", "update"], input="y\n")
+
+            assert result.exit_code != 0
+            assert "invalid type 'unknown-update-step'" in result.output
+            assert custom_type in workflows.STEP_REGISTRY
+            assert workflow_file.read_bytes() == original_workflow
+            assert registry_file.read_bytes() == original_registry
+            assert WorkflowRegistry(project_dir).get("align-wf") == original_metadata
+        finally:
+            sys.modules.pop(module_name, None)
+
     def test_update_downloaded_invalid_yaml_escapes_rich_markup(self, project_dir, monkeypatch):
         """A malformed downloaded workflow can quote the offending line verbatim; escape it before printing."""
         from unittest.mock import patch