Address third review: fan-out IDs, catalog guards, shell coercion, docs

- Fan-out generates unique per-item step IDs and collects results
- Catalog merge skips non-dict workflow entries (malformed data guard)
- Shell step coerces run_cmd to str after expression evaluation
- urlopen timeout=30 for catalog workflow installs
- yaml.dump with sort_keys=False, allow_unicode=True for catalog configs
- Document streaming timeout as intentionally unbounded (user Ctrl+C)
- Document --allow-all-tools as required for non-interactive + future enhancement
- Update test docstring and PUBLISHING.md to 10 step types with prompt

Author: mnriem

src/specify_cli/__init__.py

@@ -4394,7 +4394,7 @@ def workflow_add(
         from urllib.request import urlopen  # noqa: S310 — URL comes from catalog
 
         workflow_dir.mkdir(parents=True, exist_ok=True)
-        with urlopen(workflow_url) as response:  # noqa: S310
+        with urlopen(workflow_url, timeout=30) as response:  # noqa: S310
             workflow_file.write_bytes(response.read())
     except Exception as exc:
         if workflow_dir.exists():

src/specify_cli/integrations/base.py

@@ -173,6 +173,9 @@ def dispatch_command(
         cwd = str(project_root) if project_root else None
 
         if stream:
+            # No timeout when streaming — the user sees live output and
+            # can Ctrl+C at any time.  The timeout parameter is only
+            # applied in the captured (non-streaming) branch below.
             try:
                 result = subprocess.run(
                     exec_args,

src/specify_cli/integrations/copilot/__init__.py

@@ -83,6 +83,11 @@ def dispatch_command(
         agent_name = f"speckit.{stem}"
 
         prompt = args or ""
+        # NOTE: --allow-all-tools is required for non-interactive execution
+        # so the agent can perform file edits and shell commands.  The
+        # workflow engine's gate steps serve as the human approval mechanism.
+        # Making tool approval configurable (e.g. via workflow config or
+        # env var) is tracked as a future enhancement.
         cli_args = [
             "copilot", "-p", prompt,
             "--agent", agent_name,

src/specify_cli/workflows/catalog.py

@@ -371,11 +371,15 @@ def _get_merged_workflows(
             # Handle both dict and list formats
             if isinstance(workflows, dict):
                 for wf_id, wf_data in workflows.items():
+                    if not isinstance(wf_data, dict):
+                        continue
                     wf_data["_catalog_name"] = entry.name
                     wf_data["_install_allowed"] = entry.install_allowed
                     merged[wf_id] = wf_data
             elif isinstance(workflows, list):
                 for wf_data in workflows:
+                    if not isinstance(wf_data, dict):
+                        continue
                     wf_id = wf_data.get("id", "")
                     if wf_id:
                         wf_data["_catalog_name"] = entry.name
@@ -468,7 +472,7 @@ def add_catalog(self, url: str, name: str | None = None) -> None:
 
         config_path.parent.mkdir(parents=True, exist_ok=True)
         with open(config_path, "w", encoding="utf-8") as f:
-            yaml.dump(data, f, default_flow_style=False)
+            yaml.dump(data, f, default_flow_style=False, sort_keys=False, allow_unicode=True)
 
     def remove_catalog(self, index: int) -> str:
         """Remove a catalog source by index (0-based). Returns the removed name."""
@@ -488,6 +492,6 @@ def remove_catalog(self, index: int) -> str:
         data["catalogs"] = catalogs
 
         with open(config_path, "w", encoding="utf-8") as f:
-            yaml.dump(data, f, default_flow_style=False)
+            yaml.dump(data, f, default_flow_style=False, sort_keys=False, allow_unicode=True)
 
         return removed.get("name", f"catalog-{index + 1}")

src/specify_cli/workflows/engine.py

@@ -596,23 +596,35 @@ def _execute_steps(
                         ):
                             return
 
-            # Fan-out: execute nested step template per item
+            # Fan-out: execute nested step template per item with unique IDs
             if step_type == "fan-out" and result.output.get("items"):
                 template = result.output.get("step_template", {})
                 if template:
-                    for item_val in result.output["items"]:
+                    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
+                        item_step = dict(template)
+                        base_id = item_step.get("id", "item")
+                        item_step["id"] = f"{base_id}-{item_idx}"
                         self._execute_steps(
-                            [template], context, state, registry,
+                            [item_step], context, state, registry,
                             step_offset=-1,
                         )
+                        # Collect per-item result for fan-in
+                        item_result = context.steps.get(item_step["id"], {})
+                        fan_out_results.append(item_result.get("output", {}))
                         if state.status in (
                             RunStatus.PAUSED,
                             RunStatus.FAILED,
                             RunStatus.ABORTED,
                         ):
-                            return
+                            break
                     context.item = None
+                    # Update fan-out step output with collected results
+                    result.output["results"] = fan_out_results
+                    context.steps[step_id]["output"] = result.output
+                    state.step_results[step_id]["output"] = result.output
 
     def _resolve_inputs(
         self,

src/specify_cli/workflows/steps/shell/__init__.py

@@ -21,6 +21,7 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
         run_cmd = config.get("run", "")
         if isinstance(run_cmd, str) and "{{" in run_cmd:
             run_cmd = evaluate_expression(run_cmd, context)
+        run_cmd = str(run_cmd)
 
         cwd = context.project_root or "."
 

tests/test_workflows.py

@@ -4,7 +4,7 @@
 - Step registry & auto-discovery
 - Base classes (StepBase, StepContext, StepResult)
 - Expression engine
-- All 9 built-in step types
+- All 10 built-in step types
 - Workflow definition loading & validation
 - Workflow engine execution & state persistence
 - Workflow catalog & registry

workflows/PUBLISHING.md

@@ -90,7 +90,7 @@ steps:
 - ✅ `version` follows semantic versioning (X.Y.Z)
 - ✅ `description` is concise
 - ✅ All step IDs are unique
-- ✅ Step types are valid: `command`, `shell`, `gate`, `if`, `switch`, `while`, `do-while`, `fan-out`, `fan-in`
+- ✅ Step types are valid: `command`, `prompt`, `shell`, `gate`, `if`, `switch`, `while`, `do-while`, `fan-out`, `fan-in`
 - ✅ Required fields present per step type (e.g., `condition` for `if`, `expression` for `switch`)
 - ✅ Input types are valid: `string`, `number`, `boolean`
 - ✅ Enum values match declared type