From 00465c3645f3fcd9f2840b4e677be41be79cd6ab Mon Sep 17 00:00:00 2001
From: marcelsafin <179933638+marcelsafin@users.noreply.github.com>
Date: Sat, 11 Jul 2026 12:44:01 +0200
Subject: [PATCH] Harden workflow install/remove transactions with atomic
 staging

Addresses 5 Copilot review findings on HEAD b8269c8, all centered on
transaction integrity around workflow install/remove/registry writes,
following the atomic_write_json pattern already used in _utils.py:

1. WorkflowRegistry.save() now preserves the existing registry file's
   mode (e.g. 0640/0644) across a save instead of silently downgrading
   it to mkstemp's 0600 default; a brand-new registry still gets the
   secure 0600 default.

2. workflow_remove now stages the install directory out of the way via
   an atomic rename *before* the registry write, rather than deleting
   it directly with shutil.rmtree after the registry already claims it
   removed. This closes a real data-integrity gap: a partially-failed
   rmtree could no longer leave a damaged directory re-marked
   "installed" by the old manual restore-after-rmtree-failure code
   (now deleted -- it's structurally impossible to need it). A
   registry-write failure renames the staged directory back
   (guarded, with an explicit warning if the restore-back rename
   itself fails); a registry-write success is durable, so a later
   failure to delete the staged directory is now a warning (exit 0),
   not a contradictory "Error: Failed to remove" (exit 1) that used to
   claim failure while the registry already recorded success.

3. Local (--dev/--from/plain path) and catalog install/reinstall now
   write new content to a same-directory staging file and commit it
   onto the destination workflow.yml via a single atomic swap, instead
   of writing/downloading directly into the destination file. A prior
   file (reinstall) is renamed aside rather than overwritten in place,
   so it can be restored via rename -- never a content rewrite -- if
   registry.add() subsequently fails; a rollback failure is now
   explicitly reported as a warning instead of escaping unguarded and
   masking the original clean error. This also removes the need to
   read the prior file's bytes into memory before installing (that
   read-before-write step and its failure mode are now unreachable),
   and both local and catalog installs share the same four small
   helpers (_stage_workflow_file / _commit_workflow_file /
   _discard_staged_workflow_file / _rollback_committed_workflow_file,
   plus guarded wrappers) rather than duplicating the logic.

