From 1650af68098e50adbc795e8f069c0f1327ae900a Mon Sep 17 00:00:00 2001
From: "github-actions[bot]"
 <41898282+github-actions[bot]@users.noreply.github.com>
Date: Mon, 12 Oct 2026 14:28:10 +0000
Subject: [PATCH 1/8] Fix persisted-run-symlinks: reject symlinked workflow
 artifacts

Apply the remediation from the bug assessment on issue #4914.

Refs #4914

Assisted-by: GitHub Copilot (model: gpt-5.2-codex, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 src/specify_cli/workflows/command_status.py   |   7 +-
 src/specify_cli/workflows/engine.py           | 123 ++++++++++++++----
 .../workflows/test_run_state_storage.py       | 103 +++++++++++++++
 workflows/ARCHITECTURE.md                     |   9 ++
 workflows/README.md                           |   6 +
 5 files changed, 224 insertions(+), 24 deletions(-)
 create mode 100644 tests/specify_cli/workflows/test_run_state_storage.py

diff --git a/src/specify_cli/workflows/command_status.py b/src/specify_cli/workflows/command_status.py
index 97471eb10c..b608c6d40b 100644
--- a/src/specify_cli/workflows/command_status.py
+++ b/src/specify_cli/workflows/command_status.py
@@ -100,7 +100,12 @@ def workflow_status(
                     f"{cli._escape_markup(scope['workflow_id'])}: {scope['status']}"
                 )
     else:
-        runs = engine.list_runs()
+        err = cli._error_console(json_output)
+        try:
+            runs = engine.list_runs()
+        except (OSError, ValueError) as exc:
+            err.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}")
+            raise cli.typer.Exit(1)
 
         if json_output:
             payload = {
diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index 5e85353834..9d85b58b7e 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -610,6 +610,10 @@ def _validate_steps(
 class RunState:
     """Manages workflow run state for persistence and resume."""
 
+    _RUN_ARTIFACTS = frozenset(
+        {"state.json", "inputs.json", "workflow.yml", "log.jsonl"}
+    )
+
     # ``run_id`` is interpolated into a filesystem path (``runs/<run_id>``)
     # by both ``save()`` and ``load()``. Constrain it to a charset that
     # cannot contain path separators (``/`` ``\``), parent-directory
@@ -637,6 +641,59 @@ def _validate_run_id(cls, run_id: str) -> None:
                 "alphanumeric character)."
             )
 
+    @classmethod
+    def _run_directory(
+        cls, project_root: Path, run_id: str, *, create: bool = False
+    ) -> Path:
+        """Return a run directory after enforcing the persisted-run boundary."""
+        cls._validate_run_id(run_id)
+        runs_root = project_root / ".specify" / "workflows" / "runs"
+        if create:
+            runs_root.mkdir(parents=True, exist_ok=True)
+        cls._reject_symlink(runs_root, ".specify/workflows/runs")
+
+        run_dir = runs_root / run_id
+        if run_dir.is_symlink():
+            raise ValueError(
+                f"Refusing to use symlinked workflow run path: {run_dir}"
+            )
+        if run_dir.exists() and not run_dir.is_dir():
+            raise OSError(f"Workflow run path is not a directory: {run_dir}")
+        if create:
+            run_dir.mkdir(parents=True, exist_ok=True)
+        return run_dir
+
+    @staticmethod
+    def _reject_symlink(path: Path, label: str) -> None:
+        """Reject an existing symlink without resolving its target."""
+        if path.is_symlink():
+            raise ValueError(f"Refusing to use symlinked {label} path: {path}")
+
+    @classmethod
+    def _run_artifact(
+        cls, project_root: Path, run_id: str, name: str, *, create: bool = False
+    ) -> Path:
+        if name not in cls._RUN_ARTIFACTS:
+            raise ValueError(f"Unknown workflow run artifact: {name}")
+        run_dir = cls._run_directory(project_root, run_id, create=create)
+        artifact = run_dir / name
+        cls._reject_symlink(artifact, f"workflow run artifact {name!r}")
+        return artifact
+
+    @staticmethod
+    def _open_artifact(path: Path, mode: str, **kwargs):
+        """Open a run artifact without following a symlink at the leaf."""
+        flags = {
+            "r": os.O_RDONLY,
+            "a": os.O_WRONLY | os.O_CREAT | os.O_APPEND,
+            "w": os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
+        }.get(mode)
+        if flags is None:
+            raise ValueError(f"Unsupported workflow artifact open mode: {mode!r}")
+        nofollow = getattr(os, "O_NOFOLLOW", 0)
+        fd = os.open(path, flags | nofollow, 0o666)
+        return os.fdopen(fd, mode, **kwargs)
+
     @staticmethod
     def _validate_installed_origin(
         installed_workflow_id: str | None,
@@ -734,7 +791,7 @@ def __init__(
 
     @property
     def runs_dir(self) -> Path:
-        return self.project_root / ".specify" / "workflows" / "runs" / self.run_id
+        return self._run_directory(self.project_root, self.run_id)
 
     def save(self) -> None:
         """Persist current state to disk.
@@ -744,8 +801,7 @@ def save(self) -> None:
         nor leave a reader observing a half-written file. Racing writers only
         contend to be last; they never corrupt.
         """
-        runs_dir = self.runs_dir
-        runs_dir.mkdir(parents=True, exist_ok=True)
+        self._run_directory(self.project_root, self.run_id, create=True)
 
         with self._lock:
             if self._checkpoint_failed:
@@ -773,8 +829,18 @@ def save(self) -> None:
             if self.execution is not None:
                 state_data["execution"] = self.execution
             try:
-                self._atomic_write_json(runs_dir / "state.json", state_data)
-                self._atomic_write_json(runs_dir / "inputs.json", {"inputs": self.inputs})
+                self._atomic_write_json(
+                    self._run_artifact(
+                        self.project_root, self.run_id, "state.json", create=True
+                    ),
+                    state_data,
+                )
+                self._atomic_write_json(
+                    self._run_artifact(
+                        self.project_root, self.run_id, "inputs.json", create=True
+                    ),
+                    {"inputs": self.inputs},
+                )
             except Exception as exc:
                 # Only write failures poison the instance. A graceful interrupt
                 # (KeyboardInterrupt) propagates unchanged: during step
@@ -790,6 +856,7 @@ def save(self) -> None:
     @staticmethod
     def _atomic_write_json(path: Path, data: dict[str, Any]) -> None:
         """Write *data* as indented JSON to *path* atomically (temp + ``os.replace``)."""
+        RunState._reject_symlink(path, f"workflow run artifact {path.name!r}")
         fd, tmp = tempfile.mkstemp(
             dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp"
         )
@@ -823,12 +890,10 @@ def load(cls, run_id: str, project_root: Path) -> RunState:
         *after* the file lookup — too late to prevent the disclosure.
         Mirrors the precedent in ``agents._ensure_within_directory``.
         """
-        cls._validate_run_id(run_id)
-        runs_dir = project_root / ".specify" / "workflows" / "runs" / run_id
-        state_path = runs_dir / "state.json"
+        state_path = cls._run_artifact(project_root, run_id, "state.json")
 
         try:
-            with open(state_path, encoding="utf-8") as f:
+            with cls._open_artifact(state_path, "r", encoding="utf-8") as f:
                 state_data = json.load(f)
         except FileNotFoundError:
             msg = f"Run state not found: {state_path}"
@@ -920,9 +985,9 @@ def load(cls, run_id: str, project_root: Path) -> RunState:
 
             validate_execution(state.execution)
 
-        inputs_path = runs_dir / "inputs.json"
+        inputs_path = cls._run_artifact(project_root, run_id, "inputs.json")
         if "inputs" not in state_data and inputs_path.exists():
-            with open(inputs_path, encoding="utf-8") as f:
+            with cls._open_artifact(inputs_path, "r", encoding="utf-8") as f:
                 inputs_data = json.load(f)
             if not isinstance(inputs_data, dict):
                 raise ValueError(
@@ -951,8 +1016,7 @@ def append_log(self, entry: dict[str, Any]) -> None:
         worker can log once any checkpoint of this run has failed.
         """
         entry["timestamp"] = datetime.now(timezone.utc).isoformat()
-        runs_dir = self.runs_dir
-        runs_dir.mkdir(parents=True, exist_ok=True)
+        self._run_directory(self.project_root, self.run_id, create=True)
         with self._lock:
             if self._checkpoint_failed:
                 from ._execution import CheckpointError
@@ -960,8 +1024,10 @@ def append_log(self, entry: dict[str, Any]) -> None:
                 raise CheckpointError("A previous checkpoint failed; reload the run")
             with self._log_lock:
                 self.log_entries.append(entry)
-                with open(
-                    runs_dir / "log.jsonl",
+                with self._open_artifact(
+                    self._run_artifact(
+                        self.project_root, self.run_id, "log.jsonl", create=True
+                    ),
                     "a",
                     encoding="utf-8",
                     errors="backslashreplace",
@@ -1110,11 +1176,14 @@ def execute(
         # Persist a copy of the workflow definition so resume can
         # reload it even if the original source is no longer available
         # (e.g. a local YAML path that was moved or deleted).
-        run_dir = self.project_root / ".specify" / "workflows" / "runs" / state.run_id
-        run_dir.mkdir(parents=True, exist_ok=True)
-        workflow_copy = run_dir / "workflow.yml"
+        run_dir = RunState._run_directory(
+            self.project_root, state.run_id, create=True
+        )
+        workflow_copy = RunState._run_artifact(
+            self.project_root, state.run_id, "workflow.yml", create=True
+        )
         import yaml
-        with open(workflow_copy, "w", encoding="utf-8") as f:
+        with RunState._open_artifact(workflow_copy, "w", encoding="utf-8") as f:
             yaml.safe_dump(definition.data, f, sort_keys=False, allow_unicode=True)
 
         # Resolve inputs
@@ -1182,8 +1251,9 @@ def resume(
         # Load the workflow definition — try the persisted copy in the
         # run directory first so resume works even if the original
         # source (e.g. a local YAML path) is no longer available.
-        run_dir = self.project_root / ".specify" / "workflows" / "runs" / run_id
-        run_copy = run_dir / "workflow.yml"
+        run_copy = RunState._run_artifact(
+            self.project_root, run_id, "workflow.yml"
+        )
         if run_copy.exists():
             definition = WorkflowDefinition.from_yaml(run_copy)
         else:
@@ -1484,15 +1554,22 @@ def list_runs(self) -> list[dict[str, Any]]:
         runs_dir = self.project_root / ".specify" / "workflows" / "runs"
         if not runs_dir.exists():
             return []
+        self._reject_symlink(runs_dir, ".specify/workflows/runs")
 
         runs: list[dict[str, Any]] = []
         for run_dir in sorted(runs_dir.iterdir()):
+            if run_dir.is_symlink():
+                raise ValueError(
+                    f"Refusing to use symlinked workflow run path: {run_dir}"
+                )
             if not run_dir.is_dir():
                 continue
-            state_path = run_dir / "state.json"
+            state_path = self._run_artifact(
+                self.project_root, run_dir.name, "state.json"
+            )
             if state_path.exists():
                 try:
-                    with open(state_path, encoding="utf-8") as f:
+                    with self._open_artifact(state_path, "r", encoding="utf-8") as f:
                         state_data = json.load(f)
                 except (json.JSONDecodeError, OSError, UnicodeDecodeError):
                     continue
diff --git a/tests/specify_cli/workflows/test_run_state_storage.py b/tests/specify_cli/workflows/test_run_state_storage.py
new file mode 100644
index 0000000000..9b97a65a27
--- /dev/null
+++ b/tests/specify_cli/workflows/test_run_state_storage.py
@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import json
+import os
+
+import pytest
+
+
+pytestmark = pytest.mark.skipif(
+    not hasattr(os, "symlink"), reason="symlinks are unavailable"
+)
+
+
+def _run_dir(project_dir, run_id="run-1"):
+    path = project_dir / ".specify" / "workflows" / "runs" / run_id
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def _state(project_dir, run_id="run-1"):
+    from specify_cli.workflows.base import RunStatus
+    from specify_cli.workflows.engine import RunState
+
+    state = RunState(run_id=run_id, workflow_id="workflow", project_root=project_dir)
+    state.status = RunStatus.PAUSED
+    return state
+
+
+def test_load_rejects_symlinked_run_directory(project_dir, tmp_path):
+    from specify_cli.workflows.engine import RunState
+
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "state.json").write_text(
+        json.dumps({"run_id": "run-1", "workflow_id": "workflow", "status": "paused"}),
+        encoding="utf-8",
+    )
+    runs = project_dir / ".specify" / "workflows" / "runs"
+    runs.mkdir(parents=True)
+    (runs / "run-1").symlink_to(outside, target_is_directory=True)
+
+    with pytest.raises(ValueError, match="symlinked workflow run path"):
+        RunState.load("run-1", project_dir)
+
+
+@pytest.mark.parametrize("artifact", ["state.json", "inputs.json"])
+def test_run_state_rejects_symlinked_persisted_artifacts(
+    project_dir, tmp_path, artifact
+):
+    state = _state(project_dir)
+    run_dir = _run_dir(project_dir)
+    target = tmp_path / artifact
+    target.write_text("{}", encoding="utf-8")
+    (run_dir / artifact).symlink_to(target)
+
+    if artifact == "state.json":
+        with pytest.raises(ValueError, match="symlinked workflow run artifact"):
+            type(state).load("run-1", project_dir)
+    else:
+        state_data = {
+            "run_id": "run-1",
+            "workflow_id": "workflow",
+            "status": "paused",
+        }
+        (run_dir / "state.json").write_text(json.dumps(state_data), encoding="utf-8")
+        with pytest.raises(ValueError, match="symlinked workflow run artifact"):
+            type(state).load("run-1", project_dir)
+
+
+@pytest.mark.parametrize("artifact, operation", [
+    ("state.json", "save"),
+    ("inputs.json", "save"),
+    ("log.jsonl", "log"),
+])
+def test_run_state_rejects_symlinked_artifact_writes(
+    project_dir, tmp_path, artifact, operation
+):
+    from specify_cli.workflows._execution import CheckpointError
+
+    state = _state(project_dir)
+    run_dir = _run_dir(project_dir)
+    target = tmp_path / artifact
+    target.write_text("", encoding="utf-8")
+    (run_dir / artifact).symlink_to(target)
+
+    expected = CheckpointError if operation == "save" else ValueError
+    with pytest.raises(expected, match="symlinked workflow run artifact"):
+        if operation == "save":
+            state.save()
+        else:
+            state.append_log({"event": "test"})
+
+
+def test_workflow_snapshot_artifact_is_checked(project_dir, tmp_path):
+    from specify_cli.workflows.engine import RunState
+
+    run_dir = _run_dir(project_dir)
+    target = tmp_path / "workflow.yml"
+    target.write_text("workflow: {}", encoding="utf-8")
+    (run_dir / "workflow.yml").symlink_to(target)
+
+    with pytest.raises(ValueError, match="symlinked workflow run artifact"):
+        RunState._run_artifact(project_dir, "run-1", "workflow.yml")
diff --git a/workflows/ARCHITECTURE.md b/workflows/ARCHITECTURE.md
index 6382e0c35e..28b7f9043a 100644
--- a/workflows/ARCHITECTURE.md
+++ b/workflows/ARCHITECTURE.md
@@ -280,6 +280,15 @@ When `specify workflow add <id>` installs from catalog, it downloads the workflo
 | Project catalogs | `.specify/workflow-catalogs.yml` | YAML | Project-level catalog sources |
 | User catalogs | `~/.specify/workflow-catalogs.yml` | YAML | User-level catalog sources |
 
+### Persisted-run storage boundary
+
+The `.specify/workflows/runs` tree is a project-owned persistence boundary.
+Every run directory and persisted artifact below it must be a real filesystem
+object rather than a symlink. `RunState` enforces this policy before reading or
+writing `state.json`, `inputs.json`, `workflow.yml`, or `log.jsonl`; artifact
+opens also use no-follow flags where the platform provides them. Commands
+fail closed instead of following redirected run storage.
+
 ## Module Structure
 
 ```text
diff --git a/workflows/README.md b/workflows/README.md
index c952b4ad4b..ce552a60c6 100644
--- a/workflows/README.md
+++ b/workflows/README.md
@@ -520,6 +520,12 @@ specify workflow resume <run_id>
 
 Run states: `created` → `running` → `completed` | `paused` | `failed` | `aborted`
 
+Persisted-run storage rejects every symlink below `.specify/workflows/runs`,
+including per-run directories and artifact files. This applies consistently to
+state, inputs, workflow snapshots, and logs for both reads and writes. Workflow
+storage cannot therefore be relocated through symlinks; move the files and
+directories into the project-owned runs directory instead.
+
 ## Catalog Management
 
 Workflows are discovered through catalogs. By default, Spec Kit uses the official and community catalogs:

From 79f1da76592e906a42fce6ec659c90e0eb7b9ed9 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Mon, 12 Oct 2026 19:36:25 +0000
Subject: [PATCH 2/8] Fix persisted workflow snapshot symlinks

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>
---
 src/specify_cli/workflows/engine.py           | 17 +++++++--
 .../workflows/test_command_status.py          | 32 +++++++++++++++-
 .../workflows/test_run_state_storage.py       | 38 +++++++++++++++++++
 3 files changed, 82 insertions(+), 5 deletions(-)

diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index 9d85b58b7e..0393ca7f2a 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -1176,9 +1176,6 @@ def execute(
         # Persist a copy of the workflow definition so resume can
         # reload it even if the original source is no longer available
         # (e.g. a local YAML path that was moved or deleted).
-        run_dir = RunState._run_directory(
-            self.project_root, state.run_id, create=True
-        )
         workflow_copy = RunState._run_artifact(
             self.project_root, state.run_id, "workflow.yml", create=True
         )
@@ -1255,7 +1252,19 @@ def resume(
             self.project_root, run_id, "workflow.yml"
         )
         if run_copy.exists():
-            definition = WorkflowDefinition.from_yaml(run_copy)
+            with RunState._open_artifact(run_copy, "r", encoding="utf-8") as f:
+                try:
+                    data = yaml.safe_load(f)
+                except yaml.YAMLError as exc:
+                    msg = f"Invalid YAML in {run_copy}: {exc}"
+                    raise ValueError(msg) from exc
+            if not isinstance(data, dict):
+                msg = (
+                    "Workflow YAML must be a mapping, "
+                    f"got {type(data).__name__}."
+                )
+                raise ValueError(msg)
+            definition = WorkflowDefinition(data, source_path=run_copy)
         else:
             definition = self.load_workflow(state.workflow_id)
 
diff --git a/tests/specify_cli/workflows/test_command_status.py b/tests/specify_cli/workflows/test_command_status.py
index a29a4001eb..7b25c38330 100644
--- a/tests/specify_cli/workflows/test_command_status.py
+++ b/tests/specify_cli/workflows/test_command_status.py
@@ -3,13 +3,13 @@
 from __future__ import annotations
 
 import json
+import os
 from pathlib import Path
 
 import pytest
 import typer
 
 
-
 class TestWorkflowJsonOutput:
     """Test the --json machine-readable output for run/resume/status."""
 
@@ -95,6 +95,36 @@ def test_status_json_single_and_list(self, project_dir):
         )
         assert any(r["run_id"] == rid for r in listing["runs"])
 
+    @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
+    @pytest.mark.parametrize("symlink_target", ["run-directory", "state.json"])
+    def test_status_json_list_rejects_symlinked_storage(
+        self, project_dir, tmp_path, symlink_target
+    ):
+        runs_dir = project_dir / ".specify" / "workflows" / "runs"
+        runs_dir.mkdir(parents=True)
+        outside = tmp_path / "outside"
+        outside.mkdir()
+        (outside / "state.json").write_text(
+            json.dumps({
+                "run_id": "run-1",
+                "workflow_id": "workflow",
+                "status": "paused",
+            }),
+            encoding="utf-8",
+        )
+        if symlink_target == "run-directory":
+            (runs_dir / "run-1").symlink_to(outside, target_is_directory=True)
+        else:
+            run_dir = runs_dir / "run-1"
+            run_dir.mkdir()
+            (run_dir / "state.json").symlink_to(outside / "state.json")
+
+        result = self._invoke(project_dir, ["workflow", "status", "--json"])
+
+        assert result.exit_code == 1
+        assert result.stdout == ""
+        assert "symlinked workflow run" in result.stderr
+
     def test_composed_gate_status_and_resume(self, project_dir):
         from specify_cli.workflows.catalog import WorkflowRegistry
 
diff --git a/tests/specify_cli/workflows/test_run_state_storage.py b/tests/specify_cli/workflows/test_run_state_storage.py
index 9b97a65a27..a36d540804 100644
--- a/tests/specify_cli/workflows/test_run_state_storage.py
+++ b/tests/specify_cli/workflows/test_run_state_storage.py
@@ -101,3 +101,41 @@ def test_workflow_snapshot_artifact_is_checked(project_dir, tmp_path):
 
     with pytest.raises(ValueError, match="symlinked workflow run artifact"):
         RunState._run_artifact(project_dir, "run-1", "workflow.yml")
+
+
+def test_execute_rejects_symlinked_workflow_snapshot(project_dir, tmp_path):
+    from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
+
+    run_dir = _run_dir(project_dir)
+    target = tmp_path / "workflow.yml"
+    target.write_text("external workflow", encoding="utf-8")
+    (run_dir / "workflow.yml").symlink_to(target)
+
+    definition = WorkflowDefinition({
+        "workflow": {"id": "workflow", "name": "Workflow"},
+        "steps": [],
+    })
+    with pytest.raises(ValueError, match="symlinked workflow run artifact"):
+        WorkflowEngine(project_dir).execute(definition, run_id="run-1")
+
+    assert target.read_text(encoding="utf-8") == "external workflow"
+
+
+def test_resume_rejects_symlinked_workflow_snapshot(project_dir, tmp_path):
+    from specify_cli.workflows.engine import WorkflowEngine
+
+    state = _state(project_dir)
+    state.save()
+    run_dir = _run_dir(project_dir)
+    target = tmp_path / "workflow.yml"
+    target.write_text(
+        "workflow: {id: workflow, name: Workflow}\n"
+        "steps:\n  - {id: review, type: gate, message: Review}\n",
+        encoding="utf-8",
+    )
+    (run_dir / "workflow.yml").symlink_to(target)
+
+    with pytest.raises(ValueError, match="symlinked workflow run artifact"):
+        WorkflowEngine(project_dir).resume("run-1")
+
+    assert target.read_text(encoding="utf-8").startswith("workflow:")

From bd3507f4b40586779e4bfe1d1f3e45941bf45838 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Mon, 12 Oct 2026 20:20:41 +0000
Subject: [PATCH 3/8] Secure workflow artifacts against run-directory swaps

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>
---
 src/specify_cli/workflows/engine.py           | 206 ++++++++++++------
 .../workflows/test_run_state_storage.py       |  77 +++++++
 tests/test_workflows.py                       |  17 +-
 3 files changed, 225 insertions(+), 75 deletions(-)

diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index 0393ca7f2a..789531f718 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -13,9 +13,9 @@
 import json
 import os
 import re
-import tempfile
 import threading
 import uuid
+from contextlib import contextmanager
 from contextvars import ContextVar
 from datetime import datetime, timezone
 from pathlib import Path
@@ -680,9 +680,38 @@ def _run_artifact(
         cls._reject_symlink(artifact, f"workflow run artifact {name!r}")
         return artifact
 
+    @classmethod
+    @contextmanager
+    def _open_run_directory(cls, project_root: Path, run_id: str, *, create=False):
+        """Open a run directory without following a replacement symlink."""
+        run_path = cls._run_directory(project_root, run_id, create=create)
+        directory_flag = getattr(os, "O_DIRECTORY", None)
+        nofollow = getattr(os, "O_NOFOLLOW", None)
+        if (
+            directory_flag is None
+            or nofollow is None
+            or os.open not in os.supports_dir_fd
+        ):
+            raise OSError("Secure workflow run-directory access is unavailable")
+
+        runs_path = run_path.parent
+        runs_fd = os.open(runs_path, os.O_RDONLY | directory_flag | nofollow)
+        try:
+            run_fd = os.open(
+                run_id,
+                os.O_RDONLY | directory_flag | nofollow,
+                dir_fd=runs_fd,
+            )
+        finally:
+            os.close(runs_fd)
+        try:
+            yield run_fd
+        finally:
+            os.close(run_fd)
+
     @staticmethod
-    def _open_artifact(path: Path, mode: str, **kwargs):
-        """Open a run artifact without following a symlink at the leaf."""
+    def _open_artifact(path: Path, mode: str, *, dir_fd: int, **kwargs):
+        """Open a run artifact relative to its safely opened run directory."""
         flags = {
             "r": os.O_RDONLY,
             "a": os.O_WRONLY | os.O_CREAT | os.O_APPEND,
@@ -690,8 +719,10 @@ def _open_artifact(path: Path, mode: str, **kwargs):
         }.get(mode)
         if flags is None:
             raise ValueError(f"Unsupported workflow artifact open mode: {mode!r}")
-        nofollow = getattr(os, "O_NOFOLLOW", 0)
-        fd = os.open(path, flags | nofollow, 0o666)
+        nofollow = getattr(os, "O_NOFOLLOW", None)
+        if nofollow is None or os.open not in os.supports_dir_fd:
+            raise OSError("Secure workflow artifact access is unavailable")
+        fd = os.open(path.name, flags | nofollow, 0o666, dir_fd=dir_fd)
         return os.fdopen(fd, mode, **kwargs)
 
     @staticmethod
@@ -829,18 +860,19 @@ def save(self) -> None:
             if self.execution is not None:
                 state_data["execution"] = self.execution
             try:
-                self._atomic_write_json(
-                    self._run_artifact(
-                        self.project_root, self.run_id, "state.json", create=True
-                    ),
-                    state_data,
+                state_path = self._run_artifact(
+                    self.project_root, self.run_id, "state.json", create=True
                 )
-                self._atomic_write_json(
-                    self._run_artifact(
-                        self.project_root, self.run_id, "inputs.json", create=True
-                    ),
-                    {"inputs": self.inputs},
+                inputs_path = self._run_artifact(
+                    self.project_root, self.run_id, "inputs.json", create=True
                 )
+                with self._open_run_directory(
+                    self.project_root, self.run_id, create=True
+                ) as run_fd:
+                    self._atomic_write_json(state_path, state_data, dir_fd=run_fd)
+                    self._atomic_write_json(
+                        inputs_path, {"inputs": self.inputs}, dir_fd=run_fd
+                    )
             except Exception as exc:
                 # Only write failures poison the instance. A graceful interrupt
                 # (KeyboardInterrupt) propagates unchanged: during step
@@ -854,11 +886,16 @@ def save(self) -> None:
                 raise CheckpointError(str(exc)) from exc
 
     @staticmethod
-    def _atomic_write_json(path: Path, data: dict[str, Any]) -> None:
-        """Write *data* as indented JSON to *path* atomically (temp + ``os.replace``)."""
-        RunState._reject_symlink(path, f"workflow run artifact {path.name!r}")
-        fd, tmp = tempfile.mkstemp(
-            dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp"
+    def _atomic_write_json(path: Path, data: dict[str, Any], *, dir_fd: int) -> None:
+        """Write *data* as indented JSON to *path* atomically."""
+        if os.rename not in os.supports_dir_fd or os.unlink not in os.supports_dir_fd:
+            raise OSError("Secure workflow artifact replacement is unavailable")
+        tmp = f".{path.name}.{uuid.uuid4().hex}.tmp"
+        fd = os.open(
+            tmp,
+            os.O_WRONLY | os.O_CREAT | os.O_EXCL,
+            0o600,
+            dir_fd=dir_fd,
         )
         try:
             # Keep non-ASCII text readable in these human-auditable records;
@@ -867,10 +904,12 @@ def _atomic_write_json(path: Path, data: dict[str, Any]) -> None:
                 fd, "w", encoding="utf-8", errors="backslashreplace"
             ) as f:
                 json.dump(data, f, indent=2, ensure_ascii=False)
-            os.replace(tmp, path)
+            os.rename(
+                tmp, path.name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd
+            )
         except BaseException:
             try:
-                os.unlink(tmp)
+                os.unlink(tmp, dir_fd=dir_fd)
             except OSError:
                 pass
             raise
@@ -893,8 +932,11 @@ def load(cls, run_id: str, project_root: Path) -> RunState:
         state_path = cls._run_artifact(project_root, run_id, "state.json")
 
         try:
-            with cls._open_artifact(state_path, "r", encoding="utf-8") as f:
-                state_data = json.load(f)
+            with cls._open_run_directory(project_root, run_id) as run_fd:
+                with cls._open_artifact(
+                    state_path, "r", dir_fd=run_fd, encoding="utf-8"
+                ) as f:
+                    state_data = json.load(f)
         except FileNotFoundError:
             msg = f"Run state not found: {state_path}"
             raise FileNotFoundError(msg)
@@ -985,20 +1027,27 @@ def load(cls, run_id: str, project_root: Path) -> RunState:
 
             validate_execution(state.execution)
 
-        inputs_path = cls._run_artifact(project_root, run_id, "inputs.json")
-        if "inputs" not in state_data and inputs_path.exists():
-            with cls._open_artifact(inputs_path, "r", encoding="utf-8") as f:
-                inputs_data = json.load(f)
-            if not isinstance(inputs_data, dict):
-                raise ValueError(
-                    "Invalid run inputs: expected a JSON object"
-                )
-            inputs = inputs_data.get("inputs", {})
-            if not isinstance(inputs, dict):
-                raise ValueError(
-                    "Invalid run inputs: 'inputs' must be a JSON object"
-                )
-            state.inputs = inputs
+        if "inputs" not in state_data:
+            inputs_path = cls._run_artifact(project_root, run_id, "inputs.json")
+            try:
+                with cls._open_run_directory(project_root, run_id) as run_fd:
+                    with cls._open_artifact(
+                        inputs_path, "r", dir_fd=run_fd, encoding="utf-8"
+                    ) as f:
+                        inputs_data = json.load(f)
+            except FileNotFoundError:
+                inputs_data = None
+            if inputs_data is not None:
+                if not isinstance(inputs_data, dict):
+                    raise ValueError(
+                        "Invalid run inputs: expected a JSON object"
+                    )
+                inputs = inputs_data.get("inputs", {})
+                if not isinstance(inputs, dict):
+                    raise ValueError(
+                        "Invalid run inputs: 'inputs' must be a JSON object"
+                    )
+                state.inputs = inputs
 
         if "inputs" in state_data:
             if not isinstance(state_data["inputs"], dict):
@@ -1024,15 +1073,20 @@ def append_log(self, entry: dict[str, Any]) -> None:
                 raise CheckpointError("A previous checkpoint failed; reload the run")
             with self._log_lock:
                 self.log_entries.append(entry)
-                with self._open_artifact(
-                    self._run_artifact(
-                        self.project_root, self.run_id, "log.jsonl", create=True
-                    ),
-                    "a",
-                    encoding="utf-8",
-                    errors="backslashreplace",
-                ) as f:
-                    f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+                log_path = self._run_artifact(
+                    self.project_root, self.run_id, "log.jsonl", create=True
+                )
+                with self._open_run_directory(
+                    self.project_root, self.run_id, create=True
+                ) as run_fd:
+                    with self._open_artifact(
+                        log_path,
+                        "a",
+                        dir_fd=run_fd,
+                        encoding="utf-8",
+                        errors="backslashreplace",
+                    ) as f:
+                        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
 
 
 # -- Workflow Engine ------------------------------------------------------
@@ -1180,8 +1234,13 @@ def execute(
             self.project_root, state.run_id, "workflow.yml", create=True
         )
         import yaml
-        with RunState._open_artifact(workflow_copy, "w", encoding="utf-8") as f:
-            yaml.safe_dump(definition.data, f, sort_keys=False, allow_unicode=True)
+        with RunState._open_run_directory(
+            self.project_root, state.run_id, create=True
+        ) as run_fd:
+            with RunState._open_artifact(
+                workflow_copy, "w", dir_fd=run_fd, encoding="utf-8"
+            ) as f:
+                yaml.safe_dump(definition.data, f, sort_keys=False, allow_unicode=True)
 
         # Resolve inputs
         resolved_inputs = self._resolve_inputs(definition, inputs or {})
@@ -1251,13 +1310,19 @@ def resume(
         run_copy = RunState._run_artifact(
             self.project_root, run_id, "workflow.yml"
         )
-        if run_copy.exists():
-            with RunState._open_artifact(run_copy, "r", encoding="utf-8") as f:
-                try:
-                    data = yaml.safe_load(f)
-                except yaml.YAMLError as exc:
-                    msg = f"Invalid YAML in {run_copy}: {exc}"
-                    raise ValueError(msg) from exc
+        try:
+            with RunState._open_run_directory(self.project_root, run_id) as run_fd:
+                with RunState._open_artifact(
+                    run_copy, "r", dir_fd=run_fd, encoding="utf-8"
+                ) as f:
+                    try:
+                        data = yaml.safe_load(f)
+                    except yaml.YAMLError as exc:
+                        msg = f"Invalid YAML in {run_copy}: {exc}"
+                        raise ValueError(msg) from exc
+        except FileNotFoundError:
+            definition = self.load_workflow(state.workflow_id)
+        else:
             if not isinstance(data, dict):
                 msg = (
                     "Workflow YAML must be a mapping, "
@@ -1265,8 +1330,6 @@ def resume(
                 )
                 raise ValueError(msg)
             definition = WorkflowDefinition(data, source_path=run_copy)
-        else:
-            definition = self.load_workflow(state.workflow_id)
 
         # RunState.load() rejects a non-int/negative current_step_index but
         # can't check the upper bound — the step count isn't known until the
@@ -1563,7 +1626,7 @@ def list_runs(self) -> list[dict[str, Any]]:
         runs_dir = self.project_root / ".specify" / "workflows" / "runs"
         if not runs_dir.exists():
             return []
-        self._reject_symlink(runs_dir, ".specify/workflows/runs")
+        RunState._reject_symlink(runs_dir, ".specify/workflows/runs")
 
         runs: list[dict[str, Any]] = []
         for run_dir in sorted(runs_dir.iterdir()):
@@ -1573,18 +1636,25 @@ def list_runs(self) -> list[dict[str, Any]]:
                 )
             if not run_dir.is_dir():
                 continue
-            state_path = self._run_artifact(
+            state_path = RunState._run_artifact(
                 self.project_root, run_dir.name, "state.json"
             )
-            if state_path.exists():
-                try:
-                    with self._open_artifact(state_path, "r", encoding="utf-8") as f:
+            try:
+                with RunState._open_run_directory(
+                    self.project_root, run_dir.name
+                ) as run_fd:
+                    with RunState._open_artifact(
+                        state_path,
+                        "r",
+                        dir_fd=run_fd,
+                        encoding="utf-8",
+                    ) as f:
                         state_data = json.load(f)
-                except (json.JSONDecodeError, OSError, UnicodeDecodeError):
-                    continue
-                if not isinstance(state_data, dict) or "run_id" not in state_data:
-                    continue
-                runs.append(state_data)
+            except (json.JSONDecodeError, OSError, UnicodeDecodeError):
+                continue
+            if not isinstance(state_data, dict) or "run_id" not in state_data:
+                continue
+            runs.append(state_data)
         return runs
 
 
diff --git a/tests/specify_cli/workflows/test_run_state_storage.py b/tests/specify_cli/workflows/test_run_state_storage.py
index a36d540804..9ea3a41563 100644
--- a/tests/specify_cli/workflows/test_run_state_storage.py
+++ b/tests/specify_cli/workflows/test_run_state_storage.py
@@ -43,6 +43,83 @@ def test_load_rejects_symlinked_run_directory(project_dir, tmp_path):
         RunState.load("run-1", project_dir)
 
 
+def test_load_keeps_open_run_directory_when_path_is_replaced(
+    project_dir, tmp_path, monkeypatch
+):
+    from specify_cli.workflows.engine import RunState
+
+    state = _state(project_dir)
+    state.save()
+    run_dir = project_dir / ".specify" / "workflows" / "runs" / "run-1"
+    detached_dir = tmp_path / "detached-run"
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    outside_state = outside / "state.json"
+    outside_state.write_text(
+        json.dumps({
+            "run_id": "run-1",
+            "workflow_id": "external",
+            "status": "paused",
+        }),
+        encoding="utf-8",
+    )
+    original_open = RunState._open_artifact
+    swapped = False
+
+    def _replace_run_directory(path, mode, *, dir_fd, **kwargs):
+        nonlocal swapped
+        if path.name == "state.json" and not swapped:
+            swapped = True
+            os.rename(run_dir, detached_dir)
+            run_dir.symlink_to(outside, target_is_directory=True)
+        return original_open(path, mode, dir_fd=dir_fd, **kwargs)
+
+    monkeypatch.setattr(
+        RunState, "_open_artifact", staticmethod(_replace_run_directory)
+    )
+
+    loaded = RunState.load("run-1", project_dir)
+
+    assert loaded.workflow_id == "workflow"
+    assert json.loads(outside_state.read_text(encoding="utf-8"))["workflow_id"] == "external"
+
+
+def test_save_keeps_open_run_directory_when_path_is_replaced(
+    project_dir, tmp_path, monkeypatch
+):
+    from specify_cli.workflows.engine import RunState
+
+    state = _state(project_dir)
+    state.save()
+    run_dir = project_dir / ".specify" / "workflows" / "runs" / "run-1"
+    detached_dir = tmp_path / "detached-run"
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    outside_state = outside / "state.json"
+    outside_state.write_text("external state", encoding="utf-8")
+    original_write = RunState._atomic_write_json
+    swapped = False
+
+    def _replace_run_directory(path, data, *, dir_fd):
+        nonlocal swapped
+        if path.name == "state.json" and not swapped:
+            swapped = True
+            os.rename(run_dir, detached_dir)
+            run_dir.symlink_to(outside, target_is_directory=True)
+        return original_write(path, data, dir_fd=dir_fd)
+
+    monkeypatch.setattr(
+        RunState, "_atomic_write_json", staticmethod(_replace_run_directory)
+    )
+
+    state.save()
+
+    assert outside_state.read_text(encoding="utf-8") == "external state"
+    assert json.loads((detached_dir / "state.json").read_text(encoding="utf-8"))[
+        "run_id"
+    ] == "run-1"
+
+
 @pytest.mark.parametrize("artifact", ["state.json", "inputs.json"])
 def test_run_state_rejects_symlinked_persisted_artifacts(
     project_dir, tmp_path, artifact
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index a4ea5d1786..84c9b9b420 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -9091,8 +9091,11 @@ def test_list_skips_invalid_utf8_with_valid_sibling(self, project_dir):
         assert runs[0]["workflow_id"] == "good-run-utf8"
 
     def test_list_skips_oserror_with_valid_sibling(self, project_dir, monkeypatch):
-        import builtins
-        from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition
+        from specify_cli.workflows.engine import (
+            RunState,
+            WorkflowDefinition,
+            WorkflowEngine,
+        )
 
         runs_dir = project_dir / ".specify" / "workflows" / "runs"
         bad_dir = runs_dir / "bad-oserror"
@@ -9100,14 +9103,14 @@ def test_list_skips_oserror_with_valid_sibling(self, project_dir, monkeypatch):
         state_file = bad_dir / "state.json"
         state_file.write_text('{"run_id": "bad"}', encoding="utf-8")
 
-        original_open = builtins.open
+        original_open = RunState._open_artifact
 
-        def _mock_open(path, *args, **kwargs):
-            if str(path).endswith("state.json") and "bad-oserror" in str(path):
+        def _mock_open(path, mode, *, dir_fd, **kwargs):
+            if path.name == "state.json" and "bad-oserror" in str(path):
                 raise OSError("permission denied")
-            return original_open(path, *args, **kwargs)
+            return original_open(path, mode, dir_fd=dir_fd, **kwargs)
 
-        monkeypatch.setattr(builtins, "open", _mock_open)
+        monkeypatch.setattr(RunState, "_open_artifact", staticmethod(_mock_open))
 
         yaml_str = """
 schema_version: "1.0"

From 02159bf0928ed1b37ae351396004b885beb3629c Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Mon, 12 Oct 2026 21:03:30 +0000
Subject: [PATCH 4/8] Support workflow persistence without dir-fd APIs

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>
---
 src/specify_cli/workflows/engine.py           | 64 ++++++++++++++----
 .../workflows/test_run_state_storage.py       | 65 +++++++++++++++++++
 workflows/ARCHITECTURE.md                     |  6 +-
 3 files changed, 121 insertions(+), 14 deletions(-)

diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index 789531f718..eb4f910ef2 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -669,6 +669,24 @@ def _reject_symlink(path: Path, label: str) -> None:
         if path.is_symlink():
             raise ValueError(f"Refusing to use symlinked {label} path: {path}")
 
+    @staticmethod
+    def _supports_descriptor_access() -> bool:
+        supports_dir_fd = getattr(os, "supports_dir_fd", ())
+        return (
+            hasattr(os, "O_DIRECTORY")
+            and hasattr(os, "O_NOFOLLOW")
+            and os.open in supports_dir_fd
+            and os.rename in supports_dir_fd
+            and os.unlink in supports_dir_fd
+        )
+
+    @classmethod
+    def _validate_fallback_path(cls, path: Path) -> None:
+        """Recheck persisted-run components before path-based access."""
+        cls._reject_symlink(path.parent.parent, ".specify/workflows/runs")
+        cls._reject_symlink(path.parent, "workflow run")
+        cls._reject_symlink(path, f"workflow run artifact {path.name!r}")
+
     @classmethod
     def _run_artifact(
         cls, project_root: Path, run_id: str, name: str, *, create: bool = False
@@ -685,15 +703,12 @@ def _run_artifact(
     def _open_run_directory(cls, project_root: Path, run_id: str, *, create=False):
         """Open a run directory without following a replacement symlink."""
         run_path = cls._run_directory(project_root, run_id, create=create)
-        directory_flag = getattr(os, "O_DIRECTORY", None)
-        nofollow = getattr(os, "O_NOFOLLOW", None)
-        if (
-            directory_flag is None
-            or nofollow is None
-            or os.open not in os.supports_dir_fd
-        ):
-            raise OSError("Secure workflow run-directory access is unavailable")
+        if not cls._supports_descriptor_access():
+            yield None
+            return
 
+        directory_flag = os.O_DIRECTORY
+        nofollow = os.O_NOFOLLOW
         runs_path = run_path.parent
         runs_fd = os.open(runs_path, os.O_RDONLY | directory_flag | nofollow)
         try:
@@ -710,8 +725,14 @@ def _open_run_directory(cls, project_root: Path, run_id: str, *, create=False):
             os.close(run_fd)
 
     @staticmethod
-    def _open_artifact(path: Path, mode: str, *, dir_fd: int, **kwargs):
+    def _open_artifact(
+        path: Path, mode: str, *, dir_fd: int | None = None, **kwargs
+    ):
         """Open a run artifact relative to its safely opened run directory."""
+        if dir_fd is None:
+            RunState._validate_fallback_path(path)
+            return open(path, mode, **kwargs)
+
         flags = {
             "r": os.O_RDONLY,
             "a": os.O_WRONLY | os.O_CREAT | os.O_APPEND,
@@ -886,11 +907,30 @@ def save(self) -> None:
                 raise CheckpointError(str(exc)) from exc
 
     @staticmethod
-    def _atomic_write_json(path: Path, data: dict[str, Any], *, dir_fd: int) -> None:
+    def _atomic_write_json(
+        path: Path, data: dict[str, Any], *, dir_fd: int | None = None
+    ) -> None:
         """Write *data* as indented JSON to *path* atomically."""
-        if os.rename not in os.supports_dir_fd or os.unlink not in os.supports_dir_fd:
-            raise OSError("Secure workflow artifact replacement is unavailable")
         tmp = f".{path.name}.{uuid.uuid4().hex}.tmp"
+        if dir_fd is None:
+            tmp_path = path.with_name(tmp)
+            RunState._validate_fallback_path(path)
+            try:
+                with tmp_path.open(
+                    "x", encoding="utf-8", errors="backslashreplace"
+                ) as f:
+                    json.dump(data, f, indent=2, ensure_ascii=False)
+                RunState._validate_fallback_path(path)
+                os.replace(tmp_path, path)
+            except BaseException:
+                try:
+                    RunState._validate_fallback_path(tmp_path)
+                    tmp_path.unlink()
+                except (OSError, ValueError):
+                    pass
+                raise
+            return
+
         fd = os.open(
             tmp,
             os.O_WRONLY | os.O_CREAT | os.O_EXCL,
diff --git a/tests/specify_cli/workflows/test_run_state_storage.py b/tests/specify_cli/workflows/test_run_state_storage.py
index 9ea3a41563..203b2e9156 100644
--- a/tests/specify_cli/workflows/test_run_state_storage.py
+++ b/tests/specify_cli/workflows/test_run_state_storage.py
@@ -26,6 +26,71 @@ def _state(project_dir, run_id="run-1"):
     return state
 
 
+def test_workflow_operations_use_portable_storage_fallback(
+    project_dir, monkeypatch
+):
+    from specify_cli.workflows.base import RunStatus
+    from specify_cli.workflows.engine import (
+        RunState,
+        WorkflowDefinition,
+        WorkflowEngine,
+    )
+
+    monkeypatch.setattr(
+        RunState, "_supports_descriptor_access", staticmethod(lambda: False)
+    )
+    definition = WorkflowDefinition.from_string(
+        """
+schema_version: "1.0"
+workflow:
+  id: portable-storage
+  name: Portable Storage
+inputs:
+  verdict:
+    type: string
+    default: ""
+steps:
+  - id: review
+    type: gate
+    message: Review
+    verdict_input: verdict
+"""
+    )
+    engine = WorkflowEngine(project_dir)
+
+    state = engine.execute(definition, run_id="portable")
+
+    assert state.status == RunStatus.PAUSED
+    resumed = engine.resume("portable", {"verdict": "approve"})
+    assert resumed.status == RunStatus.COMPLETED
+    assert engine.list_runs()[0]["run_id"] == "portable"
+
+
+def test_portable_storage_fallback_rechecks_artifact_symlinks(
+    project_dir, tmp_path, monkeypatch
+):
+    from specify_cli.workflows.engine import RunState
+
+    monkeypatch.setattr(
+        RunState, "_supports_descriptor_access", staticmethod(lambda: False)
+    )
+    state = _state(project_dir)
+    state.save()
+    artifact = RunState._run_artifact(project_dir, "run-1", "state.json")
+    external = tmp_path / "state.json"
+    external.write_text("external state", encoding="utf-8")
+
+    with RunState._open_run_directory(project_dir, "run-1") as run_fd:
+        artifact.unlink()
+        artifact.symlink_to(external)
+        with pytest.raises(ValueError, match="symlinked workflow run artifact"):
+            RunState._open_artifact(
+                artifact, "r", dir_fd=run_fd, encoding="utf-8"
+            )
+
+    assert external.read_text(encoding="utf-8") == "external state"
+
+
 def test_load_rejects_symlinked_run_directory(project_dir, tmp_path):
     from specify_cli.workflows.engine import RunState
 
diff --git a/workflows/ARCHITECTURE.md b/workflows/ARCHITECTURE.md
index 28b7f9043a..3e2266e9f2 100644
--- a/workflows/ARCHITECTURE.md
+++ b/workflows/ARCHITECTURE.md
@@ -286,8 +286,10 @@ The `.specify/workflows/runs` tree is a project-owned persistence boundary.
 Every run directory and persisted artifact below it must be a real filesystem
 object rather than a symlink. `RunState` enforces this policy before reading or
 writing `state.json`, `inputs.json`, `workflow.yml`, or `log.jsonl`; artifact
-opens also use no-follow flags where the platform provides them. Commands
-fail closed instead of following redirected run storage.
+opens also use no-follow flags and directory descriptors where supported.
+Other platforms recheck path components before access and use atomic replacement
+for state writes; that fallback rejects existing symlinks but cannot prevent
+concurrent path replacement.
 
 ## Module Structure
 

From 88dee954d2076723af612e385cfafe9a2243ea06 Mon Sep 17 00:00:00 2001
From: Manfred Riem <15701806+mnriem@users.noreply.github.com>
Date: Tue, 13 Oct 2026 09:34:06 -0500
Subject: [PATCH 5/8] fix(workflows): address storage review findings and
 checkpoint tests

Close artifact handles in the symlink regression and explain cleanup failure handling. Forward directory descriptors through checkpoint test mocks to preserve failure and interrupt coverage.

Assisted-by: GitHub Copilot (model: GPT-6.1 Sol, autonomous)
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 67a8a13d-1542-470f-87b8-5a3a1dc122f5
---
 src/specify_cli/workflows/engine.py            |  1 +
 .../workflows/test_run_state_storage.py        |  5 +++--
 tests/workflows/test_composition_execution.py  | 18 +++++++++---------
 3 files changed, 13 insertions(+), 11 deletions(-)

diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index eb4f910ef2..10812395dc 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -927,6 +927,7 @@ def _atomic_write_json(
                     RunState._validate_fallback_path(tmp_path)
                     tmp_path.unlink()
                 except (OSError, ValueError):
+                    # Preserve the original write failure if cleanup is unsafe or fails.
                     pass
                 raise
             return
diff --git a/tests/specify_cli/workflows/test_run_state_storage.py b/tests/specify_cli/workflows/test_run_state_storage.py
index 203b2e9156..dbb71035d1 100644
--- a/tests/specify_cli/workflows/test_run_state_storage.py
+++ b/tests/specify_cli/workflows/test_run_state_storage.py
@@ -84,9 +84,10 @@ def test_portable_storage_fallback_rechecks_artifact_symlinks(
         artifact.unlink()
         artifact.symlink_to(external)
         with pytest.raises(ValueError, match="symlinked workflow run artifact"):
-            RunState._open_artifact(
+            with RunState._open_artifact(
                 artifact, "r", dir_fd=run_fd, encoding="utf-8"
-            )
+            ):
+                pass
 
     assert external.read_text(encoding="utf-8") == "external state"
 
diff --git a/tests/workflows/test_composition_execution.py b/tests/workflows/test_composition_execution.py
index 33c746c498..a4f3a8c385 100644
--- a/tests/workflows/test_composition_execution.py
+++ b/tests/workflows/test_composition_execution.py
@@ -669,15 +669,15 @@ def test_checkpoint_failure_leaves_running_run_not_resumable(
     original = RunState._atomic_write_json
     failed = False
 
-    def write(path, data):
+    def write(path, data, *, dir_fd=None):
         nonlocal failed
         record = data.get("step_results", {}).get("work")
         if path.name == "state.json" and record and not failed:
             failed = True
             if failure_after_replace:
-                original(path, data)
+                original(path, data, dir_fd=dir_fd)
             raise OSError("checkpoint failure")
-        original(path, data)
+        original(path, data, dir_fd=dir_fd)
 
     monkeypatch.setattr(RunState, "_atomic_write_json", staticmethod(write))
     with pytest.raises(CheckpointError, match="checkpoint failure"):
@@ -702,11 +702,11 @@ def _interrupt_state_write(monkeypatch, when):
     original = RunState._atomic_write_json
     fired = []
 
-    def write(path, data):
+    def write(path, data, *, dir_fd=None):
         if path.name == "state.json" and not fired and when(data):
             fired.append(path)
             raise KeyboardInterrupt
-        original(path, data)
+        original(path, data, dir_fd=dir_fd)
 
     monkeypatch.setattr(RunState, "_atomic_write_json", staticmethod(write))
     return fired
@@ -847,10 +847,10 @@ def notify(self, operation, occurrence, **kwargs):
             assert second_failed.wait(5)
         return original_notify(self, operation, occurrence, **kwargs)
 
-    def write(path, data):
+    def write(path, data, *, dir_fd=None):
         if items.get(threading.current_thread()) == 2:
             raise OSError("disk full")
-        original_write(path, data)
+        original_write(path, data, dir_fd=dir_fd)
 
     monkeypatch.setitem(STEP_REGISTRY, "sync", Sync())
     monkeypatch.setattr(execution.Execution, "transition", transition)
@@ -3428,10 +3428,10 @@ def test_workflow_call_start_checkpoint_failure_logs_no_start(tmp_path, monkeypa
     install(tmp_path, definition("child", [{"id": "inner", "type": "probe"}]))
     original = RunState._atomic_write_json
 
-    def write(path, data):
+    def write(path, data, *, dir_fd=None):
         if path.name == "state.json" and data.get("current_step_id") == "call":
             raise OSError("checkpoint failure")
-        original(path, data)
+        original(path, data, dir_fd=dir_fd)
 
     monkeypatch.setattr(RunState, "_atomic_write_json", staticmethod(write))
     callbacks = []

From f0743d6d374aa5a4893105f8db83f0474762ea7c Mon Sep 17 00:00:00 2001
From: Manfred Riem <15701806+mnriem@users.noreply.github.com>
Date: Tue, 13 Oct 2026 11:38:22 -0500
Subject: [PATCH 6/8] fix(workflows): cover portable storage race rejection in
 tests

Exercise native and forced portable storage access in run-directory replacement regressions. Expect safe rejection without directory descriptors and assert that external artifacts remain untouched.

Assisted-by: GitHub Copilot (model: GPT-6.1 Sol, autonomous)
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 67a8a13d-1542-470f-87b8-5a3a1dc122f5
---
 .../workflows/test_run_state_storage.py       | 42 ++++++++++++++++---
 1 file changed, 37 insertions(+), 5 deletions(-)

diff --git a/tests/specify_cli/workflows/test_run_state_storage.py b/tests/specify_cli/workflows/test_run_state_storage.py
index dbb71035d1..649d43aa2f 100644
--- a/tests/specify_cli/workflows/test_run_state_storage.py
+++ b/tests/specify_cli/workflows/test_run_state_storage.py
@@ -109,11 +109,17 @@ def test_load_rejects_symlinked_run_directory(project_dir, tmp_path):
         RunState.load("run-1", project_dir)
 
 
+@pytest.mark.parametrize("portable", [False, True])
 def test_load_keeps_open_run_directory_when_path_is_replaced(
-    project_dir, tmp_path, monkeypatch
+    project_dir, tmp_path, monkeypatch, portable
 ):
     from specify_cli.workflows.engine import RunState
 
+    if portable:
+        monkeypatch.setattr(
+            RunState, "_supports_descriptor_access", staticmethod(lambda: False)
+        )
+    descriptor_access = RunState._supports_descriptor_access()
     state = _state(project_dir)
     state.save()
     run_dir = project_dir / ".specify" / "workflows" / "runs" / "run-1"
@@ -144,17 +150,29 @@ def _replace_run_directory(path, mode, *, dir_fd, **kwargs):
         RunState, "_open_artifact", staticmethod(_replace_run_directory)
     )
 
-    loaded = RunState.load("run-1", project_dir)
+    if descriptor_access:
+        loaded = RunState.load("run-1", project_dir)
+        assert loaded.workflow_id == "workflow"
+    else:
+        with pytest.raises(ValueError, match="symlinked workflow run path"):
+            RunState.load("run-1", project_dir)
 
-    assert loaded.workflow_id == "workflow"
+    assert swapped
     assert json.loads(outside_state.read_text(encoding="utf-8"))["workflow_id"] == "external"
 
 
+@pytest.mark.parametrize("portable", [False, True])
 def test_save_keeps_open_run_directory_when_path_is_replaced(
-    project_dir, tmp_path, monkeypatch
+    project_dir, tmp_path, monkeypatch, portable
 ):
+    from specify_cli.workflows._execution import CheckpointError
     from specify_cli.workflows.engine import RunState
 
+    if portable:
+        monkeypatch.setattr(
+            RunState, "_supports_descriptor_access", staticmethod(lambda: False)
+        )
+    descriptor_access = RunState._supports_descriptor_access()
     state = _state(project_dir)
     state.save()
     run_dir = project_dir / ".specify" / "workflows" / "runs" / "run-1"
@@ -178,12 +196,26 @@ def _replace_run_directory(path, data, *, dir_fd):
         RunState, "_atomic_write_json", staticmethod(_replace_run_directory)
     )
 
-    state.save()
+    state.inputs = {"changed": True}
+    if descriptor_access:
+        state.save()
+    else:
+        with pytest.raises(CheckpointError, match="symlinked workflow run path"):
+            state.save()
 
+    assert swapped
     assert outside_state.read_text(encoding="utf-8") == "external state"
+    assert not (outside / "inputs.json").exists()
     assert json.loads((detached_dir / "state.json").read_text(encoding="utf-8"))[
         "run_id"
     ] == "run-1"
+    expected_inputs = {"changed": True} if descriptor_access else {}
+    assert json.loads((detached_dir / "state.json").read_text(encoding="utf-8"))[
+        "inputs"
+    ] == expected_inputs
+    assert json.loads((detached_dir / "inputs.json").read_text(encoding="utf-8"))[
+        "inputs"
+    ] == expected_inputs
 
 
 @pytest.mark.parametrize("artifact", ["state.json", "inputs.json"])

From 446c4f165e99cdb7273668fead2965cdf7c29967 Mon Sep 17 00:00:00 2001
From: Manfred Riem <15701806+mnriem@users.noreply.github.com>
Date: Tue, 13 Oct 2026 12:45:44 -0500
Subject: [PATCH 7/8] fix(workflows): anchor run storage access to project
 directory

Walk storage ancestors using descriptor-relative no-follow opens and use directory descriptors for creation and listing. Reject dangling run-root symlinks and add ancestor replacement regressions while preserving the portable fallback.

Assisted-by: GitHub Copilot (model: GPT-6.1 Sol, autonomous)
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 67a8a13d-1542-470f-87b8-5a3a1dc122f5
---
 src/specify_cli/workflows/engine.py           | 78 +++++++++++----
 .../workflows/test_command_status.py          | 11 ++-
 .../workflows/test_run_state_storage.py       | 94 +++++++++++++++++++
 workflows/ARCHITECTURE.md                     |  4 +
 4 files changed, 167 insertions(+), 20 deletions(-)

diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index 10812395dc..244ddd3b91 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -648,9 +648,17 @@ def _run_directory(
         """Return a run directory after enforcing the persisted-run boundary."""
         cls._validate_run_id(run_id)
         runs_root = project_root / ".specify" / "workflows" / "runs"
+        cls._validate_runs_path(project_root)
         if create:
-            runs_root.mkdir(parents=True, exist_ok=True)
-        cls._reject_symlink(runs_root, ".specify/workflows/runs")
+            with cls._open_runs_directory(project_root, create=True) as runs_fd:
+                if runs_fd is None:
+                    (runs_root / run_id).mkdir(exist_ok=True)
+                else:
+                    try:
+                        os.mkdir(run_id, dir_fd=runs_fd)
+                    except FileExistsError:
+                        # Existing entries are validated before artifact access.
+                        pass
 
         run_dir = runs_root / run_id
         if run_dir.is_symlink():
@@ -659,8 +667,6 @@ def _run_directory(
             )
         if run_dir.exists() and not run_dir.is_dir():
             raise OSError(f"Workflow run path is not a directory: {run_dir}")
-        if create:
-            run_dir.mkdir(parents=True, exist_ok=True)
         return run_dir
 
     @staticmethod
@@ -678,12 +684,53 @@ def _supports_descriptor_access() -> bool:
             and os.open in supports_dir_fd
             and os.rename in supports_dir_fd
             and os.unlink in supports_dir_fd
+            and os.mkdir in supports_dir_fd
+            and os.listdir in os.supports_fd
         )
 
+    @classmethod
+    def _validate_runs_path(cls, project_root: Path) -> None:
+        path = project_root
+        for name in (".specify", "workflows", "runs"):
+            path = path / name
+            cls._reject_symlink(path, f"workflow storage {name!r}")
+
+    @classmethod
+    @contextmanager
+    def _open_runs_directory(cls, project_root: Path, *, create=False):
+        """Walk storage ancestors relative to the trusted project directory."""
+        cls._validate_runs_path(project_root)
+        runs_path = project_root / ".specify" / "workflows" / "runs"
+        if not cls._supports_descriptor_access():
+            if create:
+                runs_path.mkdir(parents=True, exist_ok=True)
+            cls._validate_runs_path(project_root)
+            yield None
+            return
+
+        flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
+        if create:
+            project_root.mkdir(parents=True, exist_ok=True)
+        fd = os.open(project_root, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            for name in (".specify", "workflows", "runs"):
+                if create:
+                    try:
+                        os.mkdir(name, dir_fd=fd)
+                    except FileExistsError:
+                        # An existing component must still pass the no-follow open.
+                        pass
+                child_fd = os.open(name, flags, dir_fd=fd)
+                os.close(fd)
+                fd = child_fd
+            yield fd
+        finally:
+            os.close(fd)
+
     @classmethod
     def _validate_fallback_path(cls, path: Path) -> None:
         """Recheck persisted-run components before path-based access."""
-        cls._reject_symlink(path.parent.parent, ".specify/workflows/runs")
+        cls._validate_runs_path(path.parents[4])
         cls._reject_symlink(path.parent, "workflow run")
         cls._reject_symlink(path, f"workflow run artifact {path.name!r}")
 
@@ -702,23 +749,17 @@ def _run_artifact(
     @contextmanager
     def _open_run_directory(cls, project_root: Path, run_id: str, *, create=False):
         """Open a run directory without following a replacement symlink."""
-        run_path = cls._run_directory(project_root, run_id, create=create)
+        cls._run_directory(project_root, run_id, create=create)
         if not cls._supports_descriptor_access():
             yield None
             return
 
-        directory_flag = os.O_DIRECTORY
-        nofollow = os.O_NOFOLLOW
-        runs_path = run_path.parent
-        runs_fd = os.open(runs_path, os.O_RDONLY | directory_flag | nofollow)
-        try:
+        with cls._open_runs_directory(project_root) as runs_fd:
             run_fd = os.open(
                 run_id,
-                os.O_RDONLY | directory_flag | nofollow,
+                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                 dir_fd=runs_fd,
             )
-        finally:
-            os.close(runs_fd)
         try:
             yield run_fd
         finally:
@@ -1665,12 +1706,15 @@ def _coerce_input(
     def list_runs(self) -> list[dict[str, Any]]:
         """List all workflow runs in the project."""
         runs_dir = self.project_root / ".specify" / "workflows" / "runs"
-        if not runs_dir.exists():
+        try:
+            with RunState._open_runs_directory(self.project_root) as runs_fd:
+                names = sorted(os.listdir(runs_dir if runs_fd is None else runs_fd))
+        except FileNotFoundError:
             return []
-        RunState._reject_symlink(runs_dir, ".specify/workflows/runs")
 
         runs: list[dict[str, Any]] = []
-        for run_dir in sorted(runs_dir.iterdir()):
+        for name in names:
+            run_dir = runs_dir / name
             if run_dir.is_symlink():
                 raise ValueError(
                     f"Refusing to use symlinked workflow run path: {run_dir}"
diff --git a/tests/specify_cli/workflows/test_command_status.py b/tests/specify_cli/workflows/test_command_status.py
index 7b25c38330..0c3dc61fe7 100644
--- a/tests/specify_cli/workflows/test_command_status.py
+++ b/tests/specify_cli/workflows/test_command_status.py
@@ -96,7 +96,9 @@ def test_status_json_single_and_list(self, project_dir):
         assert any(r["run_id"] == rid for r in listing["runs"])
 
     @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
-    @pytest.mark.parametrize("symlink_target", ["run-directory", "state.json"])
+    @pytest.mark.parametrize(
+        "symlink_target", ["run-directory", "state.json", "dangling-runs-root"]
+    )
     def test_status_json_list_rejects_symlinked_storage(
         self, project_dir, tmp_path, symlink_target
     ):
@@ -114,16 +116,19 @@ def test_status_json_list_rejects_symlinked_storage(
         )
         if symlink_target == "run-directory":
             (runs_dir / "run-1").symlink_to(outside, target_is_directory=True)
-        else:
+        elif symlink_target == "state.json":
             run_dir = runs_dir / "run-1"
             run_dir.mkdir()
             (run_dir / "state.json").symlink_to(outside / "state.json")
+        else:
+            runs_dir.rmdir()
+            runs_dir.symlink_to(tmp_path / "missing", target_is_directory=True)
 
         result = self._invoke(project_dir, ["workflow", "status", "--json"])
 
         assert result.exit_code == 1
         assert result.stdout == ""
-        assert "symlinked workflow run" in result.stderr
+        assert "symlinked" in result.stderr
 
     def test_composed_gate_status_and_resume(self, project_dir):
         from specify_cli.workflows.catalog import WorkflowRegistry
diff --git a/tests/specify_cli/workflows/test_run_state_storage.py b/tests/specify_cli/workflows/test_run_state_storage.py
index 649d43aa2f..b61b3c9df0 100644
--- a/tests/specify_cli/workflows/test_run_state_storage.py
+++ b/tests/specify_cli/workflows/test_run_state_storage.py
@@ -314,3 +314,97 @@ def test_resume_rejects_symlinked_workflow_snapshot(project_dir, tmp_path):
         WorkflowEngine(project_dir).resume("run-1")
 
     assert target.read_text(encoding="utf-8").startswith("workflow:")
+
+
+@pytest.mark.parametrize("ancestor", [".specify", "workflows"])
+@pytest.mark.parametrize("operation", ["load", "save", "list"])
+def test_ancestor_replacement_never_accesses_external_storage(
+    project_dir, tmp_path, monkeypatch, ancestor, operation
+):
+    from specify_cli.workflows._execution import CheckpointError
+    from specify_cli.workflows.engine import RunState, WorkflowEngine
+
+    if not RunState._supports_descriptor_access():
+        pytest.skip("directory descriptors are unavailable")
+
+    state = _state(project_dir)
+    state.save()
+    runs_root = project_dir / ".specify" / "workflows" / "runs"
+    ancestor_path = (
+        project_dir / ".specify" if ancestor == ".specify" else runs_root.parent
+    )
+    outside = tmp_path / "outside"
+    outside_run = (
+        outside / "workflows" / "runs" / "run-1"
+        if ancestor == ".specify"
+        else outside / "runs" / "run-1"
+    )
+    outside_run.mkdir(parents=True)
+    external_state = outside_run / "state.json"
+    external_state.write_text(
+        json.dumps({
+            "run_id": "run-1",
+            "workflow_id": "external",
+            "status": "paused",
+            "inputs": {},
+        }),
+        encoding="utf-8",
+    )
+    original_content = external_state.read_bytes()
+    original_open = os.open
+    original_artifact_open = RunState._open_artifact
+    artifact_attempts = []
+    swapped = False
+
+    def replace_ancestor(path, flags, *args, **kwargs):
+        nonlocal swapped
+        if not swapped and path in (project_dir, runs_root):
+            swapped = True
+            ancestor_path.rename(tmp_path / "detached")
+            ancestor_path.symlink_to(outside, target_is_directory=True)
+        return original_open(path, flags, *args, **kwargs)
+
+    def track_artifact_open(*args, **kwargs):
+        artifact_attempts.append(args[0])
+        return original_artifact_open(*args, **kwargs)
+
+    monkeypatch.setattr(
+        RunState, "_supports_descriptor_access", staticmethod(lambda: True)
+    )
+    monkeypatch.setattr(os, "open", replace_ancestor)
+    monkeypatch.setattr(
+        RunState, "_open_artifact", staticmethod(track_artifact_open)
+    )
+    expected = (
+        (CheckpointError, ValueError, OSError)
+        if operation == "save"
+        else (ValueError, OSError)
+    )
+    with pytest.raises(expected):
+        if operation == "load":
+            RunState.load("run-1", project_dir)
+        elif operation == "save":
+            state.save()
+        else:
+            WorkflowEngine(project_dir).list_runs()
+
+    assert swapped
+    assert not artifact_attempts
+    assert external_state.read_bytes() == original_content
+    assert sorted(path.name for path in outside_run.iterdir()) == ["state.json"]
+
+
+@pytest.mark.parametrize("portable", [False, True])
+def test_list_rejects_dangling_runs_root(project_dir, tmp_path, monkeypatch, portable):
+    from specify_cli.workflows.engine import RunState, WorkflowEngine
+
+    if portable:
+        monkeypatch.setattr(
+            RunState, "_supports_descriptor_access", staticmethod(lambda: False)
+        )
+    runs_root = project_dir / ".specify" / "workflows" / "runs"
+    runs_root.parent.mkdir(parents=True, exist_ok=True)
+    runs_root.symlink_to(tmp_path / "missing", target_is_directory=True)
+
+    with pytest.raises(ValueError, match="symlinked"):
+        WorkflowEngine(project_dir).list_runs()
diff --git a/workflows/ARCHITECTURE.md b/workflows/ARCHITECTURE.md
index 3e2266e9f2..d7a6f1a1d9 100644
--- a/workflows/ARCHITECTURE.md
+++ b/workflows/ARCHITECTURE.md
@@ -287,6 +287,10 @@ Every run directory and persisted artifact below it must be a real filesystem
 object rather than a symlink. `RunState` enforces this policy before reading or
 writing `state.json`, `inputs.json`, `workflow.yml`, or `log.jsonl`; artifact
 opens also use no-follow flags and directory descriptors where supported.
+The descriptor-backed path opens the project root and walks `.specify`,
+`workflows`, and `runs` one component at a time with no-follow directory opens.
+Directory creation and run enumeration use those descriptors too, preventing
+an ancestor replacement from redirecting access outside the project.
 Other platforms recheck path components before access and use atomic replacement
 for state writes; that fallback rejects existing symlinks but cannot prevent
 concurrent path replacement.

From a7fdbd8177c175e3ba1c4a4657fc633a812f80a6 Mon Sep 17 00:00:00 2001
From: Manfred Riem <15701806+mnriem@users.noreply.github.com>
Date: Tue, 13 Oct 2026 13:47:26 -0500
Subject: [PATCH 8/8] fix(workflows): preserve private permissions in portable
 storage

Create portable atomic-write temporary files with explicit 0600 permissions. Cover temporary and final state/input modes under a permissive POSIX umask, including replacement of existing files. Clarify that symlink rejection is enforced on accessed storage paths, not a proactive tree scan.

Assisted-by: GitHub Copilot (model: GPT-6.1 Sol, autonomous)
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 67a8a13d-1542-470f-87b8-5a3a1dc122f5
---
 src/specify_cli/workflows/engine.py           |  7 +++-
 .../workflows/test_run_state_storage.py       | 40 +++++++++++++++++++
 workflows/ARCHITECTURE.md                     |  8 ++--
 workflows/README.md                           | 11 ++---
 4 files changed, 56 insertions(+), 10 deletions(-)

diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index 244ddd3b91..97cc26b11b 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -957,8 +957,11 @@ def _atomic_write_json(
             tmp_path = path.with_name(tmp)
             RunState._validate_fallback_path(path)
             try:
-                with tmp_path.open(
-                    "x", encoding="utf-8", errors="backslashreplace"
+                fd = os.open(
+                    tmp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
+                )
+                with os.fdopen(
+                    fd, "w", encoding="utf-8", errors="backslashreplace"
                 ) as f:
                     json.dump(data, f, indent=2, ensure_ascii=False)
                 RunState._validate_fallback_path(path)
diff --git a/tests/specify_cli/workflows/test_run_state_storage.py b/tests/specify_cli/workflows/test_run_state_storage.py
index b61b3c9df0..f5cf827db7 100644
--- a/tests/specify_cli/workflows/test_run_state_storage.py
+++ b/tests/specify_cli/workflows/test_run_state_storage.py
@@ -2,6 +2,7 @@
 
 import json
 import os
+import stat
 
 import pytest
 
@@ -26,6 +27,45 @@ def _state(project_dir, run_id="run-1"):
     return state
 
 
+@pytest.mark.skipif(os.name != "posix", reason="POSIX permission bits are required")
+@pytest.mark.parametrize("portable", [False, True])
+def test_state_and_input_writes_are_owner_only(project_dir, monkeypatch, portable):
+    from specify_cli.workflows.engine import RunState
+
+    if portable:
+        monkeypatch.setattr(
+            RunState, "_supports_descriptor_access", staticmethod(lambda: False)
+        )
+    original_dump = json.dump
+    temporary_modes = []
+
+    def check_temporary_permissions(data, file, **kwargs):
+        mode = stat.S_IMODE(os.fstat(file.fileno()).st_mode)
+        temporary_modes.append(mode)
+        assert mode == 0o600
+        return original_dump(data, file, **kwargs)
+
+    monkeypatch.setattr(json, "dump", check_temporary_permissions)
+    state = _state(project_dir)
+    state.inputs = {"value": "private"}
+    previous_umask = os.umask(0)
+    try:
+        state.save()
+        for name in ("state.json", "inputs.json"):
+            artifact = state.runs_dir / name
+            assert stat.S_IMODE(artifact.stat().st_mode) == 0o600
+            artifact.chmod(0o666)
+
+        state.save()
+        for name in ("state.json", "inputs.json"):
+            artifact = state.runs_dir / name
+            assert stat.S_IMODE(artifact.stat().st_mode) == 0o600
+            assert json.loads(artifact.read_text(encoding="utf-8"))["inputs"] == state.inputs
+        assert temporary_modes == [0o600] * 4
+    finally:
+        os.umask(previous_umask)
+
+
 def test_workflow_operations_use_portable_storage_fallback(
     project_dir, monkeypatch
 ):
diff --git a/workflows/ARCHITECTURE.md b/workflows/ARCHITECTURE.md
index d7a6f1a1d9..cdaa24a64c 100644
--- a/workflows/ARCHITECTURE.md
+++ b/workflows/ARCHITECTURE.md
@@ -283,9 +283,9 @@ When `specify workflow add <id>` installs from catalog, it downloads the workflo
 ### Persisted-run storage boundary
 
 The `.specify/workflows/runs` tree is a project-owned persistence boundary.
-Every run directory and persisted artifact below it must be a real filesystem
-object rather than a symlink. `RunState` enforces this policy before reading or
-writing `state.json`, `inputs.json`, `workflow.yml`, or `log.jsonl`; artifact
+Accessed storage ancestors, run directories, and persisted artifacts must be
+real filesystem objects rather than symlinks. `RunState` checks these paths
+before reading or writing `state.json`, `inputs.json`, `workflow.yml`, or `log.jsonl`; artifact
 opens also use no-follow flags and directory descriptors where supported.
 The descriptor-backed path opens the project root and walks `.specify`,
 `workflows`, and `runs` one component at a time with no-follow directory opens.
@@ -294,6 +294,8 @@ an ancestor replacement from redirecting access outside the project.
 Other platforms recheck path components before access and use atomic replacement
 for state writes; that fallback rejects existing symlinks but cannot prevent
 concurrent path replacement.
+Both write paths create state and input temporary files with owner-only
+permissions (`0600` on POSIX), retained when atomically replacing the artifacts.
 
 ## Module Structure
 
diff --git a/workflows/README.md b/workflows/README.md
index ce552a60c6..5eb10947c1 100644
--- a/workflows/README.md
+++ b/workflows/README.md
@@ -520,11 +520,12 @@ specify workflow resume <run_id>
 
 Run states: `created` → `running` → `completed` | `paused` | `failed` | `aborted`
 
-Persisted-run storage rejects every symlink below `.specify/workflows/runs`,
-including per-run directories and artifact files. This applies consistently to
-state, inputs, workflow snapshots, and logs for both reads and writes. Workflow
-storage cannot therefore be relocated through symlinks; move the files and
-directories into the project-owned runs directory instead.
+Persisted-run storage rejects symlinked storage ancestors, per-run directories,
+and known artifact files when those paths are accessed. This applies to reads
+and writes of state, inputs, workflow snapshots, and logs; unrelated entries
+and unused artifacts are not proactively scanned. Workflow storage cannot be
+relocated through symlinks; move the files and directories into the
+project-owned runs directory instead.
 
 ## Catalog Management