fix: address PR review feedback from Copilot

- Defensive author field handling in WorkflowCatalog.search(): coerce
  non-string author values to empty string before comparison
- Make 'source' argument optional in workflow add when --from is used
- Fix type hint: workflow_id in update command is Optional[str]
- Add workflow ID mismatch check after download in workflow update
- Use temp dir for atomic update: download+validate in staging dir,
  then swap into place (no partial state on failure)
- WorkflowRegistry.update() now sets updated_at timestamp and
  preserves installed_at
- Add 9 CLI integration tests via CliRunner (enable/disable roundtrip,
  update error paths, add validation, list disabled status)

Author: mnriem

src/specify_cli/__init__.py

@@ -4855,14 +4855,19 @@ def workflow_list():
 
 @workflow_app.command("add")
 def workflow_add(
-    source: str = typer.Argument(..., help="Workflow ID, URL, or local path"),
+    source: Optional[str] = typer.Argument(None, help="Workflow ID, URL, or local path"),
     dev: bool = typer.Option(False, "--dev", help="Install from local directory"),
     from_url: Optional[str] = typer.Option(None, "--from", help="Install from custom URL"),
 ):
     """Install a workflow from catalog, URL, or local path."""
     from .workflows.catalog import WorkflowCatalog, WorkflowRegistry, WorkflowCatalogError
     from .workflows.engine import WorkflowDefinition
 
+    # Validate that at least one source is provided
+    if not source and not from_url:
+        console.print("[red]Error:[/red] Provide a workflow ID/URL/path, or use --from <url>")
+        raise typer.Exit(1)
+
     project_root = Path.cwd()
     specify_dir = project_root / ".specify"
     if not specify_dir.exists():