4. Updated a stale comment (workflow_run's ownership-guard rationale)
   that still described WorkflowRegistry._load() as silently
   substituting an empty registry; it now fails closed by raising
   OSError, which the comment now states plainly.

Tests: rewrote the two workflow_remove tests whose assertions encoded
the old (incoherent) rmtree-then-restore contract to instead prove the
new stage-then-commit contract (post-registry-success cleanup failure
is a warning+exit 0; pre-registry-success stage-restore failure is
guarded and escapes markup correctly). Rewrote the local/catalog
"backup read failure" tests, which tested a step the new design no
longer performs, into "restore-rename failure" tests proving the new
guarded rollback boundary. Added registry file-mode preservation tests.
All other existing install/remove/reinstall tests (save-failure
cleanup, pre-existing-empty-dir handling, early-failure-during-
reinstall parametrized cases, Rich markup escaping) continue to pass
unmodified against the new implementation.

Verified via GraphQL that all 5 threads are current (not outdated/
resolved) before fixing. Full suite: 3996 passed, 110 skipped. Ruff
clean on all touched files.

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 | 303 +++++++++++++++++--------
 src/specify_cli/workflows/catalog.py   |  14 ++
 tests/test_workflows.py                | 267 ++++++++++++++--------
 3 files changed, 389 insertions(+), 195 deletions(-)

diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py
index e075de077..db9faad5a 100644
--- a/src/specify_cli/workflows/_commands.py
+++ b/src/specify_cli/workflows/_commands.py
@@ -262,6 +262,91 @@ def _safe_workflow_id_dir(workflows_dir: Path, workflow_id: str) -> Path:
     return dest_dir
 
 
+def _stage_workflow_file(dest_dir: Path) -> Path:
+    """Reserve a same-directory staging file so new/updated workflow.yml
+    content can be written and validated without ever touching (and risking
+    truncating) an existing destination file before the final atomic swap.
+    Shared by the local-install and catalog-install paths."""
+    import tempfile
+
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    fd, tmp_name = tempfile.mkstemp(dir=dest_dir, prefix=".workflow.yml.", suffix=".tmp")
+    os.close(fd)
+    return Path(tmp_name)
+
+
+def _commit_workflow_file(staged_file: Path, dest_file: Path, existed_before: bool) -> Path | None:
+    """Atomically swap ``staged_file`` onto ``dest_file``. If a prior file
+    existed, it is first renamed aside (path returned) so a later failure
+    (e.g. registry.add()) can restore it via rename instead of a content
+    rewrite -- the destination is never truncated/overwritten in place. If
+    the second rename fails after the first succeeded, the prior file is
+    put back immediately so dest_file is never left simply missing."""
+    if existed_before and dest_file.exists():
+        backup_file = dest_file.with_name(dest_file.name + ".bak")
+        os.replace(dest_file, backup_file)
+        try:
+            os.replace(staged_file, dest_file)
+        except OSError:
+            os.replace(backup_file, dest_file)
+            raise
+        return backup_file
+    os.replace(staged_file, dest_file)
+    return None
+
+
+def _discard_staged_workflow_file(staged_file: Path, dest_dir: Path, existed_before: bool) -> None:
+    """Clean up after a pre-commit failure (staged_file was never swapped
+    onto dest_file): remove the staged file, and for a fresh install (no
+    prior directory) remove the now-orphaned dest_dir too."""
+    staged_file.unlink(missing_ok=True)
+    if not existed_before:
+        import shutil
+        shutil.rmtree(dest_dir, ignore_errors=True)
+
+
+def _rollback_committed_workflow_file(
+    dest_file: Path, dest_dir: Path, existed_before: bool, backup_file: Path | None
+) -> None:
+    """Undo a successful _commit_workflow_file swap after a later failure
+    (registry.add()): restore the prior file via rename, remove the newly
+    committed file for a reinstall over a pre-existing empty directory
+    (no backup), or remove the whole directory for a fresh install."""
+    if backup_file is not None:
+        os.replace(backup_file, dest_file)
+    elif existed_before:
+        dest_file.unlink(missing_ok=True)
+    else:
+        import shutil
+        shutil.rmtree(dest_dir, ignore_errors=True)
+
+
+def _safe_discard_staged_workflow_file(staged_file: Path, dest_dir: Path, existed_before: bool) -> None:
+    """Guarded wrapper: a cleanup failure must be reported, never crash or
+    silently mask the original install error that triggered it."""
+    try:
+        _discard_staged_workflow_file(staged_file, dest_dir, existed_before)
+    except OSError as exc:
+        console.print(
+            "[yellow]Warning:[/yellow] Failed to clean up incomplete workflow "
+            f"install: {_escape_markup(str(exc))}"
+        )
+
+
+def _safe_rollback_committed_workflow_file(
+    dest_file: Path, dest_dir: Path, existed_before: bool, backup_file: Path | None
+) -> None:
+    """Guarded wrapper: a rollback failure must be reported, never crash or
+    silently claim the prior workflow file was restored when it wasn't."""
+    try:
+        _rollback_committed_workflow_file(dest_file, dest_dir, existed_before, backup_file)
+    except OSError as exc:
+        console.print(
+            "[yellow]Warning:[/yellow] Failed to restore prior workflow file "
+            f"after registry update failure: {_escape_markup(str(exc))}"
+        )
+
+
 # Root helper re-fetched at call time so test monkeypatching of
 # `specify_cli._require_specify_project` keeps working after the move.
 def _require_specify_project(*args, **kwargs):
@@ -474,9 +559,11 @@ def workflow_run(
                 # own .specify is itself a symlink to an attacker-controlled
                 # tree) -- check it explicitly rather than trusting that
                 # cwd-scoped guard, and don't rely on WorkflowRegistry's own
-                # symlinked-parent handling below (it silently substitutes
-                # an empty registry instead of raising, so a query against
-                # it can't be trusted as a safety signal here).
+                # symlinked-parent handling below as the safety signal here:
+                # it now fails closed by raising OSError at construction
+                # time (see catalog.py's _load), but that surfaces as an
+                # opaque exception rather than this guard's clean, specific
+                # CLI error for the actual owning project root.
                 _reject_unsafe_dir(registry_root / ".specify", ".specify")
                 _reject_unsafe_dir(
                     registry_root / ".specify" / "workflows", ".specify/workflows"
@@ -800,38 +887,39 @@ def workflow_add(
         dest_dir = _safe_workflow_id_dir(workflows_dir, definition.id)
         dest_file = dest_dir / "workflow.yml"
         existed_before = dest_dir.is_dir()
-        try:
-            backup_bytes = (
-                dest_file.read_bytes() if existed_before and dest_file.is_file() else None
-            )
-        except OSError as exc:
-            console.print(
-                f"[red]Error:[/red] Failed to read existing workflow "
-                f"'{_escape_markup(definition.id)}' before install: {_escape_markup(str(exc))}"
-            )
-            raise typer.Exit(1)
+
         import shutil
 
-        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.
-            # A pre-existing directory with no prior workflow.yml (no backup
-            # bytes) must have the newly written file removed instead of
-            # doing nothing, so it doesn't linger as an orphan.
-            if existed_before:
-                if backup_bytes is not None:
-                    dest_file.write_bytes(backup_bytes)
-                else:
-                    dest_file.unlink(missing_ok=True)
-            else:
-                shutil.rmtree(dest_dir, ignore_errors=True)
+        try:
+            staged_file = _stage_workflow_file(dest_dir)
+        except OSError as exc:
+            console.print(
+                f"[red]Error:[/red] Failed to install workflow "
+                f"'{_escape_markup(definition.id)}': {_escape_markup(str(exc))}"
+            )
+            raise typer.Exit(1)
 
         try:
-            dest_dir.mkdir(parents=True, exist_ok=True)
-            shutil.copy2(yaml_path, dest_file)
+            # Copied into the staging file, never dest_file directly, so a
+            # reinstall's prior working copy is never touched until the
+            # atomic commit below runs.
+            shutil.copy2(yaml_path, staged_file)
         except OSError as exc:
-            _cleanup_failed_install()
+            _safe_discard_staged_workflow_file(staged_file, dest_dir, existed_before)
+            console.print(
+                f"[red]Error:[/red] Failed to install workflow "
+                f"'{_escape_markup(definition.id)}': {_escape_markup(str(exc))}"
+            )
+            raise typer.Exit(1)
+
+        # Commit the staged copy onto dest_file via an atomic swap. A prior
+        # file (reinstall) is renamed aside rather than overwritten in
+        # place, so it can be restored by rename (not a content rewrite) if
+        # registry.add() below fails.
+        try:
+            backup_file = _commit_workflow_file(staged_file, dest_file, existed_before)
+        except OSError as exc:
+            _safe_discard_staged_workflow_file(staged_file, dest_dir, existed_before)
             console.print(
                 f"[red]Error:[/red] Failed to install workflow "
                 f"'{_escape_markup(definition.id)}': {_escape_markup(str(exc))}"
@@ -845,7 +933,7 @@ def workflow_add(
                 "source": source_label,
             })
         except OSError as exc:
-            _cleanup_failed_install()
+            _safe_rollback_committed_workflow_file(dest_file, dest_dir, existed_before, backup_file)
             console.print(
                 f"[red]Error:[/red] Failed to update workflow registry for "
                 f"'{_escape_markup(definition.id)}': {_escape_markup(str(exc))}"
@@ -1065,40 +1153,18 @@ def _install_workflow_from_catalog(
 
     # Captured before any mkdir/download writes so every failure branch below
     # can tell a fresh install from a reinstall-over-an-existing-one,
-    # mirroring _validate_and_install_local's existed-before/backup-aware
-    # rollback.
+    # mirroring _validate_and_install_local's existed-before-aware cleanup.
     existed_before = workflow_dir.is_dir()
+
     try:
-        prior_workflow_bytes = (
-            workflow_file.read_bytes() if existed_before and workflow_file.is_file() else None
-        )
+        staged_file = _stage_workflow_file(workflow_dir)
     except OSError as exc:
         console.print(
-            f"[red]Error:[/red] Failed to read existing workflow "
-            f"'{safe_wf_id}' before install: {_escape_markup(str(exc))}"
+            f"[red]Error:[/red] Failed to install workflow '{safe_wf_id}' from catalog: "
+            f"{_escape_markup(str(exc))}"
         )
         raise typer.Exit(1)
 
-    def _cleanup_failed_install() -> None:
-        """Restore the prior workflow.yml on a reinstall, or remove the
-        directory entirely for a fresh install. Every failure branch that
-        runs after the mkdir/download step -- redirect rejection, download
-        exception, invalid YAML, ID mismatch, version mismatch, and
-        registry.add() failure -- must call this instead of rmtree'ing
-        directly, so none of them can destroy a working install that
-        predates this attempt. A pre-existing directory with no prior
-        workflow.yml (no backup bytes) must have the newly downloaded file
-        removed instead of doing nothing, so it doesn't linger as an
-        orphan."""
-        if existed_before:
-            if prior_workflow_bytes is not None:
-                workflow_file.write_bytes(prior_workflow_bytes)
-            else:
-                workflow_file.unlink(missing_ok=True)
-        else:
-            import shutil
-            shutil.rmtree(workflow_dir, ignore_errors=True)
-
     try:
         from specify_cli.authentication.http import open_url as _open_url
         from specify_cli.authentication.http import github_provider_hosts as _github_provider_hosts
@@ -1112,7 +1178,6 @@ def _install_workflow_from_catalog(
             workflow_url = _resolved_workflow_url
             _wf_cat_extra_headers = {"Accept": "application/octet-stream"}
 
-        workflow_dir.mkdir(parents=True, exist_ok=True)
         with _open_url(
             workflow_url,
             timeout=30,
@@ -1131,31 +1196,35 @@ def _install_workflow_from_catalog(
                     # Host is not an IP literal (e.g., a regular hostname); treat as non-loopback.
                     pass
             if final_parsed.scheme != "https" and not (final_parsed.scheme == "http" and final_loopback):
-                _cleanup_failed_install()
+                _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
                 console.print(
                     f"[red]Error:[/red] Workflow '{safe_wf_id}' redirected to non-HTTPS URL: {_escape_markup(final_url)}"
                 )
                 raise typer.Exit(1)
-            workflow_file.write_bytes(_read_response_within_limit(response))
+            # Written to the staging file, never workflow_file directly, so a
+            # reinstall's prior working copy is never touched until the
+            # atomic commit below runs.
+            staged_file.write_bytes(_read_response_within_limit(response))
     except typer.Exit:
         raise
     except Exception as exc:
-        _cleanup_failed_install()
+        _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
         console.print(f"[red]Error:[/red] Failed to install workflow '{safe_wf_id}' from catalog: {_escape_markup(str(exc))}")
         raise typer.Exit(1)
 
-    # Validate the downloaded workflow before registering
+    # Validate the downloaded workflow (still staged, not yet committed)
+    # before registering.
     try:
-        definition = WorkflowDefinition.from_yaml(workflow_file)
+        definition = WorkflowDefinition.from_yaml(staged_file)
     except (ValueError, yaml.YAMLError) as exc:
-        _cleanup_failed_install()
+        _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
         console.print(f"[red]Error:[/red] Downloaded workflow is invalid: {_escape_markup(str(exc))}")
         raise typer.Exit(1)
 
     from .engine import validate_workflow
     errors = validate_workflow(definition)
     if errors:
-        _cleanup_failed_install()
+        _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
         console.print("[red]Error:[/red] Downloaded workflow validation failed:")
         for err in errors:
             console.print(f"  \u2022 {_escape_markup(str(err))}")
@@ -1163,7 +1232,7 @@ def _install_workflow_from_catalog(
 
     # Enforce that the workflow's internal ID matches the catalog key
     if definition.id and definition.id != workflow_id:
-        _cleanup_failed_install()
+        _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
         console.print(
             f"[red]Error:[/red] Workflow ID in YAML ({_escape_markup(repr(definition.id))}) "
             f"does not match catalog key ({_escape_markup(repr(workflow_id))}). "
@@ -1181,7 +1250,7 @@ def _install_workflow_from_catalog(
         except pkg_version.InvalidVersion:
             version_matches = str(definition.version) == expected_version
         if not version_matches:
-            _cleanup_failed_install()
+            _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
             console.print(
                 f"[red]Error:[/red] Downloaded workflow version ({_escape_markup(str(definition.version))}) "
                 f"does not match the catalog version ({_escape_markup(expected_version)}). "
@@ -1189,6 +1258,20 @@ def _install_workflow_from_catalog(
             )
             raise typer.Exit(1)
 
+    # Commit the staged download onto workflow_file via an atomic swap. A
+    # prior file (reinstall) is renamed aside rather than overwritten in
+    # place, so it can be restored by rename (not a content rewrite) if
+    # registry.add() below fails.
+    try:
+        backup_file = _commit_workflow_file(staged_file, workflow_file, existed_before)
+    except OSError as exc:
+        _safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
+        console.print(
+            f"[red]Error:[/red] Failed to install workflow '{safe_wf_id}' from catalog: "
+            f"{_escape_markup(str(exc))}"
+        )
+        raise typer.Exit(1)
+
     entry = {
         "name": definition.name or info.get("name", workflow_id),
         "version": definition.version or info.get("version", "0.0.0"),
@@ -1204,7 +1287,7 @@ def _install_workflow_from_catalog(
     try:
         registry.add(workflow_id, entry)
     except OSError as exc:
-        _cleanup_failed_install()
+        _safe_rollback_committed_workflow_file(workflow_file, workflow_dir, existed_before, backup_file)
         console.print(
             f"[red]Error:[/red] Failed to update workflow registry for "
             f"'{_escape_markup(workflow_id)}': {_escape_markup(str(exc))}"
@@ -1261,50 +1344,70 @@ def workflow_remove(
         )
         raise typer.Exit(1)
 
-    # Captured before the registry write so a subsequent directory-removal
-    # failure can restore it verbatim (bypassing add(), which would stamp a
-    # new updated_at), mirroring workflow_step_remove's same restore pattern.
-    registry_metadata = registry.get(workflow_id)
+    import shutil
+    import tempfile
 
-    # Persist the registry removal before touching any files: if save()
+    # Stage the directory out of the way via an atomic rename *before* the
+    # registry write, so a mid-delete rmtree failure can never leave a
+    # partially-deleted directory that gets re-marked "installed". A rename
+    # is a metadata-only operation (unlike rmtree), so it either fully
+    # succeeds or leaves the original directory completely untouched.
+    staged_dir: Path | None = None
+    if workflow_dir.exists():
+        try:
+            reserved = Path(
+                tempfile.mkdtemp(prefix=f".{workflow_id}.removing-", dir=workflows_dir)
+            )
+            reserved.rmdir()
+            os.rename(workflow_dir, reserved)
+            staged_dir = reserved
+        except OSError as exc:
+            console.print(
+                f"[red]Error:[/red] Failed to stage workflow directory "
+                f"{_escape_markup(str(workflow_dir))} for removal: {_escape_markup(str(exc))}"
+            )
+            raise typer.Exit(1)
+
+    # Persist the registry removal only after staging succeeded: if save()
     # fails, WorkflowRegistry.remove() rolls back its in-memory state and
-    # raises, so the workflow stays fully installed (files + registry) rather
-    # than being deleted while the registry still (or no longer) claims it.
+    # raises, so we rename the staged directory back to its original
+    # location, restoring the pre-command state exactly (files + registry
+    # both still claim the workflow installed).
     try:
         registry.remove(workflow_id)
     except OSError as exc:
+        if staged_dir is not None:
+            try:
+                os.rename(staged_dir, workflow_dir)
+            except OSError as restore_exc:
+                console.print(
+                    f"[yellow]Warning:[/yellow] Failed to restore workflow directory "
+                    f"after registry update failure; it remains staged at "
+                    f"{_escape_markup(str(staged_dir))}: {_escape_markup(str(restore_exc))}"
+                )
         console.print(
             f"[red]Error:[/red] Failed to update workflow registry for '{safe_id}': "
             f"{_escape_markup(str(exc))}"
         )
         raise typer.Exit(1)
 
-    if workflow_dir.exists():
-        import shutil
-        try:
-            shutil.rmtree(workflow_dir)
-        except OSError as exc:
-            # The registry removal already succeeded; restore the original
-            # entry verbatim so the registry doesn't claim this workflow is
-            # uninstalled while its directory is still sitting on disk.
-            try:
-                if registry_metadata is not None:
-                    registry.data["workflows"][workflow_id] = registry_metadata
-                    registry.save()
-            except Exception as restore_exc:  # noqa: BLE001
-                console.print(
-                    f"[yellow]Warning:[/yellow] Failed to restore registry entry "
-                    f"for '{safe_id}' after directory removal failure: "
-                    f"{_escape_markup(str(restore_exc))}"
-                )
-            console.print(
-                f"[red]Error:[/red] Failed to remove workflow directory "
-                f"{_escape_markup(str(workflow_dir))}: {_escape_markup(str(exc))}"
-            )
-            raise typer.Exit(1)
-
     console.print(f"[green]✓[/green] Workflow '{workflow_id}' removed")
 
+    # The registry has already durably committed the removal at this point,
+    # so it must stand regardless of what happens below: deleting the staged
+    # directory is now just cleanup, not a data-integrity concern, and a
+    # failure here is reported as a warning (not an error) to avoid
+    # contradicting the registry state that already succeeded.
+    if staged_dir is not None:
+        try:
+            shutil.rmtree(staged_dir)
+        except OSError as exc:
+            console.print(
+                f"[yellow]Warning:[/yellow] Workflow '{safe_id}' was removed, but its "
+                f"staged directory could not be deleted: {_escape_markup(str(exc))}. "
+                f"Remove it manually: {_escape_markup(str(staged_dir))}"
+            )
+
 
 @workflow_app.command("update")
 def workflow_update(
diff --git a/src/specify_cli/workflows/catalog.py b/src/specify_cli/workflows/catalog.py
index 65b24e726..2aa4764ae 100644
--- a/src/specify_cli/workflows/catalog.py
+++ b/src/specify_cli/workflows/catalog.py
@@ -13,6 +13,7 @@ from __future__ import annotations
 import hashlib
 import json
 import os
+import stat
 import tempfile
 import time
 from dataclasses import dataclass
@@ -146,6 +147,19 @@ class WorkflowRegistry:
         try:
             with os.fdopen(fd, "w", encoding="utf-8") as f:
                 json.dump(self.data, f, indent=2)
+            # mkstemp creates the temp file at 0600. A pre-existing registry
+            # may be shared more permissively (e.g. 0640/0644); preserve its
+            # mode across the replace so a save doesn't silently lock other
+            # project users out. A brand-new registry has no prior mode to
+            # preserve, so mkstemp's secure 0600 default stands. Mirrors
+            # _utils.py's atomic_write_json (best-effort; data safety over
+            # metadata preservation).
+            try:
+                if self.registry_path.exists():
+                    existing_mode = stat.S_IMODE(self.registry_path.stat().st_mode)
+                    os.chmod(tmp, existing_mode)
+            except OSError:
+                pass
             os.replace(tmp, self.registry_path)
         except BaseException:
             try:
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index 85339b95b..46a393ed4 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -15,6 +15,7 @@ from __future__ import annotations
 import json
 import os
 import shutil
+import stat
 import sys
 import tempfile
 from pathlib import Path
@@ -5985,18 +5986,28 @@ class TestWorkflowRemoveGuard:
         assert (workflow_dir / "workflow.yml").read_text(encoding="utf-8") == "keep-me"
         # The on-disk registry must still claim the workflow installed.
         assert WorkflowRegistry(project_dir).is_installed("test-wf")
+        # The directory must be restored to its exact original location, with
+        # no leftover staging directory from the stage/restore-on-failure
+        # sequence.
+        entries = [
+            p.name
+            for p in (project_dir / ".specify" / "workflows").iterdir()
+            if p.name != "workflow-registry.json"
+        ]
+        assert entries == ["test-wf"]
 
-    def test_remove_directory_failure_restores_registry_entry_verbatim(
+    def test_remove_staged_cleanup_failure_reports_warning_not_error(
         self, project_dir, monkeypatch
     ):
-        """If the registry removal already persisted successfully but the
-        subsequent shutil.rmtree fails, the directory was never actually
-        deleted (rmtree raised before removing anything usable), so the
-        registry must not be left claiming the workflow uninstalled. The
-        restored entry must be byte-for-byte the original (same
-        installed_at/updated_at) -- calling add() again would stamp a new
-        updated_at, which is why workflow_step_remove restores directly via
-        registry.data and save() instead of add()."""
+        """The directory is staged (atomically renamed out of
+        .specify/workflows/<id>) *before* the registry write, and the actual
+        deletion of the staged directory only happens *after* the registry
+        has already durably recorded the removal. If that final deletion
+        fails, the registry write already succeeded and must stand -- an
+        "Error: Failed to remove..." message at that point would contradict
+        the registry, which is exactly the incoherent state this staging
+        order exists to prevent. It must be reported as a cleanup warning,
+        and the command must still succeed."""
         from typer.testing import CliRunner
         from specify_cli import app
         from specify_cli.workflows.catalog import WorkflowRegistry
@@ -6007,8 +6018,6 @@ class TestWorkflowRemoveGuard:
         workflow_dir.mkdir(parents=True, exist_ok=True)
         (workflow_dir / "workflow.yml").write_text("keep-me", encoding="utf-8")
 
-        original_entry = WorkflowRegistry(project_dir).get("test-wf")
-
         def boom(*args, **kwargs):
             raise OSError("permission denied")
 
@@ -6017,26 +6026,31 @@ class TestWorkflowRemoveGuard:
             mp.setattr("shutil.rmtree", boom)
             result = CliRunner().invoke(app, ["workflow", "remove", "test-wf"])
 
-        assert result.exit_code != 0
-        assert result.exception is None or isinstance(result.exception, SystemExit)
-        assert "Failed to remove workflow directory" in result.output
-        # rmtree raised, so nothing was actually deleted.
-        assert workflow_dir.exists()
-        assert (workflow_dir / "workflow.yml").read_text(encoding="utf-8") == "keep-me"
-        # The registry entry must come back exactly as it was, not re-added.
-        restored = WorkflowRegistry(project_dir).get("test-wf")
-        assert restored == original_entry
-        assert WorkflowRegistry(project_dir).is_installed("test-wf")
+        assert result.exit_code == 0
+        assert "Warning" in result.output
+        # The registry write already committed -- it must stand.
+        assert not WorkflowRegistry(project_dir).is_installed("test-wf")
+        # The original install path is gone (staged away before the registry
+        # write ever ran); only a leftover staged directory remains, never
+        # at the original path the registry/CLI would treat as installed.
+        assert not workflow_dir.exists()
+        leftovers = [
+            p
+            for p in (project_dir / ".specify" / "workflows").iterdir()
+            if p.name != "workflow-registry.json"
+        ]
+        assert len(leftovers) == 1
+        assert (leftovers[0] / "workflow.yml").read_text(encoding="utf-8") == "keep-me"
 
-    def test_remove_directory_and_restore_failure_escapes_rich_markup(
+    def test_remove_stage_restore_failure_escapes_rich_markup(
         self, temp_dir, monkeypatch
     ):
-        """The project path (workflow_dir) and the rmtree/restore-save
-        exceptions interpolated into these new Rich error/warning messages
-        must be escaped like every other error path here -- unescaped Rich
-        markup characters (e.g. brackets) in a project directory name or an
-        OS/registry error message could otherwise be parsed as markup and
-        hide or corrupt the displayed text instead of showing it verbatim."""
+        """When the registry write fails (already rolled back in-memory by
+        WorkflowRegistry.remove()) and the attempt to rename the staged
+        directory back to its original location also fails, both the
+        restore exception and the registry-update exception interpolated
+        into these warning/error messages must be escaped like every other
+        error path here."""
         from typer.testing import CliRunner
         from specify_cli import app
         from specify_cli.workflows.catalog import WorkflowRegistry
@@ -6052,37 +6066,31 @@ class TestWorkflowRemoveGuard:
         workflow_dir.mkdir(parents=True, exist_ok=True)
         (workflow_dir / "workflow.yml").write_text("keep-me", encoding="utf-8")
 
-        def rmtree_boom(*args, **kwargs):
-            raise OSError("[disk] permission denied")
-
-        real_save = WorkflowRegistry.save
-        call_count = {"n": 0}
-
         def save_boom(self):
-            # The first save() call is registry.remove()'s own persist,
-            # which must succeed so we reach the rmtree failure below; only
-            # the second call (the post-rmtree-failure restore attempt)
-            # should fail, to exercise the restore-failure warning path.
-            call_count["n"] += 1
-            if call_count["n"] == 1:
-                return real_save(self)
-            raise Exception("[warn] save exploded")
+            raise OSError("[reg] disk full")
+
+        real_rename = os.rename
+        rename_calls = {"n": 0}
+
+        def rename_boom(src, dst):
+            rename_calls["n"] += 1
+            if rename_calls["n"] == 1:
+                # Allow the initial stage-out rename to succeed so the
+                # restore-back rename (the second call) is what fails.
+                return real_rename(src, dst)
+            raise OSError("[stage] permission denied")
 
         monkeypatch.chdir(project_dir)
         with pytest.MonkeyPatch.context() as mp:
-            mp.setattr("shutil.rmtree", rmtree_boom)
             mp.setattr(WorkflowRegistry, "save", save_boom)
+            mp.setattr(os, "rename", rename_boom)
             result = CliRunner().invoke(app, ["workflow", "remove", "test-wf"])
 
         assert result.exit_code != 0
         assert result.exception is None or isinstance(result.exception, SystemExit)
-        # Rich may soft-wrap the long path across lines; compare with
-        # whitespace collapsed so the wrap position doesn't affect the check.
         output_compact = "".join(result.output.split())
-        assert "".join(str(workflow_dir).split()) in output_compact
-        assert "[disk]permissiondenied" in output_compact
-        assert "[warn]saveexploded" in output_compact
-
+        assert "[stage]permissiondenied" in output_compact
+        assert "[reg]diskfull" in output_compact
 
 class TestWorkflowAddSymlinkGuard:
     def test_add_malformed_ipv6_url_exits_cleanly(self, temp_dir, monkeypatch):
@@ -8034,6 +8042,33 @@ steps:
             WorkflowRegistry(project_dir)
         assert not (outside / "workflows").exists()
 
+    def test_registry_save_preserves_existing_file_mode(self, project_dir):
+        """A registry shared as 0640/0644 must keep that mode after a save,
+        not be silently replaced by mkstemp's 0600 default -- otherwise
+        every add/remove locks other project users out of a previously
+        shared registry file."""
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        registry = WorkflowRegistry(project_dir)
+        registry.add("first-wf", {"name": "First"})
+        registry.registry_path.chmod(0o644)
+
+        registry.add("second-wf", {"name": "Second"})
+
+        mode = stat.S_IMODE(registry.registry_path.stat().st_mode)
+        assert mode == 0o644, f"expected 0644, got {oct(mode)}"
+
+    def test_registry_save_on_new_registry_uses_secure_default_mode(self, project_dir):
+        """A brand-new registry file (no prior mode to preserve) should keep
+        mkstemp's secure 0600 default rather than something more permissive."""
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        registry = WorkflowRegistry(project_dir)
+        registry.add("first-wf", {"name": "First"})
+
+        mode = stat.S_IMODE(registry.registry_path.stat().st_mode)
+        assert mode == 0o600, f"expected 0600, got {oct(mode)}"
+
     def test_add_dev_dir_with_workflow_yml_directory_errors_cleanly(self, project_dir, monkeypatch):
         from typer.testing import CliRunner
         from specify_cli import app
@@ -8089,13 +8124,16 @@ 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()."""
+    def test_add_dev_reinstall_copy_failure_leaves_prior_file_untouched(
+        self, project_dir, monkeypatch
+    ):
+        """copy2 now writes into a same-directory staging file, never
+        dest_file directly, so a copy2 failure (even one that partially
+        writes before raising, mirroring a real disk-full/interrupted-copy
+        failure) can no longer touch -- let alone truncate -- the prior
+        working workflow.yml: dest_file is only ever touched by the final
+        atomic commit swap, which never runs if copy2 raises. No leftover
+        staging file may remain either."""
         from typer.testing import CliRunner
         from specify_cli import app
         from specify_cli.workflows.catalog import WorkflowRegistry
@@ -8113,10 +8151,11 @@ steps:
             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"")
+        def boom(src_path, dst_path, *args, **kwargs):
+            # Simulate a truncating partial write followed by an OSError on
+            # the *staging* file -- the only file copy2 is now allowed to
+            # touch -- mirroring a real disk-full/interrupted-copy failure.
+            Path(dst_path).write_bytes(b"")
             raise OSError("disk full")
 
         with pytest.MonkeyPatch.context() as mp:
@@ -8128,44 +8167,58 @@ steps:
         assert result.output.strip() != ""
         assert installed_yaml.read_bytes() == original_bytes
         assert WorkflowRegistry(project_dir).get("align-wf") == original_registry_entry
+        # No orphaned staging file left behind in the workflow directory.
+        leftovers = [p.name for p in installed_yaml.parent.iterdir() if p.name != "workflow.yml"]
+        assert leftovers == []
 
-    def test_add_dev_reinstall_backup_read_failure_gives_clean_error(
+    def test_add_dev_reinstall_restore_failure_reports_warning_and_original_error(
         self, project_dir, monkeypatch
     ):
-        """The prior-file backup read (used to restore on a later install
-        failure) ran before any guarded section: a read failure on the
-        existing workflow.yml (e.g. a transient permission/FS issue) leaked
-        a raw OSError instead of the clean escaped CLI error used by every
-        other failure branch here, and left the destination untouched since
-        it happens before any write."""
+        """The prior file is now restored via an atomic rename (not a
+        content rewrite) when registry.add() fails on a reinstall. If that
+        restore rename itself also fails (e.g. a transient FS issue), it
+        must not silently claim success or crash with a raw traceback: it
+        must report a clear warning about the restore failure in addition
+        to the original clean registry error."""
         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()
 
         (src / "workflow.yml").write_text(
             self.WORKFLOW_YAML.format(version="2.0.0"), encoding="utf-8"
         )
 
-        real_read_bytes = Path.read_bytes
+        def save_boom(self):
+            raise OSError("disk full")
 
-        def boom(self_path, *args, **kwargs):
-            if self_path.resolve() == installed_yaml.resolve():
-                raise OSError("permission denied")
-            return real_read_bytes(self_path, *args, **kwargs)
+        real_replace = os.replace
+        calls = {"n": 0}
+
+        def replace_boom(src_path, dst_path):
+            # The commit swap for a reinstall makes exactly two os.replace
+            # calls (backup-aside, then staged-into-dest); let both succeed
+            # and only fail the third call -- the post-registry-failure
+            # restore-back rename.
+            calls["n"] += 1
+            if calls["n"] <= 2:
+                return real_replace(src_path, dst_path)
+            raise OSError("permission denied")
 
         with pytest.MonkeyPatch.context() as mp:
-            mp.setattr(Path, "read_bytes", boom)
+            mp.setattr(WorkflowRegistry, "save", save_boom)
+            mp.setattr(os, "replace", replace_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
+        output_compact = "".join(result.output.split())
+        assert "Warning" in result.output
+        assert "diskfull" in output_compact
+        assert "permissiondenied" in output_compact
 
     def test_add_dev_fresh_install_into_preexisting_empty_dir_cleans_new_file(
         self, project_dir, monkeypatch
@@ -8390,18 +8443,18 @@ steps:
         assert registry.is_installed("align-wf")
         assert registry.get("align-wf")["version"] == "1.0.0"
 
-    def test_add_catalog_reinstall_backup_read_failure_gives_clean_error(
+    def test_add_catalog_reinstall_restore_failure_reports_warning_and_original_error(
         self, project_dir, monkeypatch
     ):
-        """Same backup-read boundary gap as the local-install path: the
-        prior-file read used to seed the reinstall's rollback ran before
-        the download/validation error boundary, so a read failure on the
-        existing workflow.yml (e.g. a transient permission/FS issue) leaked
-        a raw OSError instead of a clean escaped CLI error, and must be
-        caught before any download/write is attempted."""
+        """Same restore-rename boundary as the local-install path: the
+        prior file is restored via an atomic rename (not a content rewrite)
+        when registry.add() fails on a reinstall. If that restore rename
+        itself also fails, it must report a clear warning in addition to
+        the original clean registry error, never crash or silently claim
+        success."""
         from typer.testing import CliRunner
         from specify_cli import app
-        from specify_cli.workflows.catalog import WorkflowCatalog
+        from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry
 
         monkeypatch.chdir(project_dir)
         monkeypatch.setattr(
@@ -8428,22 +8481,41 @@ steps:
             result = runner.invoke(app, ["workflow", "add", "align-wf"])
         assert result.exit_code == 0, result.output
 
-        dest_file = project_dir / ".specify" / "workflows" / "align-wf" / "workflow.yml"
-        real_read_bytes = Path.read_bytes
+        new_data = self.WORKFLOW_YAML.format(version="2.0.0").encode()
 
-        def boom(self_path, *args, **kwargs):
-            if self_path.resolve() == dest_file.resolve():
-                raise OSError("permission denied")
-            return real_read_bytes(self_path, *args, **kwargs)
+        def save_boom(self):
+            raise OSError("disk full")
+
+        real_replace = os.replace
+        calls = {"n": 0}
+
+        def replace_boom(src_path, dst_path):
+            # The commit swap for a reinstall makes exactly two os.replace
+            # calls (backup-aside, then staged-into-dest); let both succeed
+            # and only fail the third call -- the post-registry-failure
+            # restore-back rename.
+            calls["n"] += 1
+            if calls["n"] <= 2:
+                return real_replace(src_path, dst_path)
+            raise OSError("permission denied")
 
         with pytest.MonkeyPatch.context() as mp:
-            mp.setattr(Path, "read_bytes", boom)
+            mp.setattr(
+                "specify_cli.authentication.http.open_url",
+                lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(
+                    new_data, url
+                ),
+            )
+            mp.setattr(WorkflowRegistry, "save", save_boom)
+            mp.setattr(os, "replace", replace_boom)
             result = runner.invoke(app, ["workflow", "add", "align-wf"])
 
         assert result.exit_code != 0
         assert result.exception is None or isinstance(result.exception, SystemExit)
-        assert result.output.strip() != ""
-        assert dest_file.read_bytes() == original_data
+        output_compact = "".join(result.output.split())
+        assert "Warning" in result.output
+        assert "diskfull" in output_compact
+        assert "permissiondenied" in output_compact
 
     def test_add_catalog_fresh_install_into_preexisting_empty_dir_cleans_new_file(
         self, project_dir, monkeypatch
@@ -8612,11 +8684,15 @@ steps:
         assert seen["validator"] is _reject_insecure_download_redirect
 
     def test_registry_save_failure_preserves_file_on_disk(self, project_dir, monkeypatch):
-        """A failed dump must not truncate the persisted registry."""
+        """A failed dump must not truncate the persisted registry, and must
+        not alter its on-disk mode either -- the chmod-to-match-existing-mode
+        step operates on the temp file, never the target, so a failed save
+        (which never reaches os.replace) cannot touch the original's mode."""
         from specify_cli.workflows.catalog import WorkflowRegistry
 
         registry = WorkflowRegistry(project_dir)
         registry.add("align-wf", {"version": "1.0.0", "source": "catalog"})
+        registry.registry_path.chmod(0o644)
 
         import specify_cli.workflows.catalog as catalog_mod
 
@@ -8630,6 +8706,7 @@ steps:
 
         fresh = WorkflowRegistry(project_dir)
         assert fresh.get("align-wf")["version"] == "1.0.0"
+        assert stat.S_IMODE(registry.registry_path.stat().st_mode) == 0o644
         assert not list(registry.workflows_dir.glob("*.tmp"))
 
     def test_update_mixed_targets_does_not_claim_all_up_to_date(self, project_dir, monkeypatch):