Fix catalog validation precedence and selected workflow paths

Assisted-by: GitHub Copilot (model: GPT-6 Sol, autonomous)
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: mnriem

src/specify_cli/bundles/references.py

@@ -123,17 +123,15 @@ def _resolved_in_catalog(root: Path, component: ComponentRef) -> bool | str | No
     except Exception as exc:  # noqa: BLE001 - report malformed catalog errors
         from ..extensions import ExtensionCatalogFetchError
         from ..presets._catalog import PresetCatalogFetchError
-        from ..workflows.catalog import StepCatalogError, WorkflowCatalogError
+        from ..workflows.catalog import (
+            StepCatalogFetchError,
+            WorkflowCatalogFetchError,
+        )
 
-        if isinstance(exc, (ExtensionCatalogFetchError, PresetCatalogFetchError)):
-            return None
-        if isinstance(exc, WorkflowCatalogError) and str(exc) == (
-            "All configured catalogs failed to fetch."
-        ):
-            return None
-        if isinstance(exc, StepCatalogError) and str(exc) == (
-            "All configured step catalogs failed to fetch."
-        ):
+        if isinstance(exc, (
+            ExtensionCatalogFetchError, PresetCatalogFetchError,
+            WorkflowCatalogFetchError, StepCatalogFetchError,
+        )):
             return None
         return f"Catalog lookup failed: {exc}"
     return None

src/specify_cli/extensions/__init__.py

@@ -4808,7 +4808,7 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
             ) from e
 
     def _get_merged_extensions(
-        self, force_refresh: bool = False
+        self, force_refresh: bool = False, *, extension_id: str | None = None
     ) -> List[Dict[str, Any]]:
         """Fetch and merge extensions from all active catalogs.
 
@@ -4817,8 +4817,9 @@ def _get_merged_extensions(
           - _catalog_name: name of the source catalog
           - _install_allowed: whether installation is allowed from this catalog
 
-        Catalogs that fail to fetch are skipped. Raises ExtensionError only if
-        ALL catalogs fail.
+        An ID lookup stops at its first matching source and refuses malformed
+        higher-priority catalogs. Untargeted searches continue past malformed
+        sources so other catalog results remain discoverable.
 
         Args:
             force_refresh: If True, bypass all caches
@@ -4827,7 +4828,8 @@ def _get_merged_extensions(
             List of merged extension dicts
 
         Raises:
-            ExtensionError: If all catalogs fail to fetch
+            ExtensionError: If no catalog is readable, or an ID lookup
+                encounters malformed catalog data.
         """
         import sys
 
@@ -4846,13 +4848,19 @@ def _get_merged_extensions(
                     and validation_error is None
                 ):
                     validation_error = e
+                if extension_id is not None and isinstance(
+                    e, ExtensionCatalogValidationError
+                ):
+                    raise
                 print(
                     f"Warning: Could not fetch catalog '{catalog_entry.name}': {e}",
                     file=sys.stderr,
                 )
                 continue
 
             for ext_id, ext_data in catalog_data.get("extensions", {}).items():
+                if extension_id is not None and ext_id != extension_id:
+                    continue
                 # Per-entry guard: ``_fetch_single_catalog`` already validates
                 # that ``catalog_data["extensions"]`` is a mapping, but it
                 # does not (and should not) validate every entry shape there
