Fix 4 current Copilot review findings on workflow run/registry/install

1. workflow run ownership check followed symlinks via Path.resolve()
   before mapping a direct YAML path back to its installed workflow ID.
   A symlinked .specify/workflows/<id>/workflow.yml resolved outside the
   tree, missed the ownership match entirely, and let the disabled-workflow
   guard be silently skipped while engine.load_workflow still followed the
   symlink. Now maps ownership from a lexically-normalized path (os.path.
   normpath, no symlink following) and explicitly refuses to run if the
   installed <id> directory or workflow.yml leaf is itself a symlink.
   Direct external workflow paths that don't match .specify/workflows/...
   are unaffected.

2. WorkflowRegistry._load() caught a read OSError and silently fell back
   to an empty in-memory registry, only blocking a later save(). Callers
   that only query is_installed()/get()/list() before writing a file
   (e.g. commands/init.py's bundled speckit install, which overwrites
   workflow.yml once is_installed() reports false) could act on that
   false-empty state and destroy real data before ever reaching save().
   _load() now raises OSError immediately so an unreadable registry fails
   closed at construction, before any query or side effect is possible.
   Added _open_workflow_registry() to give every CLI command a consistent
   clean-error boundary around registry construction.

3. _validate_and_install_local's mkdir/copy2 ran before the try/except
   that protected registry.add(); a copy2 failure (e.g. a truncating
   partial write on a reinstall) was not caught at all, so the existing
   backup-restore cleanup never ran and the prior working workflow.yml
   was corrupted with a raw traceback surfaced to the user. mkdir/copy2
   now run inside the same rollback-protected section as registry.add(),
   sharing one _cleanup_failed_install() helper.

4. workflow update's skip message claimed any non-catalog source was
   installed "from a local path or URL", which is wrong for the bundled
   speckit workflow (source: "bundled"). Message is now source-neutral.

Verified all 4 threads are current (not outdated) via GraphQL review
thread query on PR #3419, HEAD 812050a.

Tests: strict TDD per fix (red test proving each bug, minimal production
change, green). tests/test_workflows.py: 474 passed. Full suite: 3976
passed, 110 skipped. ruff check: all checks passed on touched files and
full src tree.

