feat(workflows): select exact workflow catalog releases

docs/reference/workflows.md

@@ -105,6 +105,7 @@ specify workflow add <source>
 | --------------- | ------------------------------------------------------ |
 | `--dev`         | Install from a local YAML file, package directory, or archive |
 | `--from <url>`  | Install from a custom URL (`<source>` names the expected workflow ID) |
+| `--version <version>` | Install an exact advertised catalog release (`<source>` must be a workflow ID) |
 
 Installs a workflow from the catalog, an HTTPS URL, a local YAML file, a
 directory containing `workflow.yml`, or a `.zip`, `.tar.gz`, or `.tgz`
@@ -115,6 +116,37 @@ Directory and archive installs preserve the complete workflow package,
 including scripts and other companion files. ZIP, `.tar.gz`, and `.tgz`
 archives follow the same validation and installation behavior.
 
+Catalog entries keep the current release's `version`, `url`, optional `sha256`,
+and optional `requires` at the top level. An optional `releases` mapping
+advertises historical versions without changing what unqualified `add`,
+`search`, `info`, or `update` select:
+
+```json
+{
+  "id": "example",
+  "version": "2.0.0",
+  "url": "https://example.com/example-2.0.0.zip",
+  "releases": {
+    "1.0.0": {
+      "url": "https://example.com/example-1.0.0.zip",
+      "sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+      "requires": {"speckit_version": ">=1.0.0"}
+    }
+  }
+}
+```
+
+Each historical release needs its own URL and SHA-256 digest; `requires` is
+optional and, when present, must match the downloaded workflow definition.
+Advertised versions use the workflow definition's `X.Y.Z` version format;
+`--version` also accepts equivalent spellings such as `v1.0` when selecting an
+advertised `1.0.0` release.
+The requested version must exist in the highest-priority catalog that provides
+the workflow. A missing version does not fall back to another source, and
+discovery-only catalogs cannot be installed from. The downloaded workflow ID,
+version, and declared digest are verified before installation. `--version` does
+not apply to local paths, direct URLs, or `--from` installations.
+
 ## Workflow Overlays
 
 Workflow overlays let a project extend or override an installed workflow without editing the installed `workflow.yml`.  This keeps local customizations safe across `specify bundle update` or `specify workflow add` upgrades.
@@ -378,9 +410,14 @@ Searches all active catalogs for workflows matching the query.
 
 ```bash
 specify workflow info <workflow_id>
+specify workflow info <workflow_id> --versions
 ```
 
 Shows detailed information about a workflow, including its steps, inputs, and requirements.
+`--versions` lists the current catalog version followed by advertised historical
+versions and indicates whether the winning catalog is installable or
+discovery-only (not installable). It also works when a different version is
+installed locally.
 
 ## Catalog Management
 

src/specify_cli/workflows/_commands.py

@@ -724,6 +724,7 @@ def _install_workflow_package(
     *,
     expected_id: str | None = None,
     expected_version: str | None = None,
+    expected_requires: dict[str, Any] | None = None,
     expected_installed_version: str | None = None,
     catalog_info: dict[str, Any] | None = None,
 ) -> None:
@@ -768,6 +769,12 @@ def _install_workflow_package(
             f"version ({_escape_markup(expected_version)})."
         )
         raise typer.Exit(1)
