Address eighth review: fan-out namespaced IDs, early return, catalog validation

- Fan-out per-item step IDs use _fanout_{step_id}_{base}_{idx} namespace
  to avoid collisions with user-defined step IDs
- Early return after fan-out loop when state is paused/failed/aborted
- Catalog installs parse + validate downloaded YAML before registering,
  using definition metadata instead of catalog entry for registry

Author: mnriem

src/specify_cli/__init__.py

@@ -4469,10 +4469,29 @@ def _validate_and_install_local(yaml_path: Path, source_label: str) -> None:
         console.print(f"[red]Error:[/red] Failed to install workflow '{source}' from catalog: {exc}")
         raise typer.Exit(1)
 
+    # Validate the downloaded workflow before registering
+    try:
+        definition = WorkflowDefinition.from_yaml(workflow_file)
+    except (ValueError, yaml.YAMLError) as exc:
+        import shutil
+        shutil.rmtree(workflow_dir, ignore_errors=True)
+        console.print(f"[red]Error:[/red] Downloaded workflow is invalid: {exc}")
+        raise typer.Exit(1)
+
+    from .workflows.engine import validate_workflow
+    errors = validate_workflow(definition)
+    if errors:
+        import shutil
+        shutil.rmtree(workflow_dir, ignore_errors=True)
+        console.print("[red]Error:[/red] Downloaded workflow validation failed:")
+        for err in errors:
+            console.print(f"  \u2022 {err}")
+        raise typer.Exit(1)
+
     registry.add(source, {
-        "name": info.get("name", source),
-        "version": info.get("version", "0.0.0"),
-        "description": info.get("description", ""),
+        "name": definition.name or info.get("name", source),
+        "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,

src/specify_cli/workflows/engine.py

@@ -610,10 +610,10 @@ def _execute_steps(
                     fan_out_results = []
                     for item_idx, item_val in enumerate(result.output["items"]):
                         context.item = item_val
-                        # Create a per-item copy with a unique ID
+                        # Create a per-item copy with an ID in a reserved namespace
                         item_step = dict(template)
                         base_id = item_step.get("id", "item")
-                        item_step["id"] = f"{base_id}-{item_idx}"
+                        item_step["id"] = f"_fanout_{step_id}_{base_id}_{item_idx}"
                         self._execute_steps(
                             [item_step], context, state, registry,
                             step_offset=-1,
@@ -635,6 +635,12 @@ def _execute_steps(
                     fan_out_output["results"] = fan_out_results
                     context.steps[step_id]["output"] = fan_out_output
                     state.step_results[step_id]["output"] = fan_out_output
+                    if state.status in (
+                        RunStatus.PAUSED,
+                        RunStatus.FAILED,
+                        RunStatus.ABORTED,
+                    ):
+                        return
 
     def _resolve_inputs(
         self,