Assisted-by: GitHub Copilot (model: Claude Sonnet 5, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

src/specify_cli/workflows/_commands.py

@@ -58,6 +58,25 @@ def _error_console(json_output: bool):
     return err_console if json_output else console
 
 
+def _open_workflow_registry(project_root: Path, out=None):
+    """Construct a WorkflowRegistry, exiting cleanly on an unreadable file.
+
+    WorkflowRegistry fails closed (raises OSError) at construction when its
+    file can't be read, rather than falling back to an empty registry a
+    caller could mistake for "nothing installed". Every CLI command that
+    opens a registry needs this same clean-error boundary.
+    """
+    from .catalog import WorkflowRegistry
+
+    try:
+        return WorkflowRegistry(project_root)
+    except OSError as exc:
+        (out or console).print(
+            f"[red]Error:[/red] Failed to read workflow registry: {_escape_markup(str(exc))}"
+        )
+        raise typer.Exit(1)
+
+
 def _parse_input_values(
     input_values: list[str] | None, *, json_output: bool = False
 ) -> dict[str, Any]:
@@ -369,8 +388,6 @@ def workflow_run(
 
     err = _error_console(json_output)
 
-    from .catalog import WorkflowRegistry
-
     registered_id: str | None = None
     registry_root = project_root
     if not is_file_source:
@@ -385,18 +402,35 @@ def workflow_run(
         registered_id = source
     else:
         # A direct YAML path may still point at an installed workflow's own
-        # file; map it back to its owning project and ID from the canonical
-        # path itself so the guard is independent of the caller's cwd.
-        resolved = source_path.resolve()
-        parts = resolved.parts
+        # file; map it back to its owning project and ID from the *lexical*
+        # path (collapsing .. / . without resolving symlinks) rather than
+        # resolve(): resolving first would follow a symlinked workflow.yml
+        # out of .specify/workflows, fail to find an owner, and let
+        # engine.load_workflow below run the symlink target unchecked --
+        # silently bypassing a disabled workflow's guard.
+        lexical = Path(os.path.normpath(str(source_path.absolute())))
+        parts = lexical.parts
         for i in range(len(parts) - 2):
             if parts[i] == ".specify" and parts[i + 1] == "workflows":
-                registry_root = Path(*parts[:i]) if i else Path(resolved.anchor or ".")
+                registry_root = Path(*parts[:i]) if i else Path(lexical.anchor or ".")
                 registered_id = parts[i + 2]
+                # A legitimately installed workflow's own directory tree
+                # never contains a symlink (workflow add/remove both refuse
+                # one at install time); one appearing here means the file
+                # actually loaded below would not be the file this ownership
+                # match is based on, so refuse rather than silently mismatch.
+                for k in range(i + 2, len(parts) + 1):
+                    if Path(*parts[:k]).is_symlink():
+                        err.print(
+                            "[red]Error:[/red] Refusing to run: "
+                            f".specify/workflows/{_escape_markup(registered_id)} "
+                            "contains a symlinked path component"
+                        )
+                        raise typer.Exit(1)
                 break
 
     if registered_id is not None:
-        installed_meta = WorkflowRegistry(registry_root).get(registered_id)
+        installed_meta = _open_workflow_registry(registry_root, err).get(registered_id)
         if isinstance(installed_meta, dict) and not installed_meta.get("enabled", True):
             err.print(
                 f"[red]Error:[/red] Workflow '{_escape_markup(registered_id)}' is disabled. "
@@ -612,10 +646,8 @@ def workflow_status(
 @workflow_app.command("list")
 def workflow_list():
     """List installed workflows."""
-    from .catalog import WorkflowRegistry
-
     project_root = _require_specify_project()
-    registry = WorkflowRegistry(project_root)
+    registry = _open_workflow_registry(project_root)
     installed = registry.list()
 
     if not installed:
@@ -647,11 +679,10 @@ def workflow_add(
     from_url: str | None = typer.Option(None, "--from", help="Install from a custom URL"),
 ):
     """Install a workflow from catalog, URL, or local path."""
-    from .catalog import WorkflowRegistry
     from .engine import WorkflowDefinition
 
     project_root = _require_specify_project()
-    registry = WorkflowRegistry(project_root)
+    registry = _open_workflow_registry(project_root)
     workflows_dir = project_root / ".specify" / "workflows"
     # With --from, source names the expected workflow ID: validate it up
     # front so a URL/path/typo fails without a network fetch.
@@ -705,17 +736,9 @@ def workflow_add(
         backup_bytes = (
             dest_file.read_bytes() if existed_before and dest_file.is_file() else None
         )
-        dest_dir.mkdir(parents=True, exist_ok=True)
         import shutil
-        shutil.copy2(yaml_path, dest_file)
-        try:
-            registry.add(definition.id, {
-                "name": definition.name,
-                "version": definition.version,
-                "description": definition.description,
-                "source": source_label,
-            })
-        except OSError as exc:
+
+        def _cleanup_failed_install() -> None:
             # Don't leave an orphan directory behind for a fresh install; for
             # a reinstall over an existing local workflow, restore the prior
             # workflow.yml instead of clobbering it with the failed update.
@@ -724,6 +747,26 @@ def workflow_add(
                     dest_file.write_bytes(backup_bytes)
             else:
                 shutil.rmtree(dest_dir, ignore_errors=True)
+
+        try:
+            dest_dir.mkdir(parents=True, exist_ok=True)
+            shutil.copy2(yaml_path, dest_file)
+        except OSError as exc:
+            _cleanup_failed_install()
+            console.print(
+                f"[red]Error:[/red] Failed to install workflow "
+                f"'{_escape_markup(definition.id)}': {_escape_markup(str(exc))}"
+            )
+            raise typer.Exit(1)
+        try:
+            registry.add(definition.id, {
+                "name": definition.name,
+                "version": definition.version,
+                "description": definition.description,
+                "source": source_label,
+            })
+        except OSError as exc:
+            _cleanup_failed_install()
             console.print(
                 f"[red]Error:[/red] Failed to update workflow registry for "
                 f"'{_escape_markup(definition.id)}': {_escape_markup(str(exc))}"
@@ -1080,13 +1123,11 @@ def workflow_remove(
     workflow_id: str = typer.Argument(..., help="Workflow ID to uninstall"),
 ):
     """Uninstall a workflow."""
-    from .catalog import WorkflowRegistry
-
     project_root = _require_specify_project()
     workflows_dir = project_root / ".specify" / "workflows"
     _validate_workflow_id_or_exit(workflow_id)
 
-    registry = WorkflowRegistry(project_root)
+    registry = _open_workflow_registry(project_root)
 
     if not registry.is_installed(workflow_id):
         console.print(f"[red]Error:[/red] Workflow '{workflow_id}' is not installed")
@@ -1172,10 +1213,10 @@ def workflow_update(
     """Update installed workflow(s) to the latest catalog version."""
     from packaging import version as pkg_version
 
-    from .catalog import WorkflowCatalog, WorkflowCatalogError, WorkflowRegistry
+    from .catalog import WorkflowCatalog, WorkflowCatalogError
 
     project_root = _require_specify_project()
-    registry = WorkflowRegistry(project_root)
+    registry = _open_workflow_registry(project_root)
     workflows_dir = project_root / ".specify" / "workflows"
     _reject_unsafe_dir(project_root / ".specify", ".specify")
     _reject_unsafe_dir(workflows_dir, ".specify/workflows")
@@ -1205,7 +1246,7 @@ def workflow_update(
             console.print(f"⚠  {safe_id}: Registry entry is corrupted (skipping)")
             continue
         if metadata.get("source") != "catalog":
-            console.print(f"⚠  {safe_id}: Installed from a local path or URL — re-add to update (skipping)")
+            console.print(f"⚠  {safe_id}: Not installed from a catalog — re-add to update (skipping)")
             continue
         try:
             installed_version = pkg_version.Version(str(metadata.get("version")))
@@ -1310,10 +1351,8 @@ def workflow_enable(
     workflow_id: str = typer.Argument(..., help="Workflow ID to enable"),
 ):
     """Enable a disabled workflow."""
-    from .catalog import WorkflowRegistry
-
     project_root = _require_specify_project()
-    registry = WorkflowRegistry(project_root)
+    registry = _open_workflow_registry(project_root)
     metadata = registry.get(workflow_id)
     if metadata is None:
         console.print(f"[red]Error:[/red] Workflow '{_escape_markup(workflow_id)}' is not installed")
@@ -1344,10 +1383,8 @@ def workflow_disable(
     workflow_id: str = typer.Argument(..., help="Workflow ID to disable"),
 ):
     """Disable a workflow without removing it."""
-    from .catalog import WorkflowRegistry
-
     project_root = _require_specify_project()
-    registry = WorkflowRegistry(project_root)
+    registry = _open_workflow_registry(project_root)
     metadata = registry.get(workflow_id)
     if metadata is None:
         console.print(f"[red]Error:[/red] Workflow '{_escape_markup(workflow_id)}' is not installed")
@@ -1416,13 +1453,13 @@ def workflow_info(
     workflow_id: str = typer.Argument(..., help="Workflow ID"),
 ):
     """Show workflow details and step graph."""
-    from .catalog import WorkflowCatalog, WorkflowRegistry, WorkflowCatalogError
+    from .catalog import WorkflowCatalog, WorkflowCatalogError
     from .engine import WorkflowEngine
 
     project_root = _require_specify_project()
 
     # Check installed first
-    registry = WorkflowRegistry(project_root)
+    registry = _open_workflow_registry(project_root)
     installed = registry.get(workflow_id)
 
     engine = WorkflowEngine(project_root)

src/specify_cli/workflows/catalog.py

@@ -70,10 +70,6 @@ class WorkflowRegistry:
         self.project_root = project_root
         self.workflows_dir = project_root / ".specify" / "workflows"
         self.registry_path = self.workflows_dir / self.REGISTRY_FILE
-        # Set before _load() so a read failure (distinct from a corrupted or
-        # missing file) can flip it to block a later save() from silently
-        # persisting an empty registry over unreadable-but-intact data.
-        self._load_error = False
         self.data = self._load()
 
     def _has_symlinked_parent(self) -> bool:
@@ -99,14 +95,17 @@ class WorkflowRegistry:
             try:
                 with open(self.registry_path, encoding="utf-8") as f:
                     data = json.load(f)
-            except OSError:
+            except OSError as exc:
                 # An I/O failure (e.g. permissions, transient FS issue) is not
                 # the same as a corrupted file: the real data may still be
-                # intact on disk. Flag it so save() refuses to overwrite that
-                # data with this in-memory default instead of silently
-                # discarding every prior entry.
-                self._load_error = True
-                return default_registry
+                # intact on disk. Fail closed here, at construction, rather
+                # than falling back to an empty registry -- a caller that
+                # only queries is_installed()/get()/list() before writing a
+                # file (never reaching save()) would otherwise mistake this
+                # for "nothing installed" and overwrite real data.
+                raise OSError(
+                    f"Failed to read workflow registry at {self.registry_path}: {exc}"
+                ) from exc
             except (json.JSONDecodeError, ValueError, UnicodeError):
                 # Corrupted registry file — reset to default
                 return default_registry
@@ -126,12 +125,6 @@ class WorkflowRegistry:
             raise OSError(
                 "Refusing to write workflow registry through a symlinked path."
             )
-        if self._load_error:
-            raise OSError(
-                f"Refusing to save workflow registry at {self.registry_path}: "
-                "the existing file could not be read, so saving now would "
-                "discard its contents."
-            )
         self.workflows_dir.mkdir(parents=True, exist_ok=True)
         # Unique, exclusive temp then replace: a failed dump cannot truncate
         # the registry, a pre-created symlink cannot redirect the write, and

tests/test_workflows.py

@@ -4771,8 +4771,10 @@ class TestWorkflowRegistry:
 
     def test_load_read_oserror_refuses_to_save_over_existing_data(self, project_dir, monkeypatch):
         """A transient read failure (e.g. temporarily unreadable file) must not be
-        treated the same as a corrupted/missing registry: saving afterwards would
-        silently discard every previously persisted workflow entry."""
+        treated the same as a corrupted/missing registry: constructing a registry
+        on top of it -- and thus any query a caller makes before ever calling
+        save() -- must fail closed instead of silently reporting an empty
+        registry that a caller could then act on and overwrite."""
         from specify_cli.workflows.catalog import WorkflowRegistry
         import builtins
 
@@ -4787,15 +4789,37 @@ class TestWorkflowRegistry:
             return real_open(file, mode, *args, **kwargs)
 
         monkeypatch.setattr(builtins, "open", _raising_open)
-        registry2 = WorkflowRegistry(project_dir)
-        # The in-memory view may fall back to empty, but a save must not be
-        # allowed to persist that empty state over the real file on disk.
         with pytest.raises(OSError):
-            registry2.save()
+            WorkflowRegistry(project_dir)
         # The original entry must survive on disk untouched.
         data = json.loads(registry_path.read_text(encoding="utf-8"))
         assert "test-wf" in data["workflows"]
 
+    def test_load_read_oserror_fails_closed_not_silently_empty(self, project_dir, monkeypatch):
+        """Root cause: a registry that failed to read must never let a query
+        method (is_installed/get/list) report as if nothing were installed --
+        a caller (e.g. bundled workflow install) that only checks
+        is_installed() before writing a file would otherwise overwrite real
+        data on a transient read failure, long before any save() call could
+        catch it. The failure must surface at construction, before any query
+        or side effect is possible."""
+        from specify_cli.workflows.catalog import WorkflowRegistry
+        import builtins
+
+        registry1 = WorkflowRegistry(project_dir)
+        registry1.add("test-wf", {"name": "Test", "version": "1.0.0"})
+        registry_path = registry1.registry_path
+        real_open = builtins.open
+
+        def _raising_open(file, mode="r", *args, **kwargs):
+            if Path(file) == registry_path and "r" in mode:
+                raise OSError("simulated read failure")
+            return real_open(file, mode, *args, **kwargs)
+
+        monkeypatch.setattr(builtins, "open", _raising_open)
+        with pytest.raises(OSError):
+            WorkflowRegistry(project_dir)
+
 
 # ===== Workflow Catalog Tests =====
 
@@ -7676,6 +7700,26 @@ steps:
         assert "No workflows were eligible for update" in result.output
         assert "up to date!" not in result.output
 
+    def test_update_skip_message_accurate_for_bundled_source(self, project_dir, monkeypatch):
+        """A workflow registered with source "bundled" (e.g. the speckit
+        workflow installed by `specify init`) was never installed from a
+        local path or URL; the skip message must not claim otherwise."""
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        monkeypatch.chdir(project_dir)
+        registry = WorkflowRegistry(project_dir)
+        registry.add(
+            "speckit",
+            {"name": "Speckit", "version": "1.0.0", "source": "bundled"},
+        )
+        runner = CliRunner()
+        result = runner.invoke(app, ["workflow", "update"])
+        assert result.exit_code == 0, result.output
+        assert "local path or URL" not in result.output
+        assert "re-add to update" in result.output
+
     def test_registry_add_rolls_back_memory_on_save_failure(self, project_dir, monkeypatch):
         from specify_cli.workflows.catalog import WorkflowRegistry
 
@@ -7782,6 +7826,46 @@ steps:
         assert not dest_dir.exists()
         assert not WorkflowRegistry(project_dir).is_installed("align-wf")
 
+    def test_add_dev_reinstall_copy_failure_restores_prior_file(self, project_dir, monkeypatch):
+        """_validate_and_install_local's copy2 call currently runs *before* the
+        try/except block that protects registry.add(): a copy2 failure (e.g. a
+        truncating partial write on a reinstall) is not caught at all, so the
+        existing backup-restore cleanup never runs and the prior working
+        workflow.yml is left corrupted. copy2 must be covered by the same
+        rollback-protected section as registry.add()."""
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        monkeypatch.chdir(project_dir)
+        runner = CliRunner()
+        src = self._install_dev(runner, app, project_dir)
+        installed_yaml = project_dir / ".specify" / "workflows" / "align-wf" / "workflow.yml"
+        original_bytes = installed_yaml.read_bytes()
+        original_registry_entry = WorkflowRegistry(project_dir).get("align-wf")
+
+        # Point --dev at a new version of the same workflow to trigger a
+        # reinstall (overwrite) rather than a fresh install.
+        (src / "workflow.yml").write_text(
+            self.WORKFLOW_YAML.format(version="2.0.0"), encoding="utf-8"
+        )
+
+        def boom(*args, **kwargs):
+            # Simulate a truncating partial write followed by an OSError,
+            # mirroring a real disk-full/interrupted-copy failure.
+            installed_yaml.write_bytes(b"")
+            raise OSError("disk full")
+
+        with pytest.MonkeyPatch.context() as mp:
+            mp.setattr("shutil.copy2", boom)
+            result = runner.invoke(app, ["workflow", "add", str(src), "--dev"])
+
+        assert result.exit_code != 0
+        assert result.exception is None or isinstance(result.exception, SystemExit)
+        assert result.output.strip() != ""
+        assert installed_yaml.read_bytes() == original_bytes
+        assert WorkflowRegistry(project_dir).get("align-wf") == original_registry_entry
+
     def test_add_catalog_save_failure_leaves_no_orphan_directory(self, project_dir, monkeypatch):
         """Same guarantee as the local-install paths, but for a fresh catalog
         install: a registry.add() failure must clean up the freshly-downloaded
@@ -8435,6 +8519,37 @@ steps:
         assert "corrupted" in result.output
         assert "OK Workflow" in result.output
 
+    def test_list_unreadable_registry_fails_closed_with_clean_error(
+        self, project_dir, monkeypatch
+    ):
+        """An unreadable registry file must produce a clean CLI error, not a
+        raw traceback and not a silent "nothing installed" list -- the latter
+        is exactly the fail-open state a caller could otherwise mistake for
+        "safe to (re)install", overwriting real files. Covers the read/query
+        boundary fix required at every WorkflowRegistry call site."""
+        import builtins
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        monkeypatch.chdir(project_dir)
+        runner = CliRunner()
+        self._install_dev(runner, app, project_dir)
+
+        registry_path = WorkflowRegistry(project_dir).registry_path.resolve()
+        real_open = builtins.open
+
+        def _raising_open(file, mode="r", *args, **kwargs):
+            if Path(file).resolve() == registry_path and "r" in mode:
+                raise OSError("simulated read failure")
+            return real_open(file, mode, *args, **kwargs)
+
+        monkeypatch.setattr(builtins, "open", _raising_open)
+        result = runner.invoke(app, ["workflow", "list"])
+        assert result.exit_code != 0
+        assert result.exception is None or isinstance(result.exception, SystemExit)
+        assert "Error" in result.output
+
     def test_list_escapes_rich_markup_in_registry_fields(self, project_dir, monkeypatch):
         """User-editable name/description/id fields must not be parsed as Rich markup."""
         import json
@@ -8702,6 +8817,40 @@ steps:
         assert result.exit_code != 0
         assert "disabled" in result.output
 
+    @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
+    def test_disable_blocks_run_when_installed_yaml_is_symlinked(
+        self, project_dir, monkeypatch
+    ):
+        """A disabled workflow's own workflow.yml being replaced with a symlink
+        must not bypass the disabled check. Resolving the path before mapping
+        it back to its registry owner would follow the symlink out of
+        .specify/workflows, fail to find an owner, and let engine.load_workflow
+        run the original symlink target anyway -- ownership must be
+        determined from the normalized *lexical* path (not resolve()), and a
+        symlinked path component in the installed tree must be refused."""
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        monkeypatch.chdir(project_dir)
+        runner = CliRunner()
+        self._install_dev(runner, app, project_dir)
+
+        result = runner.invoke(app, ["workflow", "disable", "align-wf"])
+        assert result.exit_code == 0, result.output
+
+        installed_yaml = project_dir / ".specify" / "workflows" / "align-wf" / "workflow.yml"
+        external_target = project_dir / "external-workflow.yml"
+        external_target.write_text(
+            self.WORKFLOW_YAML.format(version="9.9.9"), encoding="utf-8"
+        )
+        installed_yaml.unlink()
+        installed_yaml.symlink_to(external_target)
+
+        result = runner.invoke(app, ["workflow", "run", str(installed_yaml)])
+        assert result.exit_code != 0
+        assert result.exception is None or isinstance(result.exception, SystemExit)
+        assert "disabled" in result.output or "symlink" in result.output.lower()
+
     def test_disable_shows_marker_in_list(self, project_dir, monkeypatch):
         from typer.testing import CliRunner
         from specify_cli import app