+    if expected_requires is not None and definition.requires != expected_requires:
+        console.print(
+            "[red]Error:[/red] Downloaded workflow requirements do not match "
+            "the selected catalog release."
+        )
+        raise typer.Exit(1)
 
     dest_dir = _safe_workflow_id_dir(workflows_dir, definition.id)
     staged_dir = Path(
@@ -1067,6 +1074,7 @@ def _install_workflow_from_catalog(
     workflow_id: str,
     expected_version: str | None = None,
     expected_installed_version: str | None = None,
+    requested_version: str | None = None,
 ) -> None:
     """Download, validate, and register a catalog workflow.
 
@@ -1094,15 +1102,28 @@ def versions_match(actual: object, expected: str) -> bool:
 
     catalog = WorkflowCatalog(project_root)
     try:
-        info = catalog.get_workflow_info(workflow_id)
+        info = (
+            catalog.get_workflow_info(workflow_id, requested_version)
+            if requested_version is not None
+            else catalog.get_workflow_info(workflow_id)
+        )
     except WorkflowCatalogError as exc:
         console.print(f"[red]Error:[/red] {_escape_markup(str(exc))}")
         raise typer.Exit(1)
 
     if not info:
+        if requested_version is not None:
+            console.print(
+                f"[red]Error:[/red] Workflow '{safe_wf_id}' version "
+                f"'{_escape_markup(requested_version)}' not found in the winning catalog."
+            )
+            raise typer.Exit(1)
         console.print(f"[red]Error:[/red] Workflow '{safe_wf_id}' not found in catalog")
         raise typer.Exit(1)
 
+    if requested_version is not None:
+        expected_version = info["version"]
+
     if not info.get("_install_allowed", True):
         console.print(f"[yellow]Warning:[/yellow] Workflow '{safe_wf_id}' is from a discovery-only catalog")
         console.print("Direct installation is not enabled for this catalog source.")
@@ -1235,14 +1256,17 @@ def versions_match(actual: object, expected: str) -> bool:
         console.print(f"[red]Error:[/red] Failed to install workflow '{safe_wf_id}' from catalog: {_escape_markup(str(exc))}")
         raise typer.Exit(1)
 
+    try:
+        verify_archive_sha256(
+            downloaded_content, info.get("sha256"), workflow_id, ValueError
+        )
+    except ValueError as exc:
+        _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
+        console.print(f"[red]Error:[/red] {_escape_markup(str(exc))}")
+        raise typer.Exit(1)
+
     if downloaded_archive_format is not None:
         try:
-            verify_archive_sha256(
-                downloaded_content,
-                info.get("sha256"),
-                workflow_id,
-                ValueError,
-            )
             import tempfile
             from io import BytesIO
 
@@ -1270,6 +1294,9 @@ def versions_match(actual: object, expected: str) -> bool:
                     workflow_url,
                     expected_id=workflow_id,
                     expected_version=expected_version,
+                    expected_requires=(
+                        info.get("requires") if requested_version is not None else None
+                    ),
                     expected_installed_version=expected_installed_version,
                     catalog_info={**info, "url": workflow_url},
                 )
@@ -1321,15 +1348,31 @@ def versions_match(actual: object, expected: str) -> bool:
     # A stale or misconfigured URL can serve a different version than the
     # catalog advertised; without this check `update` would report success
     # while leaving the old version installed (or even downgrading).
-    if expected_version is not None:
-        if not versions_match(definition.version, expected_version):
-            _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
-            console.print(
-                f"[red]Error:[/red] Downloaded workflow version ({_escape_markup(str(definition.version))}) "
-                f"does not match the catalog version ({_escape_markup(expected_version)}). "
-                f"The catalog entry may be stale or misconfigured."
-            )
-            raise typer.Exit(1)
+    if expected_version is not None and not (
+        str(definition.version) == expected_version
+        if requested_version is not None
+        else versions_match(definition.version, expected_version)
+    ):
+        _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
+        console.print(
+            f"[red]Error:[/red] Downloaded workflow version ({_escape_markup(str(definition.version))}) "
+            f"does not match the catalog version ({_escape_markup(expected_version)}). "
+            f"The catalog entry may be stale or misconfigured."
+        )
+        raise typer.Exit(1)
+    if (
+        requested_version is not None
+        and "requires" in info
+        and definition.requires != info["requires"]
+    ):
+        _safe_discard_staged_workflow_file(
+            staged_file, workflow_dir, existed_before
+        )
+        console.print(
+            "[red]Error:[/red] Downloaded workflow requirements do not match "
+            "the selected catalog release."
+        )
+        raise typer.Exit(1)
 
     try:
         transaction = _workflow_install_transaction(project_root)

src/specify_cli/workflows/catalog/_domain.py

@@ -692,13 +692,36 @@ def search(
             results.append(wf_data)
         return results
 
-    def get_workflow_info(self, workflow_id: str) -> dict[str, Any] | None:
-        """Get details for a specific workflow from the catalog."""
+    def get_workflow_info(
+        self, workflow_id: str, version: str | None = None
+    ) -> dict[str, Any] | None:
+        """Get the current or an exact advertised release from the winning source."""
+        from ._versions import select_release
+
+        merged = self._get_merged_workflows()
+        wf = merged.get(workflow_id)
+        if wf is None:
+            return None
+        wf.setdefault("id", workflow_id)
+        return select_release(wf, version)
+
+    def get_workflow_versions(self, workflow_id: str) -> list[str]:
+        """List versions advertised by the winning catalog entry."""
+        details = self.get_workflow_version_details(workflow_id)
+        return details[0] if details is not None else []
+
+    def get_workflow_version_details(
+        self, workflow_id: str
+    ) -> tuple[list[str], bool] | None:
+        """Return advertised versions and whether their source allows installation."""
+        from ._versions import available_versions
+
         merged = self._get_merged_workflows()
         wf = merged.get(workflow_id)
-        if wf:
-            wf.setdefault("id", workflow_id)
-        return wf
+        if wf is None:
+            return None
+        wf.setdefault("id", workflow_id)
+        return available_versions(wf), bool(wf.get("_install_allowed", True))
 
     def get_catalog_configs(self) -> list[dict[str, Any]]:
         """Return current catalog configuration as a list of dicts."""

src/specify_cli/workflows/catalog/_versions.py

@@ -0,0 +1,129 @@
+"""Version selection for workflow catalog entries."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from packaging.version import InvalidVersion, Version
+
+from ..._download_security import is_https_or_localhost_http
+from ..engine import _is_valid_workflow_version
+from ._domain import WorkflowValidationError
+
+_SHA256 = re.compile(r"^(?:sha256:)?[0-9a-fA-F]{64}$", re.IGNORECASE)
+_CURRENT_ONLY = frozenset(
+    {"version", "url", "sha256", "requires", "bundled", "releases"}
+)
+_RESERVED = frozenset(
+    {"id", "version", "releases", "_catalog_name", "_install_allowed"}
+)
+
+
+def _validated_releases(entry: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    if "releases" not in entry:
+        return {}
+    releases = entry["releases"]
+    workflow_id = entry.get("id", "<unknown>")
+    if not isinstance(releases, dict):
+        raise WorkflowValidationError(
+            f"Workflow '{workflow_id}' has an invalid releases mapping."
+        )
+
+    current = entry.get("version")
+    if not isinstance(current, str) or not current.strip():
+        raise WorkflowValidationError(
+            f"Workflow '{workflow_id}' has releases but no current version."
+        )
+    if not _is_valid_workflow_version(current):
+        raise WorkflowValidationError(
+            f"Workflow '{workflow_id}' has an invalid current version '{current}'."
+        )
+    try:
+        seen = {Version(current)}
+    except InvalidVersion:
+        raise WorkflowValidationError(
+            f"Workflow '{workflow_id}' has an invalid current version '{current}'."
+        ) from None
+
+    for release_version, record in releases.items():
+        if not isinstance(release_version, str) or not release_version.strip():
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' has an invalid release version key."
+            )
+        if not _is_valid_workflow_version(release_version):
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' has invalid release version '{release_version}'."
+            )
+        try:
+            normalized = Version(release_version)
+        except InvalidVersion:
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' has invalid release version '{release_version}'."
+            ) from None
+        if normalized in seen:
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' repeats release version '{release_version}'."
+            )
+        seen.add(normalized)
+        if not isinstance(record, dict):
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' release '{release_version}' must be an object."
+            )
+        if _RESERVED.intersection(record):
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' release '{release_version}' contains reserved fields."
+            )
+        if not isinstance(record.get("url"), str) or not record["url"].strip():
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' release '{release_version}' needs a url."
+            )
+        if not is_https_or_localhost_http(record["url"]):
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' release '{release_version}' has an invalid URL."
+            )
+        if not isinstance(record.get("sha256"), str) or not _SHA256.fullmatch(
+            record["sha256"]
+        ):
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' release '{release_version}' needs a SHA-256 digest."
+            )
+        if "requires" in record and not isinstance(record["requires"], dict):
+            raise WorkflowValidationError(
+                f"Workflow '{workflow_id}' release '{release_version}' has invalid requires."
+            )
+    return releases
+
+
+def select_release(entry: dict[str, Any], version: str | None) -> dict[str, Any] | None:
+    """Select from the winning source, preserving the advertised version."""
+    releases = _validated_releases(entry)
+    current = entry.get("version")
+    if version is None or version == current:
+        return entry
+    try:
+        requested = Version(version)
+    except InvalidVersion:
+        return None
+    if isinstance(current, str):
+        try:
+            if requested == Version(current):
+                return entry
+        except InvalidVersion:
+            pass  # Legacy entries without history can use non-PEP-440 versions.
+    for advertised, record in releases.items():
+        if requested == Version(advertised):
+            common = {
+                key: value for key, value in entry.items() if key not in _CURRENT_ONLY
+            }
+            return {**common, **record, "version": advertised}
+    return None
+
+
+def available_versions(entry: dict[str, Any]) -> list[str]:
+    """Current first, followed by historical versions newest to oldest."""
+    releases = _validated_releases(entry)
+    current = entry.get("version")
+    if not isinstance(current, str) or not current:
+        return []
+    return [current, *sorted(releases, key=Version, reverse=True)]