From eabce4efe1d2ac2b6a1b85a1bd50c943f77d3312 Mon Sep 17 00:00:00 2001
From: marcelsafin <179933638+marcelsafin@users.noreply.github.com>
Date: Sat, 11 Jul 2026 00:34:05 +0200
Subject: [PATCH] fix(workflows): exclusive temp file for registry save and
 cwd-independent disabled guard

- save() now uses tempfile.mkstemp in the workflows dir (matching the
  engine's atomic writer), so a pre-created symlink at a predictable
  .tmp path cannot redirect the write and concurrent processes cannot
  collide.
- The direct-path disabled guard derives the owning project from the
  resolved file path instead of the caller's cwd, so running an
  installed workflow's YAML from outside the project still refuses when
  disabled.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/specify_cli/workflows/_commands.py | 17 ++++++++++-------
 src/specify_cli/workflows/catalog.py   | 19 +++++++++++++------
 tests/test_workflows.py                | 10 ++++++++++
 3 files changed, 33 insertions(+), 13 deletions(-)

diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py
index 973e69fe0..334c7e254 100644
--- a/src/specify_cli/workflows/_commands.py
+++ b/src/specify_cli/workflows/_commands.py
@@ -352,6 +352,7 @@ def workflow_run(
     from .catalog import WorkflowRegistry
 
     registered_id: str | None = None
+    registry_root = project_root
     if not is_file_source:
         # Reject path-equivalent spellings ("align-wf/", "align-wf/.") that
         # would miss the registry lookup yet still load the installed file,
@@ -364,16 +365,18 @@ 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 ID so disabled state is enforced there too.
-        workflows_root = (project_root / ".specify" / "workflows").resolve()
+        # 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()
-        if resolved.is_relative_to(workflows_root):
-            rel_parts = resolved.relative_to(workflows_root).parts
-            if rel_parts:
-                registered_id = rel_parts[0]
+        parts = resolved.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 ".")
+                registered_id = parts[i + 2]
+                break
 
     if registered_id is not None:
-        installed_meta = WorkflowRegistry(project_root).get(registered_id)
+        installed_meta = WorkflowRegistry(registry_root).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. "
diff --git a/src/specify_cli/workflows/catalog.py b/src/specify_cli/workflows/catalog.py
index 942a6cfd0..87fe1ca0b 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 tempfile
 import time
 from dataclasses import dataclass
 from pathlib import Path
@@ -85,15 +86,21 @@ class WorkflowRegistry:
     def save(self) -> None:
         """Persist registry to disk atomically."""
         self.workflows_dir.mkdir(parents=True, exist_ok=True)
-        # Write-then-replace so a failed dump cannot truncate the registry.
-        tmp_path = self.registry_path.with_name(self.registry_path.name + ".tmp")
+        # Unique, exclusive temp then replace: a failed dump cannot truncate
+        # the registry, a pre-created symlink cannot redirect the write, and
+        # concurrent CLI processes cannot collide on the same temp path.
+        fd, tmp = tempfile.mkstemp(
+            dir=str(self.registry_path.parent),
+            prefix=f".{self.registry_path.name}.",
+            suffix=".tmp",
+        )
         try:
-            with open(tmp_path, "w", encoding="utf-8") as f:
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
                 json.dump(self.data, f, indent=2)
-            os.replace(tmp_path, self.registry_path)
-        except OSError:
+            os.replace(tmp, self.registry_path)
+        except BaseException:
             try:
-                tmp_path.unlink()
+                os.unlink(tmp)
             except OSError:
                 pass
             raise
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index 77589fb61..810a94dc3 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -8211,6 +8211,16 @@ steps:
         assert result.exit_code != 0
         assert "disabled" in result.output
 
+        # Same guard must hold when invoked from outside the project.
+        outside = project_dir.parent / "outside-cwd"
+        outside.mkdir(exist_ok=True)
+        monkeypatch.chdir(outside)
+        result = runner.invoke(
+            app, ["workflow", "run", str(project_dir / installed_yaml)]
+        )
+        assert result.exit_code != 0
+        assert "disabled" in result.output
+
     def test_disable_shows_marker_in_list(self, project_dir, monkeypatch):
         from typer.testing import CliRunner
         from specify_cli import app