Address fifth review: expression parsing, fan-out output, URL install, gate options

- Move string literal parsing before operator detection in expressions
  so quoted strings with operators (e.g. 'a in b') are not mis-parsed
- Fan-out: remove max_concurrency from persisted output, fix docstring
  to reflect sequential execution
- workflow add: support URL sources with HTTPS/redirect validation,
  validate workflow ID is non-empty before writing files
- Deduplicate local install logic via _validate_and_install_local()
- Remove 'edit' gate option from speckit workflow (not implemented)

Author: mnriem

src/specify_cli/__init__.py

@@ -4294,53 +4294,87 @@ def workflow_add(
     registry = WorkflowRegistry(project_root)
     workflows_dir = project_root / ".specify" / "workflows"
 
+    def _validate_and_install_local(yaml_path: Path, source_label: str) -> None:
+        """Validate and install a workflow from a local YAML file."""
+        try:
+            definition = WorkflowDefinition.from_yaml(yaml_path)
+        except (ValueError, yaml.YAMLError) as exc:
+            console.print(f"[red]Error:[/red] Invalid workflow YAML: {exc}")
+            raise typer.Exit(1)
+        if not definition.id or not definition.id.strip():
+            console.print("[red]Error:[/red] Workflow definition has an empty or missing 'id'")
+            raise typer.Exit(1)
+        dest_dir = workflows_dir / definition.id
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        import shutil
+        shutil.copy2(yaml_path, dest_dir / "workflow.yml")
+        registry.add(definition.id, {
+            "name": definition.name,
+            "version": definition.version,
+            "description": definition.description,
+            "source": source_label,
+        })
+        console.print(f"[green]✓[/green] Workflow '{definition.name}' ({definition.id}) installed")
+
+    # Try as URL (http/https)
+    if source.startswith("http://") or source.startswith("https://"):
+        from ipaddress import ip_address
+        from urllib.parse import urlparse
+        from urllib.request import urlopen  # noqa: S310
+
+        parsed_src = urlparse(source)
+        src_host = parsed_src.hostname or ""
+        src_loopback = src_host == "localhost"
+        if not src_loopback:
+            try:
+                src_loopback = ip_address(src_host).is_loopback
+            except ValueError:
+                pass
+        if parsed_src.scheme != "https" and not (parsed_src.scheme == "http" and src_loopback):
+            console.print("[red]Error:[/red] Only HTTPS URLs are allowed, except HTTP for localhost.")
+            raise typer.Exit(1)
+
+        import tempfile
+        try:
+            with urlopen(source, timeout=30) as resp:  # noqa: S310
+                final_url = resp.geturl()
+                final_parsed = urlparse(final_url)
+                final_host = final_parsed.hostname or ""
+                final_lb = final_host == "localhost"
+                if not final_lb:
+                    try:
+                        final_lb = ip_address(final_host).is_loopback
+                    except ValueError:
+                        pass
+                if final_parsed.scheme != "https" and not (final_parsed.scheme == "http" and final_lb):
+                    console.print(f"[red]Error:[/red] URL redirected to non-HTTPS: {final_url}")
+                    raise typer.Exit(1)
+                with tempfile.NamedTemporaryFile(suffix=".yml", delete=False) as tmp:
+                    tmp.write(resp.read())
+                    tmp_path = Path(tmp.name)
+        except typer.Exit:
+            raise
+        except Exception as exc:
+            console.print(f"[red]Error:[/red] Failed to download workflow: {exc}")
+            raise typer.Exit(1)
+        try:
+            _validate_and_install_local(tmp_path, source)
+        finally:
+            tmp_path.unlink(missing_ok=True)
+        return
+
     # Try as a local file/directory
     source_path = Path(source)
     if source_path.exists():
         if source_path.is_file() and source_path.suffix in (".yml", ".yaml"):
-            # Install from local YAML file
-            try:
-                definition = WorkflowDefinition.from_yaml(source_path)
-            except (ValueError, yaml.YAMLError) as exc:
-                console.print(f"[red]Error:[/red] Invalid workflow YAML: {exc}")
-                raise typer.Exit(1)
-
-            dest_dir = workflows_dir / definition.id
-            dest_dir.mkdir(parents=True, exist_ok=True)
-            import shutil
-            shutil.copy2(source_path, dest_dir / "workflow.yml")
-
-            registry.add(definition.id, {
-                "name": definition.name,
-                "version": definition.version,
-                "description": definition.description,
-                "source": str(source_path),
-            })
-            console.print(f"[green]✓[/green] Workflow '{definition.name}' ({definition.id}) installed")
+            _validate_and_install_local(source_path, str(source_path))
             return
         elif source_path.is_dir():
             wf_file = source_path / "workflow.yml"
             if not wf_file.exists():
                 console.print(f"[red]Error:[/red] No workflow.yml found in {source}")
                 raise typer.Exit(1)
-            try:
-                definition = WorkflowDefinition.from_yaml(wf_file)
-            except (ValueError, yaml.YAMLError) as exc:
-                console.print(f"[red]Error:[/red] Invalid workflow YAML: {exc}")
-                raise typer.Exit(1)
-
-            dest_dir = workflows_dir / definition.id
-            dest_dir.mkdir(parents=True, exist_ok=True)
-            import shutil
-            shutil.copy2(wf_file, dest_dir / "workflow.yml")
-
-            registry.add(definition.id, {
-                "name": definition.name,
-                "version": definition.version,
-                "description": definition.description,
-                "source": str(source_path),
-            })
-            console.print(f"[green]✓[/green] Workflow '{definition.name}' ({definition.id}) installed")
+            _validate_and_install_local(wf_file, str(source_path))
             return
 
     # Try from catalog

src/specify_cli/workflows/engine.py

@@ -626,10 +626,13 @@ def _execute_steps(
                         ):
                             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
+                    # Fan-out items are executed sequentially in this engine,
+                    # so do not surface max_concurrency in persisted output.
+                    fan_out_output = dict(result.output)
+                    fan_out_output.pop("max_concurrency", None)
+                    fan_out_output["results"] = fan_out_results
+                    context.steps[step_id]["output"] = fan_out_output
+                    state.step_results[step_id]["output"] = fan_out_output
 
     def _resolve_inputs(
         self,

src/specify_cli/workflows/expressions.py

@@ -143,6 +143,13 @@ def _evaluate_simple_expression(expr: str, namespace: dict[str, Any]) -> Any:
             return _filter_default(value)
         return value
 
+    # String literal — check before operators so quoted strings
+    # containing operator keywords (e.g. 'a in b') are not mis-parsed.
+    if (expr.startswith("'") and expr.endswith("'")) or (
+        expr.startswith('"') and expr.endswith('"')
+    ):
+        return expr[1:-1]
+
     # Boolean operators
     if " and " in expr:
         parts = expr.split(" and ", 1)
@@ -183,12 +190,6 @@ def _evaluate_simple_expression(expr: str, namespace: dict[str, Any]) -> Any:
             if op == " not in ":
                 return left not in right if right is not None else True
 
-    # String literal
-    if (expr.startswith("'") and expr.endswith("'")) or (
-        expr.startswith('"') and expr.endswith('"')
-    ):
-        return expr[1:-1]
-
     # Numeric literal
     try:
         if "." in expr:

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

@@ -1,4 +1,4 @@
-"""Fan-out step — parallel dispatch over a collection."""
+"""Fan-out step — dispatch a step template over a collection."""
 
 from __future__ import annotations
 
@@ -9,10 +9,12 @@
 
 
 class FanOutStep(StepBase):
-    """Parallel dispatch over ``items:`` collection.
+    """Dispatch a step template for each item in a collection.
 
-    Iterates over items and dispatches the nested ``step:`` template
-    for each item, up to ``max_concurrency:`` at a time.
+    The engine executes the nested ``step:`` template once per item,
+    setting ``context.item`` for each iteration.  Execution is
+    currently sequential; ``max_concurrency`` is accepted but not
+    enforced.
     """
 
     type_key = "fan-out"

workflows/speckit/workflow.yml

@@ -31,7 +31,7 @@ steps:
   - id: review-spec
     type: gate
     message: "Review the generated spec before planning."
-    options: [approve, edit, reject]
+    options: [approve, reject]
     on_reject: abort
 
   - id: plan