Fix a divergent partition function shipping as a rate coefficient, plus restart, submission and AutoTST fixes

arc/job/adapters/scripts/autotst_script.py

@@ -14,6 +14,8 @@
 import argparse
 import numpy as np
 import os
+import sys
+import traceback
 import yaml
 
 from autotst.reaction import Reaction
@@ -67,7 +69,13 @@ def main(reaction_label: str | None = None,
         autotst_reaction = Reaction(label=reaction_label)
         autotst_reaction.get_labeled_reaction()
     except Exception:
-        return None
+        # Report, then fail loudly. A bare `return None` here exits 0 and writes no output
+        # file, so ARC sees an adapter that ran cleanly and found nothing: a broken AutoTST
+        # setup (e.g. a missing RMG database) becomes indistinguishable from a chemical
+        # failure. The traceback goes to stderr, which run_in_conda_env captures and logs, and
+        # the non-zero exit makes the adapter's return-code warning fire.
+        traceback.print_exc()
+        sys.exit(1)
 
     ts_list = autotst_reaction.ts['forward'] + autotst_reaction.ts['reverse']
     results = list()

arc/job/adapters/scripts/autotst_script_test.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+# encoding: utf-8
+
+"""
+Unit tests for ``arc.job.adapters.scripts.autotst_script``.
+
+The script runs inside ``tst_env``, where AutoTST is importable. These tests run it in ARC's own
+interpreter against a stub ``autotst`` package, so they exercise the script's exit contract
+without AutoTST installed.
+"""
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from arc.common import ARC_PATH
+
+SCRIPT_PATH = os.path.join(ARC_PATH, 'arc', 'job', 'adapters', 'scripts', 'autotst_script.py')
+
+STUB_REACTION_MODULE = '''
+class Reaction(object):
+    """A stand-in for autotst.reaction.Reaction that fails the way a broken AutoTST setup does."""
+    def __init__(self, label):
+        self.label = label
+
+    def get_labeled_reaction(self):
+        raise RuntimeError('stub AutoTST could not label the reaction')
+'''
+
+
+class TestAutoTSTScriptFailure(unittest.TestCase):
+    """A failure inside AutoTST must reach ARC as a non-zero exit with a traceback."""
+
+    def setUp(self):
+        self.tmp_dir = tempfile.mkdtemp(prefix='test_autotst_script_')
+        self.addCleanup(shutil.rmtree, self.tmp_dir, ignore_errors=True)
+        stub_dir = os.path.join(self.tmp_dir, 'stub', 'autotst')
+        os.makedirs(stub_dir)
+        with open(os.path.join(stub_dir, '__init__.py'), 'w') as f:
+            f.write('')
+        with open(os.path.join(stub_dir, 'reaction.py'), 'w') as f:
+            f.write(STUB_REACTION_MODULE)
+        self.output_path = os.path.join(self.tmp_dir, 'out', 'ts_results.yml')
+
+    def run_script(self) -> subprocess.CompletedProcess:
+        """Run the script against the stub AutoTST package."""
+        env = dict(os.environ, PYTHONPATH=os.path.join(self.tmp_dir, 'stub'))
+        return subprocess.run([sys.executable, SCRIPT_PATH, 'C+[OH]_[CH3]+O', self.output_path],
+                              capture_output=True, text=True, env=env, timeout=120)
+
+    def test_autotst_failure_exits_non_zero_with_a_traceback(self):
+        """A bare `return None` exits 0 and reads as "AutoTST found no TS" instead of an error."""
+        result = self.run_script()
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn('Traceback', result.stderr)
+        self.assertIn('stub AutoTST could not label the reaction', result.stderr)
+        self.assertFalse(os.path.isfile(self.output_path))
+
+
+if __name__ == '__main__':
+    unittest.main(testRunner=unittest.TextTestRunner(verbosity=2))

arc/job/adapters/ts/autotst_test.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+# encoding: utf-8
+
+"""
+Unit tests for the AutoTST TS-search adapter, with the tst_env subprocess boundary mocked.
+"""
+
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+import unittest.mock
+
+import arc.job.adapters.ts.autotst_ts as autotst_ts
+from arc.reaction import ARCReaction
+from arc.species import ARCSpecies
+
+
+class TestAutoTSTAdapter(unittest.TestCase):
+    """Contains unit tests for the AutoTSTAdapter class."""
+
+    def setUp(self):
+        self.project_directory = tempfile.mkdtemp(prefix='test_AutoTST_')
+        self.addCleanup(shutil.rmtree, self.project_directory, ignore_errors=True)
+        self.rxn = ARCReaction(r_species=[ARCSpecies(label='CH4', smiles='C'), ARCSpecies(label='OH', smiles='[OH]')],
+                               p_species=[ARCSpecies(label='CH3', smiles='[CH3]'), ARCSpecies(label='H2O', smiles='O')])
+
+    def test_autotst_runs_without_the_callers_pythonpath(self):
+        """tst_env's own rmgpy must not be shadowed by a source checkout on ARC's PYTHONPATH."""
+        self.assertEqual(self.rxn.family, 'H_Abstraction')
+        adapter = autotst_ts.AutoTSTAdapter(job_type='tsg',
+                                            reactions=[self.rxn],
+                                            testing=True,
+                                            project='test',
+                                            project_directory=self.project_directory,
+                                            )
+        calls = list()
+
+        def fake_run_in_conda_env(python_executable, script_path, *script_args, check=False,
+                                  strip_pythonpath=False):
+            """Record the call and report an AutoTST run that found nothing."""
+            calls.append({'script_path': script_path, 'script_args': script_args,
+                          'strip_pythonpath': strip_pythonpath})
+            return subprocess.CompletedProcess(args=[], returncode=0, stdout='', stderr='')
+
+        with unittest.mock.patch.object(autotst_ts, 'AUTOTST_PYTHON', autotst_ts.__file__), \
+                unittest.mock.patch.object(autotst_ts, 'run_in_conda_env', side_effect=fake_run_in_conda_env):
+            adapter.execute_incore()
+
+        self.assertEqual(len(calls), 2)  # forward and reverse
+        for call in calls:
+            self.assertTrue(call['script_path'].endswith(os.path.join('scripts', 'autotst_script.py')))
+            self.assertTrue(call['strip_pythonpath'])
+
+
+if __name__ == '__main__':
+    unittest.main(testRunner=unittest.TextTestRunner(verbosity=2))

arc/job/adapters/ts/autotst_ts.py

@@ -248,12 +248,18 @@ def execute_incore(self):
                     # run_in_conda_env keeps arc_env's activation vars
                     # (BABEL_LIBDIR, LD_LIBRARY_PATH, ...) from leaking into the
                     # child and corrupting openbabel plugin loading
-                    # (see arc/job/env_run.py).
+                    # (see arc/job/env_run.py). PYTHONPATH is stripped so an
+                    # RMG-Py source checkout on the caller's PYTHONPATH, built for
+                    # arc_env's Python, cannot shadow tst_env's own rmgpy. AutoTST
+                    # itself must then come from tst_env (a .pth file or the
+                    # activation hook written by install_autotst.sh --conda).
                     script_path = os.path.join(ARC_PATH, 'arc', 'job', 'adapters', 'scripts', 'autotst_script.py')
 
                     tic = datetime.datetime.now()
 
-                    output = run_in_conda_env(AUTOTST_PYTHON, script_path, reaction_label, self.output_path)
+                    output = run_in_conda_env(AUTOTST_PYTHON, script_path, reaction_label, self.output_path,
+                                              strip_pythonpath=True,
+                                              )
 
                     tok = datetime.datetime.now() - tic
 

arc/job/pipe/pipe_run.py

@@ -30,6 +30,7 @@
 
 from arc.job.pipe.pipe_state import (
     PipeRunState,
+    ROTOR_SCAN_1D_TASK_FAMILY,
     TaskState,
     TaskSpec,
     get_task_attempt_dir,
@@ -1048,6 +1049,23 @@ def build_ts_opt_tasks(species, label: str, xyzs: list[dict],
     return tasks
 
 
+def rotor_scan_1d_task_id(label: str, rotor_index: int) -> str:
+    """
+    The task_id of a 1D rotor scan pipe task.
+
+    Named so that a reader of a staged pipe run - e.g. the Scheduler's restart loop, which looks for
+    rotors a previous session handed to a pipe - locates a task by the same rule that wrote it.
+
+    Args:
+        label (str): The species label.
+        rotor_index (int): The rotor's index in the species' ``rotors_dict``.
+
+    Returns:
+        str: The task identifier.
+    """
+    return f'{label}_scan_r{rotor_index}'
+
+
 def build_rotor_scan_1d_tasks(species, label: str, rotor_indices: list[int],
                               level_dict: dict, job_adapter: str,
                               memory_mb: int, scan_res: float | None = None) -> list[TaskSpec]:
@@ -1074,12 +1092,13 @@ def build_rotor_scan_1d_tasks(species, label: str, rotor_indices: list[int],
         }
         if scan_res is not None:
             input_payload['scan_res'] = scan_res
+        task_id = rotor_scan_1d_task_id(label, ri)
         tasks.append(TaskSpec(
-            task_id=f'{label}_scan_r{ri}',
-            task_family='rotor_scan_1d',
+            task_id=task_id,
+            task_family=ROTOR_SCAN_1D_TASK_FAMILY,
             owner_type='species',
             owner_key=label,
-            input_fingerprint=f'{label}_scan_r{ri}',
+            input_fingerprint=task_id,
             engine=job_adapter,
             level=level_dict,
             required_cores=cores,

arc/job/pipe/pipe_state.py

@@ -80,13 +80,16 @@ class PipeRunState(str, Enum):
     COMPLETED_PARTIAL = 'COMPLETED_PARTIAL'
     FAILED = 'FAILED'
 
+# Named because the Scheduler's restart loop locates staged 1D rotor scan runs by this family.
+ROTOR_SCAN_1D_TASK_FAMILY = 'rotor_scan_1d'
+
 # Task families currently supported by the pipe system.
 # Only families listed here pass TaskSpec validation.
 SUPPORTED_TASK_FAMILIES = (
     'conf_opt', 'conf_sp',
     'ts_guess_batch_method', 'ts_opt',
     'species_sp', 'species_freq', 'irc',
-    'rotor_scan_1d',
+    ROTOR_SCAN_1D_TASK_FAMILY,
 )
 
 # Owner types mapping to ARC object categories.
@@ -131,6 +134,22 @@ class PipeRunState(str, Enum):
 }
 
 
+def is_terminal_task_state(status: TaskState | str) -> bool:
+    """
+    Whether a task state is terminal, i.e. the pipe system will take no further action on the task.
+
+    Derived from ``TASK_TRANSITIONS`` rather than from a second list of states, so a state that gains
+    or loses an outgoing transition does not need to be remembered in two places.
+
+    Args:
+        status (TaskState or str): The task state.
+
+    Returns:
+        bool: ``True`` if no transition out of ``status`` exists.
+    """
+    return not TASK_TRANSITIONS[TaskState(status)]
+
+
 def check_valid_transition(current_state: TaskState | PipeRunState,
                            new_state: TaskState | PipeRunState,
                            ) -> None:

arc/job/pipe/pipe_state_test.py

@@ -20,6 +20,7 @@
     SUPPORTED_TASK_FAMILIES,
     TASK_FAMILY_TO_JOB_TYPE,
     check_valid_transition,
+    is_terminal_task_state,
     TaskSpec,
     TaskStateRecord,
     generate_claim_token,
@@ -65,6 +66,19 @@ def test_cross_type_raises(self):
         with self.assertRaises(TypeError):
             check_valid_transition(TaskState.PENDING, PipeRunState.CREATED)
 
+    def test_is_terminal_task_state(self):
+        """Only the states the pipe system takes no further action on are terminal"""
+        terminal = {TaskState.COMPLETED, TaskState.FAILED_ESS, TaskState.FAILED_TERMINAL, TaskState.CANCELLED}
+        for state in list(TaskState):
+            self.assertEqual(is_terminal_task_state(state), state in terminal, msg=state.value)
+
+    def test_is_terminal_task_state_accepts_a_persisted_string(self):
+        """A status read back from a state.json is a plain string, not a TaskState"""
+        self.assertTrue(is_terminal_task_state('COMPLETED'))
+        self.assertFalse(is_terminal_task_state('RUNNING'))
+        with self.assertRaises(ValueError):
+            is_terminal_task_state('NOT_A_STATE')
+
 
 class TestTaskSpec(unittest.TestCase):
 

arc/job/ssh.py

@@ -484,13 +484,20 @@ def submit_job(self, remote_path: str,
                 if 'Memory specification can not be satisfied' in line:
                     logger.warning('User may be requesting more memory than is available. Please check server '
                                    'settings, such as cpus and memory, in ARC/arc/settings/settings.py.')
-                if cluster_soft.lower() == 'slurm' and 'AssocMaxSubmitJobLimit' in line:
+                if (cluster_soft.lower() == 'slurm' and 'AssocMaxSubmitJobLimit' in line) \
+                        or (cluster_soft.lower() == 'pbs' and 'Maximum number of jobs already in queue' in line):
+                    if recursion:
+                        # Already slept once for this job. Let the caller mark it errored and move on
+                        # rather than sleeping in a loop while the whole scheduler is blocked.
+                        logger.warning('Max number of submitted jobs still reached after sleeping, giving up '
+                                       'on this submission.')
+                        return job_status, job_id
                     logger.warning(f'Max number of submitted jobs was reached, sleeping...')
                     time.sleep(5 * 60)
-                    self.submit_job(remote_path=remote_path, recursion=True)
-        if recursion:
-            return None, None
-        elif cluster_soft.lower() in ['oge', 'sge'] and stdout and 'submitted' in stdout[0].lower():
+                    # Return the retry's own result: this call's stdout is empty, so falling through
+                    # to the parsing below would discard a successful resubmission.
+                    return self.submit_job(remote_path=remote_path, recursion=True)
+        if cluster_soft.lower() in ['oge', 'sge'] and stdout and 'submitted' in stdout[0].lower():
             job_id = stdout[0].split()[2]
         elif cluster_soft.lower() == 'slurm' and stdout and 'submitted' in stdout[0].lower():
             job_id = stdout[0].split()[3]
@@ -501,8 +508,17 @@ def submit_job(self, remote_path: str,
             # 1 job(s) submitted to cluster 443069.
             if len(stdout) and len(stdout[1].split()) and len(stdout[1].split()[-1].split('.')):
                 job_id = stdout[1].split()[-1][:-1]
-        else:
+        elif cluster_soft.lower() not in ['oge', 'sge', 'slurm', 'pbs', 'htcondor']:
             raise ValueError(f'Unrecognized cluster software: {cluster_soft}')
+        else:
+            # A recognized scheduler that returned nothing parsable on stdout. This is a failed
+            # submission (a full queue, a rejected resource request), not an unsupported cluster:
+            # raising here would kill the whole ARC process over one job, under a message that
+            # blames the scheduler name. Report it as an errored job instead, and let the
+            # scheduler's own troubleshooting decide whether to resubmit.
+            logger.error(f'Could not parse a job ID from the {cluster_soft} submission in {remote_path}. '
+                         f'stdout was:\n{stdout}\nstderr was:\n{stderr}')
+            job_status = 'errored'
         job_status = 'running' if job_id else job_status
         return job_status, job_id