Fix pipe worker OpenBabel failure by injecting LD_LIBRARY_PATH

Pipe workers invoke Python directly, bypassing conda activation hooks. This causes LD_LIBRARY_PATH to be missing, so OpenBabel plugin .so files fail to load their shared library dependencies. Auto-inject CONDA_PREFIX and LD_LIBRARY_PATH exports into the submit script preamble, and add a configurable pre_cmd field to pipe_settings for additional global setup.

Author: alongd

arc/job/pipe/pipe_run.py

@@ -198,6 +198,39 @@ def _submission_resources(self):
         array_size = min(self.max_workers, len(self.tasks)) if self.tasks else self.max_workers
         return cpus, memory_mb, array_size
 
+    def _build_env_preamble(self) -> str:
+        """
+        Build the shell preamble injected into ``{env_setup}`` in submit templates.
+
+        Combines (in order):
+          1. Auto-detected ``CONDA_PREFIX`` and ``LD_LIBRARY_PATH`` exports
+             (needed because pipe workers invoke Python directly, bypassing
+             conda activation hooks that normally set these).
+          2. User-configured ``pre_cmd`` from ``pipe_settings``.
+          3. Engine-specific ``env_setup`` from ``pipe_settings``.
+          4. Scratch directory setup from ``pipe_settings``.
+        """
+        lines = []
+        prefix_lib = os.path.join(sys.prefix, 'lib')
+        if os.path.isdir(prefix_lib):
+            lines.append(f'export CONDA_PREFIX="{sys.prefix}"')
+            lines.append(
+                f'export LD_LIBRARY_PATH="{prefix_lib}${{LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}}"'
+            )
+        pre_cmd = pipe_settings.get('pre_cmd', '')
+        if pre_cmd:
+            lines.append(pre_cmd)
+        engine = self.tasks[0].engine if self.tasks else ''
+        engine_setup = pipe_settings.get('env_setup', {}).get(engine, '')
+        if engine_setup:
+            lines.append(engine_setup)
+        scratch_base = pipe_settings.get('scratch_base', '')
+        if scratch_base:
+            lines.append(
+                f'export TMPDIR="{scratch_base}/${{PBS_JOBID%%[*}}/$PBS_ARRAY_INDEX"\nmkdir -p "$TMPDIR"'
+            )
+        return '\n'.join(lines)
+
     def write_submit_script(self) -> str:
         """
         Generate an array submission script for the configured cluster scheduler.
@@ -217,12 +250,7 @@ def write_submit_script(self) -> str:
         cpus, memory_mb, array_size = self._submission_resources()
         server = servers_dict.get('local', {})
         queue, _ = next(iter(server.get('queues', {}).items()), ('', None))
-        engine = self.tasks[0].engine if self.tasks else ''
-        env_setup = pipe_settings.get('env_setup', {}).get(engine, '')
-        scratch_base = pipe_settings.get('scratch_base', '')
-        if scratch_base:
-            scratch_export = f'export TMPDIR="{scratch_base}/${{PBS_JOBID%%[*}}/$PBS_ARRAY_INDEX"\nmkdir -p "$TMPDIR"'
-            env_setup = f'{env_setup}\n{scratch_export}' if env_setup else scratch_export
+        env_setup = self._build_env_preamble()
         content = pipe_submit[template_key].format(
             name=f'pipe_{self.run_id}',
             max_task_num=array_size,

arc/job/pipe/pipe_run_test.py

@@ -198,6 +198,62 @@ def test_htcondor_sub_not_executable(self):
         self.assertFalse(mode & stat.S_IXUSR, '.sub should not be executable')
 
 
+class TestPipeRunEnvPreamble(unittest.TestCase):
+    """Tests for _build_env_preamble and env injection into submit scripts."""
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp(prefix='pipe_env_preamble_')
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def _make_run(self, cluster_software='slurm', n_tasks=3):
+        tasks = [_make_spec(f't_{i}') for i in range(n_tasks)]
+        run = PipeRun(project_directory=self.tmpdir, run_id='env_test',
+                      tasks=tasks, cluster_software=cluster_software,
+                      max_workers=n_tasks)
+        run.stage()
+        return run
+
+    def test_ld_library_path_in_slurm_script(self):
+        """LD_LIBRARY_PATH export appears in generated slurm submit script."""
+        run = self._make_run('slurm')
+        path = run.write_submit_script()
+        with open(path) as f:
+            content = f.read()
+        self.assertIn('export LD_LIBRARY_PATH=', content)
+        self.assertIn('export CONDA_PREFIX=', content)
+
+    def test_ld_library_path_in_pbs_script(self):
+        """LD_LIBRARY_PATH export appears in generated PBS submit script."""
+        run = self._make_run('pbs')
+        path = run.write_submit_script()
+        with open(path) as f:
+            content = f.read()
+        self.assertIn('export LD_LIBRARY_PATH=', content)
+
+    def test_ld_library_path_before_python(self):
+        """LD_LIBRARY_PATH export appears before the python command."""
+        run = self._make_run('slurm')
+        path = run.write_submit_script()
+        with open(path) as f:
+            content = f.read()
+        ld_pos = content.index('export LD_LIBRARY_PATH=')
+        py_pos = content.index('-m arc.scripts.pipe_worker')
+        self.assertLess(ld_pos, py_pos)
+
+    def test_pre_cmd_injected(self):
+        """User-configured pre_cmd appears in the submit script."""
+        from unittest.mock import patch
+        run = self._make_run('slurm')
+        patched = {'pre_cmd': 'module load openbabel/3.1'}
+        with patch.dict('arc.job.pipe.pipe_run.pipe_settings', patched):
+            path = run.write_submit_script()
+        with open(path) as f:
+            content = f.read()
+        self.assertIn('module load openbabel/3.1', content)
+
+
 class TestPipeRunReconcile(unittest.TestCase):
 
     def setUp(self):

arc/settings/settings.py

@@ -311,6 +311,9 @@
     'env_setup': {},           # Engine-specific shell setup commands, e.g.,
                                # {'gaussian': 'source /usr/local/g09/setup.sh',
                                #  'orca': 'source /usr/local/orca-5.0.4/setup.sh && source /usr/local/openmpi-4.1.1/setup.sh'}
+    'pre_cmd': '',             # Global shell commands injected before every pipe worker invocation.
+                               # Useful for e.g. 'conda activate arc_env' when the auto-detected
+                               # LD_LIBRARY_PATH is not sufficient.
     'scratch_base': '',        # Base directory for worker scratch (e.g., '/gtmp'). Leave empty for system default.
 }