fix: reject mismatched step catalog versions

src/specify_cli/workflows/_commands.py

@@ -3408,6 +3408,26 @@ def _safe_fetch(url: str) -> bytes:
             )
             raise typer.Exit(1)
 
+        catalog_version = info.get("version")
+        downloaded_version = step_meta.get("version")
+        if "version" in info and "version" in step_meta:
+            from packaging import version as pkg_version
+
+            try:
+                versions_match = pkg_version.Version(
+                    str(downloaded_version)
+                ) == pkg_version.Version(str(catalog_version))
+            except pkg_version.InvalidVersion:
+                versions_match = str(downloaded_version) == str(catalog_version)
+            if not versions_match:
+                console.print(
+                    f"[red]Error:[/red] step.yml version "
+                    f"({_escape_markup(repr(downloaded_version))}) does not match "
+                    f"the catalog version ({_escape_markup(repr(catalog_version))}). "
+                    "The catalog entry may be stale or misconfigured."
+                )
+                raise typer.Exit(1)
+
         # Write the two required files.
         try:
             (tmp_path / "step.yml").write_bytes(step_yml_content)

tests/test_workflows.py

@@ -10577,6 +10577,93 @@ def test_list_escapes_installed_metadata(
 
 
 class TestWorkflowStepAddCLI:
+    @pytest.mark.parametrize(
+        ("catalog_version", "downloaded_version"),
+        [
+            ("1.0.0", "2.0.0"),
+            ("1.0.0", 0),
+            ("1.0.0", False),
+            ("1.0.0", ""),
+            ("1.0.0", None),
+            ("release-a", " release-a "),
+        ],
+    )
+    def test_add_rejects_step_yml_version_mismatch(
+        self, project_dir, monkeypatch, catalog_version, downloaded_version
+    ):
+        from typer.testing import CliRunner
+
+        from specify_cli import app
+        from specify_cli.authentication import http as auth_http
+        from specify_cli.workflows.catalog import StepCatalog, StepRegistry
+
+        monkeypatch.chdir(project_dir)
+        monkeypatch.setattr(
+            StepCatalog,
+            "get_step_info",
+            lambda self, step_id: {
+                "id": step_id,
+                "name": "Test Step",
+                "version": catalog_version,
+                "url": "https://example.com/step.yml",
+                "init_url": "https://example.com/__init__.py",
+                "_install_allowed": True,
+            },
+        )
+        bodies = {
+            "https://example.com/step.yml": yaml.safe_dump(
+                {
+                    "step": {
+                        "type_key": "my-step",
+                        "version": downloaded_version,
+                    }
+                }
+            ).encode(),
+            "https://example.com/__init__.py": b"# custom step\n",
+        }
+
+        class _FakeResponse:
+            def __init__(self, url):
+                self.url = url
+                self.body = bodies[url]
+                self.offset = 0
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+            def getheader(self, name):
+                return None
+
+            def geturl(self):
+                return self.url
+
+            def read(self, size=-1):
+                if size < 0:
+                    size = len(self.body) - self.offset
+                chunk = self.body[self.offset : self.offset + size]
+                self.offset += len(chunk)
+                return chunk
+
+        monkeypatch.setattr(
+            auth_http,
+            "open_url",
+            lambda url, timeout=30, redirect_validator=None: _FakeResponse(url),
+        )
+
+        result = CliRunner().invoke(
+            app, ["workflow", "step", "add", "my-step"]
+        )
+
+        assert result.exit_code != 0
+        assert "does not match the catalog version" in result.output
+        assert not StepRegistry(project_dir).is_installed("my-step")
+        assert not (
+            project_dir / ".specify" / "workflows" / "steps" / "my-step"
+        ).exists()
+
     @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
     def test_add_rejects_symlinked_steps_base_dir(self, project_dir, monkeypatch):
         from typer.testing import CliRunner