fix(cwl_reana): preserve staged inputs during output copy (reanahub#322)

Unlink InitialWorkDirRequirement input symlinks immediately before
copying job outputs back into the CWL outdir.

Prevent cp from following a destination symlink and overwriting the
original workspace input, which otherwise depends on its write mode.

Closes reanahub#321

Author: tiborsimko

reana_workflow_engine_cwl/cwl_reana.py

@@ -160,6 +160,39 @@ def add_volumes(self, pathmapper):
                     with os.fdopen(fd, "wb") as f:
                         f.write(vol.resolved.encode("utf-8"))
 
+    def _initial_workdir_symlink_cleanup_command(self):
+        """Build a command that unlinks staged inputs before copying outputs."""
+        generatemapper = getattr(self, "generatemapper", None)
+        if not generatemapper:
+            return ""
+
+        container_outdir = self.builder.outdir.rstrip("/")
+        target_prefix = container_outdir + "/"
+        host_outdir = os.path.abspath(self.outdir)
+        cleanup_command = ""
+        for _, volume in generatemapper.items():
+            is_linked_type = volume.type in ("File", "Directory") or (
+                self.inplace_update
+                and volume.type in ("WritableFile", "WritableDirectory")
+            )
+            if (
+                not volume.staged
+                or not is_linked_type
+                or volume.resolved.startswith("_:")
+                or not volume.target.startswith(target_prefix)
+            ):
+                continue
+
+            relative_target = volume.target[len(target_prefix) :]
+            host_target = os.path.abspath(os.path.join(host_outdir, relative_target))
+            if os.path.commonpath([host_outdir, host_target]) != host_outdir:
+                continue
+            quoted_target = shellescape.quote(host_target)
+            cleanup_command += f"; if [ -L {quoted_target} ]; then "
+            cleanup_command += f"rm -f {quoted_target}; fi"
+
+        return cleanup_command
+
     def create_task_msg(self, working_dir, workflow_uuid):  # noqa: C901
         """Create job message spec to be sent to REANA-Job-Controller."""
         job_name = self.name
@@ -247,6 +280,7 @@ def shouldquote(x):
         docker_req, _ = self.get_requirement("DockerRequirement")
         if docker_req:
             docker_output_dir = docker_req.get("dockerOutputDirectory", None)
+        wf_space_cmd += self._initial_workdir_symlink_cleanup_command()
         if docker_output_dir:
             wf_space_cmd = (
                 f"mkdir -p {docker_output_dir} && {wf_space_cmd}"

tests/test_cwl_reana.py

@@ -0,0 +1,58 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of REANA.
+# Copyright (C) 2026 CERN.
+#
+# REANA is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""REANA CWL job tests."""
+
+import subprocess
+from types import SimpleNamespace
+
+import shellescape
+
+from reana_workflow_engine_cwl.cwl_reana import ReanaPipelineJob
+
+
+def test_initial_workdir_input_is_not_overwritten_during_copy(tmp_path):
+    """Unlink a staged input before copying an output with the same name."""
+    uploaded_input = tmp_path / "workspace" / "gendata.C"
+    uploaded_input.parent.mkdir()
+    uploaded_input.write_text("uploaded input")
+
+    outdir = tmp_path / "cwl" / "outdir"
+    outdir.mkdir(parents=True)
+    staged_input = outdir / uploaded_input.name
+    staged_input.symlink_to(uploaded_input)
+
+    container_outdir = "/var/lib/cwl/job"
+    job = object.__new__(ReanaPipelineJob)
+    job.builder = SimpleNamespace(outdir=container_outdir)
+    job.generatemapper = {
+        "input": SimpleNamespace(
+            staged=True,
+            type="File",
+            resolved=str(uploaded_input),
+            target=f"{container_outdir}/{uploaded_input.name}",
+        )
+    }
+    job.inplace_update = False
+    job.outdir = str(outdir)
+
+    generated_output = tmp_path / "job-output"
+    generated_output.mkdir()
+    (generated_output / uploaded_input.name).write_text("generated output")
+
+    cleanup_command = job._initial_workdir_symlink_cleanup_command()
+    command = (
+        f"true{cleanup_command}; cp -r "
+        f"{shellescape.quote(str(generated_output))}/* "
+        f"{shellescape.quote(str(outdir))}"
+    )
+    subprocess.run(["/bin/sh", "-c", command], check=True)
+
+    assert uploaded_input.read_text() == "uploaded input"
+    assert not staged_input.is_symlink()
+    assert staged_input.read_text() == "generated output"