@@ -4862,6 +4870,11 @@ def _get_merged_extensions(
                 # the valid entries without crashing on ``**ext_data``.
                 # Mirrors ``integrations/catalog.py:245``.
                 if not isinstance(ext_data, dict):
+                    if extension_id is not None:
+                        raise ExtensionCatalogValidationError(
+                            f"Invalid extension catalog entry for '{ext_id}' "
+                            f"from {catalog_entry.url}: expected a JSON object"
+                        )
                     continue
                 if ext_id not in merged:  # Higher-priority catalog wins
                     merged[ext_id] = {
@@ -4870,6 +4883,8 @@ def _get_merged_extensions(
                         "_catalog_name": catalog_entry.name,
                         "_install_allowed": catalog_entry.install_allowed,
                     }
+                    if extension_id is not None:
+                        return list(merged.values())
 
         if not any_success and active_catalogs:
             if validation_error is not None:
@@ -5093,7 +5108,7 @@ def get_extension_info(
             Extension metadata (annotated with ``_catalog_name`` and
             ``_install_allowed``) or None if not found.
         """
-        all_extensions = self._get_merged_extensions()
+        all_extensions = self._get_merged_extensions(extension_id=extension_id)
         for ext_data in all_extensions:
             if ext_data["id"] == extension_id:
                 from ._catalog_versions import select_release
@@ -5105,7 +5120,7 @@ def get_extension_versions(self, extension_id: str) -> list[str]:
         """List versions advertised by the winning catalog source."""
         from ._catalog_versions import available_versions
 
-        for ext_data in self._get_merged_extensions():
+        for ext_data in self._get_merged_extensions(extension_id=extension_id):
             if ext_data["id"] == extension_id:
                 return available_versions(ext_data)
         return []

src/specify_cli/workflows/catalog/__init__.py

@@ -29,13 +29,17 @@ def register(app: typer.Typer) -> None:
     "WorkflowCatalog",
     "WorkflowCatalogEntry",
     "WorkflowCatalogError",
+    "WorkflowCatalogFetchError",
+    "WorkflowCatalogValidationError",
     "WorkflowRegistry",
     "WorkflowValidationError",
 }
 _STEP_COMPATIBILITY_EXPORTS = {
     "StepCatalog",
     "StepCatalogEntry",
     "StepCatalogError",
+    "StepCatalogFetchError",
+    "StepCatalogValidationError",
     "StepRegistry",
     "StepValidationError",
 }

src/specify_cli/workflows/catalog/_domain.py

@@ -43,6 +43,14 @@ class WorkflowCatalogError(Exception):
     """Base error for workflow catalog operations."""
 
 
+class WorkflowCatalogFetchError(WorkflowCatalogError):
+    """A configured workflow catalog could not be fetched."""
+
+
+class WorkflowCatalogValidationError(WorkflowCatalogError):
+    """A workflow catalog supplied malformed metadata."""
+
+
 class WorkflowValidationError(WorkflowCatalogError):
     """Validation error for catalog config or workflow data."""
 
@@ -517,18 +525,36 @@ def _fetch_single_catalog(
         """Fetch a single catalog, using cache when possible."""
         cache_file, meta_file = self._get_cache_paths(entry.url)
 
+        def validate_payload(data: Any) -> dict[str, Any]:
+            if not isinstance(data, dict):
+                raise WorkflowCatalogValidationError(
+                    f"Catalog from {entry.url} is not a valid JSON object."
+                )
+            if "workflows" in data and not isinstance(data["workflows"], (dict, list)):
+                raise WorkflowCatalogValidationError(
+                    f"Catalog from {entry.url} has malformed workflows metadata."
+                )
+            return data
+
         if not force_refresh and self._is_url_cache_valid(entry.url):
             try:
                 with open(cache_file, encoding="utf-8") as f:
                     cached = json.load(f)
-                if isinstance(cached, dict):
-                    return cached
-            except (UnicodeDecodeError, json.JSONDecodeError, OSError):
+                return validate_payload(cached)
+            except (
+                UnicodeDecodeError,
+                json.JSONDecodeError,
+                OSError,
+                WorkflowCatalogValidationError,
+            ):
                 # Ignore invalid/unreadable cache and fall back to fetching from source.
                 pass
 
         # Fetch from URL — validate scheme before opening and after redirects
+        from http.client import HTTPException
         from urllib.parse import urlparse
+
+        from specify_cli.authentication.http import RedirectPolicyError
         from specify_cli.authentication.http import open_url as _open_url
 
         def _validate_catalog_url(url: str) -> None:
@@ -542,18 +568,18 @@ def _validate_catalog_url(url: str) -> None:
                 hostname = parsed.hostname
                 _ = parsed.port
             except (TypeError, ValueError):
-                raise WorkflowCatalogError(
+                raise WorkflowCatalogValidationError(
                     f"Refusing to fetch catalog from malformed URL: {url}"
                 ) from None
             is_localhost = hostname in ("localhost", "127.0.0.1", "::1")
             if parsed.scheme != "https" and not (
                 parsed.scheme == "http" and is_localhost
             ):
-                raise WorkflowCatalogError(
+                raise WorkflowCatalogValidationError(
                     f"Refusing to fetch catalog from non-HTTPS URL: {url}"
                 )
             if not hostname:
-                raise WorkflowCatalogError(
+                raise WorkflowCatalogValidationError(
                     f"Refusing to fetch catalog from URL with no hostname: {url}"
                 )
 
@@ -578,29 +604,39 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
                     read_response_limited(
                         resp,
                         max_bytes=_max_json_catalog_bytes(),
-                        error_type=WorkflowCatalogError,
+                        error_type=WorkflowCatalogValidationError,
                         label="workflow catalog",
                     ).decode("utf-8")
                 )
-        except Exception as exc:
+        except (
+            WorkflowCatalogValidationError,
+            RedirectPolicyError,
+            UnicodeError,
+            json.JSONDecodeError,
+        ) as exc:
+            raise WorkflowCatalogValidationError(
+                f"Invalid workflow catalog from {entry.url}: {exc}"
+            ) from exc
+        except (OSError, HTTPException) as exc:
             # Fall back to cache if available
             if cache_file.exists():
                 try:
                     with open(cache_file, encoding="utf-8") as f:
                         cached = json.load(f)
-                    if isinstance(cached, dict):
-                        return cached
-                except (json.JSONDecodeError, ValueError, OSError):
+                    return validate_payload(cached)
+                except (
+                    json.JSONDecodeError,
+                    ValueError,
+                    OSError,
+                    WorkflowCatalogValidationError,
+                ):
                     # Stale-cache read failed; let the original fetch error propagate.
                     pass
-            raise WorkflowCatalogError(
+            raise WorkflowCatalogFetchError(
                 f"Failed to fetch catalog from {entry.url}: {exc}"
             ) from exc
 
-        if not isinstance(data, dict):
-            raise WorkflowCatalogError(
-                f"Catalog from {entry.url} is not a valid JSON object."
-            )
+        data = validate_payload(data)
 
         # Write cache
         try:
@@ -615,26 +651,44 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
         return data
 
     def _get_merged_workflows(
-        self, force_refresh: bool = False
+        self, force_refresh: bool = False, *, workflow_id: str | None = None
     ) -> dict[str, dict[str, Any]]:
-        """Merge workflows from all active catalogs (lower priority number wins)."""
+        """Merge for search, or resolve one ID from the first valid winning source."""
         catalogs = self.get_active_catalogs()
         merged: dict[str, dict[str, Any]] = {}
         fetch_errors = 0
+        validation_error: WorkflowCatalogValidationError | None = None
 
-        # Process later/higher-numbered entries first so earlier/lower-numbered
-        # entries overwrite them on workflow ID conflicts.
-        for entry in reversed(catalogs):
+        # Search uses overwrite order; exact-ID lookup visits the highest
+        # priority source first and stops at its matching entry.
+        sources = catalogs if workflow_id is not None else reversed(catalogs)
+        for entry in sources:
             try:
                 data = self._fetch_single_catalog(entry, force_refresh)
-            except WorkflowCatalogError:
+            except WorkflowCatalogError as exc:
+                if workflow_id is not None and isinstance(
+                    exc, WorkflowCatalogValidationError
+                ):
+                    raise
+                if (
+                    isinstance(exc, WorkflowCatalogValidationError)
+                    and validation_error is None
+                ):
+                    validation_error = exc
                 fetch_errors += 1
                 continue
             workflows = data.get("workflows", {})
             # Handle both dict and list formats
             if isinstance(workflows, dict):
                 for wf_id, wf_data in workflows.items():
+                    if workflow_id is not None and wf_id != workflow_id:
+                        continue
                     if not isinstance(wf_data, dict):
+                        if workflow_id is not None:
+                            raise WorkflowCatalogValidationError(
+                                f"Invalid workflow catalog entry for '{wf_id}' "
+                                f"from {entry.url}: expected a JSON object"
+                            )
                         continue
                     wf_data["_catalog_name"] = entry.name
                     wf_data["_install_allowed"] = entry.install_allowed
@@ -645,11 +699,17 @@ def _get_merged_workflows(
                         continue
                     wf_id = wf_data.get("id", "")
                     if wf_id:
+                        if workflow_id is not None and wf_id != workflow_id:
+                            continue
                         wf_data["_catalog_name"] = entry.name
                         wf_data["_install_allowed"] = entry.install_allowed
                         merged[wf_id] = wf_data
+            if workflow_id is not None and workflow_id in merged:
+                return merged
         if fetch_errors == len(catalogs) and catalogs:
-            raise WorkflowCatalogError(
+            if validation_error is not None:
+                raise validation_error
+            raise WorkflowCatalogFetchError(
                 "All configured catalogs failed to fetch."
             )
         return merged
@@ -698,7 +758,7 @@ def get_workflow_info(
         """Get the current or an exact advertised release from the winning source."""
         from ._versions import select_release
 
-        merged = self._get_merged_workflows()
+        merged = self._get_merged_workflows(workflow_id=workflow_id)
         wf = merged.get(workflow_id)
         if wf is None:
             return None
@@ -716,7 +776,7 @@ def get_workflow_version_details(
         """Return advertised versions and whether their source allows installation."""
         from ._versions import available_versions
 
-        merged = self._get_merged_workflows()
+        merged = self._get_merged_workflows(workflow_id=workflow_id)
         wf = merged.get(workflow_id)
         if wf is None:
             return None

src/specify_cli/workflows/command_add.py

@@ -32,19 +32,23 @@ def _workflow_package_has_companions(package_dir: cli.Path) -> bool:
 
 def _prepare_workflow_add(
     project_root: cli.Path, source: str, *, dev: bool, from_url: str | None,
-    version: str | None,
+    version: str | None, selected_catalog: bool = False,
 ) -> cli.Path:
     from . import load_custom_steps
 
     if version is not None and (
-        dev or from_url is not None or source.startswith(("http://", "https://"))
-        or cli.Path(source).exists()
+        dev or from_url is not None or (
+            not selected_catalog and (
+                source.startswith(("http://", "https://"))
+                or cli.Path(source).exists()
+            )
+        )
     ):
         cli.console.print(
             "[red]Error:[/red] --version requires a workflow ID from a catalog."
         )
         raise cli.typer.Exit(1)
-    if version is not None:
+    if version is not None or selected_catalog:
         cli._validate_workflow_id_or_exit(source)
     load_custom_steps(project_root)
     cli._open_workflow_registry(project_root)
@@ -62,10 +66,9 @@ def _install_preselected_workflow(
     """Install a bundle-selected release with the normal workflow preflights."""
     project_root = cli._require_specify_project()
     workflows_dir = _prepare_workflow_add(
-        project_root, workflow_id, dev=False, from_url=None, version=version,
+        project_root, workflow_id, dev=False, from_url=None,
+        version=version, selected_catalog=True,
     )
-    if version is None:
-        cli._validate_workflow_id_or_exit(workflow_id)
     cli._install_workflow_from_catalog(
         project_root, workflows_dir, workflow_id,
         requested_version=version, selected_info=selected_info,

src/specify_cli/workflows/step/catalog/__init__.py

@@ -29,6 +29,8 @@ def register(app: typer.Typer) -> None:
     "StepCatalog",
     "StepCatalogEntry",
     "StepCatalogError",
+    "StepCatalogFetchError",
+    "StepCatalogValidationError",
     "StepRegistry",
     "StepValidationError",
 }