@@ -5304,13 +5309,14 @@ def workflow_info(
 
 @workflow_app.command("update")
 def workflow_update(
-    workflow_id: str = typer.Argument(None, help="Workflow ID to update (or all)"),
+    workflow_id: Optional[str] = typer.Argument(None, help="Workflow ID to update (or all)"),
 ):
     """Update workflow(s) to latest version."""
     from .workflows.catalog import WorkflowCatalog, WorkflowRegistry, WorkflowCatalogError
     from .workflows.engine import WorkflowDefinition, validate_workflow
     from packaging import version as pkg_version
     import shutil
+    import tempfile
 
     project_root = Path.cwd()
     specify_dir = project_root / ".specify"
@@ -5420,37 +5426,41 @@ def workflow_update(
             continue
 
         wf_dir = workflows_dir / wf_id
-        backup_dir = workflows_dir / ".backup" / wf_id
         try:
-            # Backup existing
-            if wf_dir.exists():
-                backup_dir.parent.mkdir(parents=True, exist_ok=True)
-                if backup_dir.exists():
-                    shutil.rmtree(backup_dir)
-                shutil.copytree(wf_dir, backup_dir)
-
-            # Download new version
-            wf_dir.mkdir(parents=True, exist_ok=True)
-            with urlopen(wf_url, 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):
-                    raise ValueError(f"Redirected to non-HTTPS: {final_url}")
-                wf_file = wf_dir / "workflow.yml"
-                wf_file.write_bytes(resp.read())
+            # Download and validate in a temporary directory so failures do not
+            # leave a partially-created workflow directory behind.
+            with tempfile.TemporaryDirectory(prefix=f"{wf_id}-", dir=workflows_dir) as tmp_dir:
+                staged_dir = Path(tmp_dir) / wf_id
+                staged_dir.mkdir(parents=True, exist_ok=True)
+                with urlopen(wf_url, 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):
+                        raise ValueError(f"Redirected to non-HTTPS: {final_url}")
+                    wf_file = staged_dir / "workflow.yml"
+                    wf_file.write_bytes(resp.read())
+
+                # Validate staged contents before replacing the live workflow
+                definition = WorkflowDefinition.from_yaml(wf_file)
+                if definition.id != wf_id:
+                    raise ValueError(
+                        f"Workflow ID mismatch: expected '{wf_id}', got '{definition.id}'"
+                    )
+                errors = validate_workflow(definition)
+                if errors:
+                    raise ValueError(f"Validation failed: {'; '.join(errors)}")
 
-            # Validate
-            definition = WorkflowDefinition.from_yaml(wf_file)
-            errors = validate_workflow(definition)
-            if errors:
-                raise ValueError(f"Validation failed: {'; '.join(errors)}")
+                # Swap: remove old, move staged into place
+                if wf_dir.exists():
+                    shutil.rmtree(wf_dir)
+                shutil.move(str(staged_dir), str(wf_dir))
 
             # Update registry
             registry.update(wf_id, {
@@ -5460,22 +5470,8 @@ def workflow_update(
             })
             console.print(f"  [green]✓[/green] {wf_id}: {update['installed']} → {update['available']}")
 
-            # Clean up backup
-            if backup_dir.exists():
-                shutil.rmtree(backup_dir)
-
         except Exception as exc:
             console.print(f"  [red]✗[/red] {wf_id}: {exc}")
-            # Restore from backup
-            if backup_dir.exists():
-                if wf_dir.exists():
-                    shutil.rmtree(wf_dir)
-                shutil.move(str(backup_dir), str(wf_dir))
-
-    # Clean up backup directory if empty
-    backup_parent = workflows_dir / ".backup"
-    if backup_parent.exists() and not any(backup_parent.iterdir()):
-        backup_parent.rmdir()
 
 
 @workflow_app.command("enable")

src/specify_cli/workflows/catalog.py

@@ -109,9 +109,18 @@ def remove(self, workflow_id: str) -> bool:
 
     def update(self, workflow_id: str, fields: dict[str, Any]) -> None:
         """Update specific fields on an installed workflow entry."""
+        from datetime import datetime, timezone
+
         if workflow_id not in self.data["workflows"]:
             return
-        self.data["workflows"][workflow_id].update(fields)
+
+        existing = self.data["workflows"][workflow_id]
+        installed_at = existing.get("installed_at")
+
+        existing.update(fields)
+        if installed_at is not None and "installed_at" not in fields:
+            existing["installed_at"] = installed_at
+        existing["updated_at"] = datetime.now(timezone.utc).isoformat()
         self.save()
 
     def get(self, workflow_id: str) -> dict[str, Any] | None:
@@ -428,7 +437,9 @@ def search(
         for wf_id, wf_data in merged.items():
             wf_data.setdefault("id", wf_id)
             if author:
-                if wf_data.get("author", "").lower() != author.lower():
+                raw_author = wf_data.get("author", "")
+                normalized_author = raw_author if isinstance(raw_author, str) else ""
+                if normalized_author.lower() != author.lower():
                     continue
             if query:
                 q = query.lower()

tests/test_workflows.py

@@ -1645,13 +1645,19 @@ def test_update(self, project_dir):
         registry = WorkflowRegistry(project_dir)
         registry.add("test-wf", {"name": "Test", "version": "1.0.0"})
 
+        original_entry = registry.get("test-wf")
+        original_installed_at = original_entry["installed_at"]
+
         registry.update("test-wf", {"version": "2.0.0", "enabled": False})
 
         entry = registry.get("test-wf")
         assert entry["version"] == "2.0.0"
         assert entry["enabled"] is False
         # Original fields preserved
         assert entry["name"] == "Test"
+        # installed_at preserved, updated_at set
+        assert entry["installed_at"] == original_installed_at
+        assert "updated_at" in entry
 
     def test_update_nonexistent(self, project_dir):
         from specify_cli.workflows.catalog import WorkflowRegistry
@@ -1914,3 +1920,127 @@ def test_switch_workflow(self, project_dir):
         assert state.status == RunStatus.COMPLETED
         assert "do-plan" in state.step_results
         assert "do-specify" not in state.step_results
+
+
+# ===== CLI Integration Tests =====
+
+
+class TestWorkflowCLI:
+    """CLI integration tests for workflow commands via CliRunner."""
+
+    def test_enable_not_installed(self, project_dir, monkeypatch):
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        runner = CliRunner()
+        monkeypatch.chdir(project_dir)
+        result = runner.invoke(app, ["workflow", "enable", "nonexistent"])
+        assert result.exit_code != 0
+        assert "not installed" in result.output
+
+    def test_disable_not_installed(self, project_dir, monkeypatch):
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        runner = CliRunner()
+        monkeypatch.chdir(project_dir)
+        result = runner.invoke(app, ["workflow", "disable", "nonexistent"])
+        assert result.exit_code != 0
+        assert "not installed" in result.output
+
+    def test_enable_disable_roundtrip(self, project_dir, monkeypatch):
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        runner = CliRunner()
+        monkeypatch.chdir(project_dir)
+
+        # Pre-install a workflow in the registry
+        registry = WorkflowRegistry(project_dir)
+        registry.add("test-wf", {"name": "Test", "version": "1.0.0"})
+
+        # Disable
+        result = runner.invoke(app, ["workflow", "disable", "test-wf"])
+        assert result.exit_code == 0
+        assert "disabled" in result.output
+
+        # Already disabled
+        result = runner.invoke(app, ["workflow", "disable", "test-wf"])
+        assert result.exit_code == 0
+        assert "already disabled" in result.output
+
+        # Enable
+        result = runner.invoke(app, ["workflow", "enable", "test-wf"])
+        assert result.exit_code == 0
+        assert "enabled" in result.output
+
+        # Already enabled
+        result = runner.invoke(app, ["workflow", "enable", "test-wf"])
+        assert result.exit_code == 0
+        assert "already enabled" in result.output
+
+    def test_update_no_project(self, tmp_path, monkeypatch):
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        runner = CliRunner()
+        monkeypatch.chdir(tmp_path)
+        result = runner.invoke(app, ["workflow", "update"])
+        assert result.exit_code != 0
+        assert "Not a spec-kit project" in result.output
+
+    def test_update_not_installed(self, project_dir, monkeypatch):
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        runner = CliRunner()
+        monkeypatch.chdir(project_dir)
+        result = runner.invoke(app, ["workflow", "update", "nonexistent"])
+        assert result.exit_code != 0
+        assert "not installed" in result.output
+
+    def test_add_no_source(self, project_dir, monkeypatch):
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        runner = CliRunner()
+        monkeypatch.chdir(project_dir)
+        result = runner.invoke(app, ["workflow", "add"])
+        assert result.exit_code != 0
+
+    def test_add_from_url_without_source(self, project_dir, monkeypatch):
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        runner = CliRunner()
+        monkeypatch.chdir(project_dir)
+        # --from with non-HTTPS should be rejected with URL scheme error, not missing source
+        result = runner.invoke(app, ["workflow", "add", "--from", "http://evil.example.com/wf.yml"])
+        assert result.exit_code != 0
+        assert "HTTPS" in result.output
+
+    def test_add_dev_missing_path(self, project_dir, monkeypatch):
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        runner = CliRunner()
+        monkeypatch.chdir(project_dir)
+        result = runner.invoke(app, ["workflow", "add", "--dev", "/nonexistent/path"])
+        assert result.exit_code != 0
+        assert "not found" in result.output.lower() or "Error" in result.output
+
+    def test_list_shows_disabled(self, project_dir, monkeypatch):
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        runner = CliRunner()
+        monkeypatch.chdir(project_dir)
+
+        registry = WorkflowRegistry(project_dir)
+        registry.add("test-wf", {"name": "Test WF", "version": "1.0.0", "enabled": False})
+
+        result = runner.invoke(app, ["workflow", "list"])
+        assert result.exit_code == 0
+        assert "disabled" in result.output