From ae2789dd17ce6ea5c544bf8025a1557e6220bc6a Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:17:37 +0200 Subject: [PATCH 01/13] arkane: discard kinetics computed from a near-zero vibrational frequency A harmonic oscillator at ~0 cm^-1 has a divergent partition function, so a rate coefficient computed from a conformer carrying one can be off by many orders of magnitude, even above the collision limit, while the geometry, the imaginary-frequency check and the Arrhenius fit all look sound. Arkane's rotor projection keeps the n_vib largest Hessian eigenvalues without checking that they are vibrations, so a TS with a second, small imaginary mode (which LOWEST_MAJOR_TS_FREQ tolerates) retains a residual ~0 eigenvalue. After parsing each reaction's kinetics, scan the Arkane conformer blocks of its reactants, products and TS for frequencies below MIN_PHYSICAL_FREQ (5 cm^-1). On a hit, log the offenders, set the reaction's kinetics to None, and note the discard in the TS's long_thermo_description. Co-Authored-By: Claude --- arc/processor.py | 5 ++ arc/processor_test.py | 19 ++++++ arc/statmech/arkane.py | 126 ++++++++++++++++++++++++++++++++---- arc/statmech/arkane_test.py | 81 +++++++++++++++++++++++ 4 files changed, 217 insertions(+), 14 deletions(-) diff --git a/arc/processor.py b/arc/processor.py index 28b6ad7ec1..71bb4f150c 100644 --- a/arc/processor.py +++ b/arc/processor.py @@ -176,6 +176,11 @@ def process_arc_project(thermo_adapter: str, T_min=T_min[0], T_max=T_max[0], ) + elif os.path.isdir(os.path.join(libraries_path, 'kinetics')): + # A library from a previous session of this project may hold rates that this session + # discarded (e.g., computed from a near-zero frequency); do not leave it to be consumed. + logger.warning('No rate coefficient survived; removing the kinetics library of a previous session.') + shutil.rmtree(os.path.join(libraries_path, 'kinetics'), ignore_errors=True) # 2. Thermo if compute_thermo: diff --git a/arc/processor_test.py b/arc/processor_test.py index b667932803..07c0ced65b 100644 --- a/arc/processor_test.py +++ b/arc/processor_test.py @@ -7,6 +7,7 @@ import os import shutil +import tempfile import unittest import arc.processor as processor @@ -78,6 +79,24 @@ def test_compare_rates(self): self.assertIn(TS_IRC_FAILED_MARKER, content[0]['ts_validation']) self.assertNotIn('ts_validation', content[1]) + def test_process_arc_project_removes_a_stale_kinetics_library(self): + """A kinetics library left by a previous session is removed when no rate coefficient survives""" + project_directory = tempfile.mkdtemp(prefix='arc_test_stale_kinetics_lib_') + self.addCleanup(shutil.rmtree, project_directory, ignore_errors=True) + lib_path = os.path.join(project_directory, 'output', 'RMG libraries', 'kinetics') + os.makedirs(lib_path) + with open(os.path.join(lib_path, 'reactions.py'), 'w') as f: + f.write('# a rate computed by a previous session\n') + ch3, h2 = ARCSpecies(label='CH3', smiles='[CH3]'), ARCSpecies(label='H2', smiles='[H][H]') + rxn = ARCReaction(r_species=[self.ch4, self.h], p_species=[ch3, h2]) + rxn.ts_label = 'TS0' + output_dict = {label: {'convergence': label != 'TS0'} for label in ['CH4', 'H', 'CH3', 'H2', 'TS0']} + processor.process_arc_project(thermo_adapter='arkane', kinetics_adapter='arkane', project='stale', + project_directory=project_directory, + species_dict={spc.label: spc for spc in [self.ch4, self.h, ch3, h2]}, + reactions=[rxn], output_dict=output_dict, compute_thermo=False, + compare_to_rmg=False) + self.assertFalse(os.path.isdir(lib_path)) @classmethod def tearDownClass(cls): diff --git a/arc/statmech/arkane.py b/arc/statmech/arkane.py index e18d17fef3..8c13cd02ab 100644 --- a/arc/statmech/arkane.py +++ b/arc/statmech/arkane.py @@ -575,6 +575,7 @@ def parse_arkane_kinetics_output(self, statmech_dir: str) -> None: for rxn in self.reactions: parse_reaction_kinetics(rxn, output_content) + self._reject_kinetics_on_unphysical_frequencies(rxn, output_content) # Parse conformer statmech (external_symmetry, optical_isomers) for all # species involved in the kinetics — the thermo path handles this via @@ -584,6 +585,41 @@ def parse_arkane_kinetics_output(self, statmech_dir: str) -> None: for rxn in self.reactions: _parse_conformer_statmech(rxn.ts_species, output_content) + @staticmethod + def _reject_kinetics_on_unphysical_frequencies(rxn: 'ARCReaction', output_content: str) -> None: + """ + Discard a rate coefficient computed from a conformer carrying a near-zero frequency. + + Reporting nothing is strictly better than reporting a number that can be wrong by six + orders of magnitude while looking converged: a missing rate is visible to whoever reads + the library, and this error is not. See ``find_unphysical_frequencies`` for the mechanism. + """ + if rxn.kinetics is None: + return + labels = [spc.label for spc in rxn.r_species + rxn.p_species] + if rxn.ts_species is not None: + labels.append(rxn.ts_species.label) + offenders = {label: find_unphysical_frequencies(label, output_content) for label in labels} + offenders = {label: freqs for label, freqs in offenders.items() if freqs} + if not offenders: + return + detail = '; '.join(f'{label}: {freqs} cm^-1' for label, freqs in offenders.items()) + logger.error( + f'Discarding the computed rate coefficient for {rxn.label}: the statistical-mechanics ' + f'stage used one or more non-physical near-zero vibrational frequencies ({detail}). A ' + f'harmonic oscillator at 0 cm^-1 has a divergent partition function, so any rate ' + f'computed from it is meaningless -- it can exceed the collision limit -- even though ' + f'the geometry, the imaginary-frequency check and the Arrhenius fit all appear sound. ' + f'This usually means the transition state has a second, small imaginary frequency that ' + f'Arkane retained as a vibration; treat that mode explicitly (e.g. as a hindered or ' + f'free rotor) rather than trusting this fit.') + rxn.kinetics = None + for spc in ([rxn.ts_species] if rxn.ts_species is not None else []): + if spc.label in offenders: + spc.long_thermo_description += ( + f'\nWARNING: non-physical near-zero frequencies {offenders[spc.label]} cm^-1 ' + f'were present in the statmech treatment; the rate coefficient was discarded.') + def run_arkane(statmech_dir: str) -> bool: """ @@ -1432,13 +1468,30 @@ def _parse_conformer_statmech(species, content: str) -> None: NonlinearRotor(symmetry=2, ...) optical_isomers = 1, """ - label = species.label - # Find the start of the conformer block, then use balanced-paren matching - # to find the full block (regex `.*?)` fails because of nested parens). + block = _extract_conformer_block(species.label, content) + if block is None: + return + # optical_isomers + oi_match = re.search(r'optical_isomers\s*=\s*(\d+)', block) + if oi_match and species.optical_isomers is None: + species.optical_isomers = int(oi_match.group(1)) + # external_symmetry from NonlinearRotor or LinearRotor + sym_match = re.search(r'(?:NonlinearRotor|LinearRotor)\s*\(.*?symmetry\s*=\s*(\d+)', block, re.DOTALL) + if sym_match and species.external_symmetry is None: + species.external_symmetry = int(sym_match.group(1)) + + +def _extract_conformer_block(label: str, content: str) -> str | None: + """ + Return the full ``conformer(...)`` block for a label from Arkane's output. + + Uses balanced-paren matching from the block's start rather than a regex: the block contains + nested parentheses, so a lazy `.*?)` stops at the first inner one. + """ start_pattern = rf"conformer\(\s*label\s*=\s*['\"]{re.escape(label)}['\"]" m_start = re.search(start_pattern, content, re.DOTALL) if not m_start: - return + return None idx = m_start.start() + len('conformer(') depth = 1 while idx < len(content) and depth > 0: @@ -1448,16 +1501,61 @@ def _parse_conformer_statmech(species, content: str) -> None: depth -= 1 idx += 1 if depth != 0: - return - block = content[m_start.start():idx] - # optical_isomers - oi_match = re.search(r'optical_isomers\s*=\s*(\d+)', block) - if oi_match and species.optical_isomers is None: - species.optical_isomers = int(oi_match.group(1)) - # external_symmetry from NonlinearRotor or LinearRotor - sym_match = re.search(r'(?:NonlinearRotor|LinearRotor)\s*\(.*?symmetry\s*=\s*(\d+)', block, re.DOTALL) - if sym_match and species.external_symmetry is None: - species.external_symmetry = int(sym_match.group(1)) + return None + return content[m_start.start():idx] + + +# A harmonic oscillator's partition function, 1/(1 - exp(-h*nu/kT)), diverges as nu -> 0, so a +# mode at or near zero inflates a partition function without bound. Real vibrations of a bound +# species never come near this floor -- the softest torsions seen in practice are tens of cm^-1 -- +# so anything below it is an artifact rather than chemistry. +MIN_PHYSICAL_FREQ = 5.0 # cm^-1 + + +def find_unphysical_frequencies(label: str, content: str) -> list[float]: + """ + Return any near-zero vibrational frequencies in a species' Arkane conformer block. + + This failure is catastrophic rather than merely inaccurate, and it is silent. A single mode at + 0 cm^-1 can multiply a transition state's partition function by roughly seven orders of + magnitude at combustion temperatures, and the rate coefficient follows it, to the point of + exceeding the hard-sphere collision limit. Nothing else in the output marks it: the geometry + converges, the TS passes its imaginary-frequency check, and the Arrhenius fit looks clean + because it is a faithful fit to corrupted points. + + The mechanism is upstream, in Arkane's rotor projection. ``arkane/statmech.py`` returns + ``np.sqrt(eig[-n_vib:])``, keeping the ``n_vib`` largest eigenvalues of the projected Hessian + without requiring them to be real vibrations. A transition state with a second, small + imaginary frequency -- legitimately accepted by ``arc.checks.ts.check_imaginary_frequencies``, + whose ``LOWEST_MAJOR_TS_FREQ`` threshold exists precisely to tolerate small extra imaginary + modes -- leaves that count one short, so a residual ~0 eigenvalue is retained as a vibration. + ARC cannot head this off by filtering the frequencies it passes in, because Arkane re-derives + them from the Hessian whenever rotors are present. Hence the check runs on Arkane's output. + + Args: + label (str): The species label whose conformer block to inspect. + content (str): The contents of Arkane's ``output.py``. + + Returns: + list[float]: The offending frequencies in cm^-1; empty when the conformer is sound. + """ + block = _extract_conformer_block(label, content) + if block is None: + return list() + unphysical = list() + for freq_list in re.findall(r'HarmonicOscillator\s*\(\s*frequencies\s*=\s*\(\s*\[([^\]]*)\]', + block, re.DOTALL): + for token in freq_list.split(','): + token = token.strip() + if not token: + continue + try: + freq = float(token) + except ValueError: + continue + if abs(freq) < MIN_PHYSICAL_FREQ: + unphysical.append(freq) + return unphysical def parse_e0(label: str, content: str) -> float | None: diff --git a/arc/statmech/arkane_test.py b/arc/statmech/arkane_test.py index a724f49eb4..5254cd9398 100644 --- a/arc/statmech/arkane_test.py +++ b/arc/statmech/arkane_test.py @@ -25,6 +25,7 @@ _available_years_for_level, _extract_section, find_best_across_files, + find_unphysical_frequencies, _find_best_level_key_for_sp_level, get_qm_corrections_files, _level_to_str, @@ -663,6 +664,86 @@ def test_parse_reaction_kinetics_does_not_mark_an_unchecked_or_valid_ts(self): self.assertNotIn('ts_validation', rxn.kinetics) self.assertNotIn(TS_IRC_FAILED_MARKER, rxn.kinetics['comment']) + near_zero_freq_kinetics_output_content = """ +conformer( + label = 'TS0', + E0 = (50.0, 'kJ/mol'), + modes = [ + NonlinearRotor(inertia=([10.0, 20.0, 30.0], 'amu*angstrom^2'), symmetry=1), + HarmonicOscillator(frequencies=([0.0, 3.2, 125.4, 1450.1, 3050.7], 'cm^-1')), + ], + spin_multiplicity = 1, + optical_isomers = 1, +) + +conformer( + label = 'nitroethane', + E0 = (-100.0, 'kJ/mol'), + modes = [ + NonlinearRotor(inertia=([10.0, 20.0, 30.0], 'amu*angstrom^2'), symmetry=1), + HarmonicOscillator(frequencies=([45.3, 610.2, 1500.0], 'cm^-1')), + ], + spin_multiplicity = 1, + optical_isomers = 1, +) + +kinetics( + label = 'nitroethane <=> ethyl_nitrite', + kinetics = Arrhenius( + A = (5.0e20, 's^-1'), + n = 1.0, + Ea = (20.0, 'kJ/mol'), + T0 = (1, 'K'), + Tmin = (300, 'K'), + Tmax = (2000, 'K'), + ), +) +""" + + def test_find_unphysical_frequencies(self): + """Near-zero harmonic frequencies are reported per conformer; physical ones are not.""" + content = self.near_zero_freq_kinetics_output_content + self.assertEqual(find_unphysical_frequencies('TS0', content), [0.0, 3.2]) + self.assertEqual(find_unphysical_frequencies('nitroethane', content), []) + self.assertEqual(find_unphysical_frequencies('missing_species', content), []) + + def test_kinetics_parsed_from_a_near_zero_frequency_conformer_are_discarded(self): + """A rate coefficient whose TS carried a ~0 cm^-1 oscillator is discarded and flagged.""" + statmech_dir = tempfile.mkdtemp(prefix='test_Arkane_near_zero_') + self.addCleanup(shutil.rmtree, statmech_dir, ignore_errors=True) + with open(os.path.join(statmech_dir, 'output.py'), 'w') as f: + f.write(self.near_zero_freq_kinetics_output_content) + rxn = self.isomerization_reaction() + adapter = ArkaneAdapter(output_directory=os.path.join(statmech_dir, 'output'), + calcs_directory=os.path.join(statmech_dir, 'calcs'), + output_dict=dict(), + bac_type=None, + species=rxn.r_species + rxn.p_species + [rxn.ts_species], + reactions=[rxn], + ) + adapter.parse_arkane_kinetics_output(statmech_dir) + self.assertIsNone(rxn.kinetics) + self.assertIn('non-physical near-zero frequencies [0.0, 3.2]', rxn.ts_species.long_thermo_description) + + def test_kinetics_parsed_from_physical_conformers_are_kept(self): + """Kinetics whose conformers carry only physical frequencies are kept as parsed.""" + statmech_dir = tempfile.mkdtemp(prefix='test_Arkane_physical_') + self.addCleanup(shutil.rmtree, statmech_dir, ignore_errors=True) + content = self.near_zero_freq_kinetics_output_content.replace('[0.0, 3.2, 125.4', '[85.0, 125.4') + with open(os.path.join(statmech_dir, 'output.py'), 'w') as f: + f.write(content) + rxn = self.isomerization_reaction() + adapter = ArkaneAdapter(output_directory=os.path.join(statmech_dir, 'output'), + calcs_directory=os.path.join(statmech_dir, 'calcs'), + output_dict=dict(), + bac_type=None, + species=rxn.r_species + rxn.p_species + [rxn.ts_species], + reactions=[rxn], + ) + adapter.parse_arkane_kinetics_output(statmech_dir) + self.assertAlmostEqual(rxn.kinetics['A'][0], 5.0e20) + self.assertNotIn('non-physical', rxn.ts_species.long_thermo_description) + def test_parse_thermo_data_block_scalars_are_float(self): """Verify Tmin, Tmax, H298, S298 are parsed as floats, not strings.""" block = """ From 08c6a3a945114c86b9ba50ebec703b9d6490bb48 Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:17:37 +0200 Subject: [PATCH 02/13] plotter: decide the "No rotors considered" TS note from the rotor data save_kinetics_lib appended "No rotors considered for this TS." whenever the word "rotors" was absent from the TS's long_thermo_description. A TS that Arkane modelled with a hindered rotor was therefore described as having none, and a description that merely mentioned rotors suppressed the note for a TS that had none. Decide it from rotors_dict instead, using the same success-is-truthy filter the Arkane adapter applies when it builds the model. Co-Authored-By: Claude --- arc/plotter.py | 39 ++++++++++++++++++++++++-- arc/plotter_test.py | 67 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 2 deletions(-) diff --git a/arc/plotter.py b/arc/plotter.py index b3581501ed..70d7817f94 100644 --- a/arc/plotter.py +++ b/arc/plotter.py @@ -892,6 +892,31 @@ def save_transport_lib(species_list, path, name, lib_long_desc=''): pass +#: Appended to a TS's ``long_thermo_description`` when no rotor of it was treated as a hindered +#: rotor. Named because it is both written and stripped: see ``save_kinetics_lib``. +NO_TS_ROTORS_NOTE = 'No rotors considered for this TS.' + + +def ts_has_treated_rotors(ts_species) -> bool: + """ + Whether any of a TS's torsions was actually treated as a rotor in the statmech stage. + + Mirrors the filter the statmech adapter applies (``arc/statmech/arkane.py``, which keeps only + rotors whose ``success`` is truthy), so the library's description matches the model it was + generated from. Both ``False`` (the scan ran and was rejected) and ``None`` (the scan never + completed) mean untreated, and are indistinguishable downstream. + + Args: + ts_species (ARCSpecies): The TS species to inspect. + + Returns: + bool: True when at least one rotor was treated, False when the TS entered as a pure + rigid-rotor harmonic oscillator. + """ + rotors_dict = getattr(ts_species, 'rotors_dict', None) or dict() + return any(rotor.get('success') for rotor in rotors_dict.values()) + + def save_kinetics_lib(rxn_list: list, path: str, name: str, @@ -931,8 +956,18 @@ def save_kinetics_lib(rxn_list: list, f'kinetics library.') continue rxn.ts_species.make_ts_report() - if 'rotors' not in rxn.ts_species.long_thermo_description: - rxn.ts_species.long_thermo_description += '\nNo rotors considered for this TS.' + # Decide this from the rotor data, not from whether the word "rotors" happens to occur in a + # free-text description. That substring test reported "No rotors considered for this TS" + # for a TS that Arkane had in fact modelled WITH a hindered rotor, so the shipped library + # asserted the opposite of the model it was generated from. + # The note is derived state, recomputed on every save, so any copy left by an earlier save is + # removed before deciding again. ``long_thermo_description`` is a persisted species attribute + # (it survives into restart.yml), so appending unconditionally would both stack the sentence up + # once per export and keep asserting "no rotors" after a rotor had since been treated. + rxn.ts_species.long_thermo_description = rxn.ts_species.long_thermo_description.replace( + f'\n{NO_TS_ROTORS_NOTE}', '').replace(NO_TS_ROTORS_NOTE, '') + if not ts_has_treated_rotors(rxn.ts_species): + rxn.ts_species.long_thermo_description += f'\n{NO_TS_ROTORS_NOTE}' long_desc = f'{rxn.ts_species.ts_report}\n\n' \ f'TS external symmetry: {rxn.ts_species.external_symmetry}, ' \ f'TS optical isomers: {rxn.ts_species.optical_isomers}\n\n' \ diff --git a/arc/plotter_test.py b/arc/plotter_test.py index d62dd2faf7..79b2438282 100644 --- a/arc/plotter_test.py +++ b/arc/plotter_test.py @@ -13,6 +13,7 @@ import arc.plotter as plotter from arc.common import ARC_PATH, ARC_TESTING_PATH, read_yaml_file, safe_copy_file +from arc.reaction import ARCReaction from arc.species.converter import str_to_xyz from arc.species.species import ARCSpecies @@ -115,6 +116,72 @@ def test_generated_library_loads_in_rmg(self): self.assertIn('LOADED 1', result.stdout) +class TestSaveKineticsLibRotorNote(unittest.TestCase): + """The kinetics library states "No rotors considered" exactly when no TS rotor was treated.""" + + def setUp(self): + self.tmpdir = tempfile.mkdtemp(prefix='test_kinetics_lib_rotors_') + self.addCleanup(shutil.rmtree, self.tmpdir, ignore_errors=True) + + def save_and_read(self, rotors_dict, long_thermo_description='', saves=1): + """Save a one-reaction library whose TS carries ``rotors_dict`` and return its reactions.py text. + + ``saves`` repeats the export on the SAME reaction object, which is what an ARC run does across + repeated exports and restarts: ``long_thermo_description`` is a persisted species attribute, not + a local, so anything appended to it survives into the next save. + """ + rxn = ARCReaction(r_species=[ARCSpecies(label='nitroethane', smiles='CC[N+](=O)[O-]')], + p_species=[ARCSpecies(label='ethyl_nitrite', smiles='CCON=O')]) + rxn.ts_species = ARCSpecies(label='TS0', is_ts=True, xyz=str_to_xyz("""C 0.0 0.0 0.0 + H 0.0 0.0 1.09""")) + rxn.ts_species.final_xyz = rxn.ts_species.get_xyz() + rxn.ts_species.rotors_dict = rotors_dict + rxn.ts_species.long_thermo_description = long_thermo_description + rxn.kinetics = {'A': (5.0, 's^-1'), 'n': 1.0, 'Ea': (20.0, 'kJ/mol')} + for _ in range(saves): + plotter.save_kinetics_lib(rxn_list=[rxn], path=self.tmpdir, name='test', lib_long_desc='') + with open(os.path.join(self.tmpdir, 'kinetics', 'reactions.py'), 'r') as f: + return f.read() + + def test_treated_rotor_is_not_reported_as_absent(self): + """A TS with a successfully treated rotor must not be described as having none.""" + content = self.save_and_read(rotors_dict={0: {'pivots': [1, 2], 'success': True}}) + self.assertNotIn('No rotors considered for this TS.', content) + + def test_untreated_rotors_are_reported_even_if_the_description_mentions_rotors(self): + """Rotors that were rejected or never completed count as untreated, whatever the free text says.""" + content = self.save_and_read(rotors_dict={0: {'pivots': [1, 2], 'success': False}, + 1: {'pivots': [2, 3], 'success': None}}, + long_thermo_description='Hindered rotors were scanned at B3LYP.') + self.assertIn('No rotors considered for this TS.', content) + + def test_ts_without_rotors_dict_is_reported(self): + """A TS that never had a rotors_dict is described as having no rotors.""" + content = self.save_and_read(rotors_dict=None) + self.assertIn('No rotors considered for this TS.', content) + + def test_the_note_is_written_once_however_often_the_library_is_saved(self): + """Exporting the same reaction repeatedly must not stack the sentence up. + + The note is appended to ``long_thermo_description``, which persists on the species and through + restart.yml, so an unconditional append writes it once per save. The substring guard this + replaced happened to prevent that; deciding from the rotor data has to prevent it explicitly. + """ + content = self.save_and_read(rotors_dict={0: {'pivots': [1, 2], 'success': None}}, saves=3) + self.assertEqual(content.count('No rotors considered for this TS.'), 1) + + def test_a_stale_note_is_dropped_once_a_rotor_is_treated(self): + """A note left by an earlier export must not outlive the condition that produced it. + + A TS whose rotor failed in one session and succeeded in the next carries the old sentence in + its description, which would then contradict the model actually shipped -- the same class of + error as the substring guard this change replaced, just in the other direction. + """ + content = self.save_and_read(rotors_dict={0: {'pivots': [1, 2], 'success': True}}, + long_thermo_description='\nNo rotors considered for this TS.') + self.assertNotIn('No rotors considered for this TS.', content) + + class TestPlotter(unittest.TestCase): """ Contains unit tests for the parser functions From 21b7ceab083b464cd3a8ca2fa8ddd01221633a8c Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:20:32 +0200 Subject: [PATCH 03/13] scheduler: skip IRC endpoint rotors on restart IRC endpoint species are diagnostic, not thermo/kinetics targets, and the normal flow never scans their rotors. They still carry a populated rotors_dict without a scan_path, so the restart loop spawned a scan per IRC rotor. Exclude them both as the restarted species and in the scan over all species. Co-Authored-By: Claude --- arc/scheduler.py | 8 +++-- arc/scheduler_test.py | 70 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 2 deletions(-) diff --git a/arc/scheduler.py b/arc/scheduler.py index 43461c2396..f00285c010 100644 --- a/arc/scheduler.py +++ b/arc/scheduler.py @@ -614,8 +614,12 @@ def __init__(self, and not self.output[species.label]['paths']['freq'] and self.job_types['freq'] \ and (species.is_ts or species.number_of_atoms > 1) and species.irc_label is None: self.run_freq_job(species.label) - if self.output[species.label]['paths']['geo'] and self.job_types['rotors'] and \ - any(spc.rotors_dict is not None and + # IRC endpoint species are diagnostic, not thermo/kinetics targets, and the normal + # flow never scans their rotors. They nevertheless carry a populated rotors_dict + # without a scan_path, so they must not trigger (nor be) a restart scan spawn. + if self.output[species.label]['paths']['geo'] and self.job_types['rotors'] \ + and not species.label.startswith('IRC_') and \ + any(spc.rotors_dict is not None and not spc.label.startswith('IRC_') and any(rotor_dict['success'] is False and not rotor_dict['invalidation_reason'] for rotor_dict in spc.rotors_dict.values()) for spc in self.species_list): diff --git a/arc/scheduler_test.py b/arc/scheduler_test.py index 1f5749df1e..29bd47c444 100644 --- a/arc/scheduler_test.py +++ b/arc/scheduler_test.py @@ -320,6 +320,76 @@ def test_restore_running_jobs_conformer_reconnects(self): # And the restored names are the live '{job_type}_{i}' format, not the fossil 'conformer{i}'. self.assertEqual(sched.running_jobs[label], ['conf_opt_0', 'conf_sp_0']) + _METHYLAMINE_XYZ = """C -0.57422867 -0.01669771 0.01229213 +N 0.82084044 0.08279104 -0.37769346 +H -1.05737005 -0.84067772 -0.52007494 +H -1.10211468 0.90879867 -0.23383011 +H -0.66133128 -0.19490562 1.08785111 +H 0.88047852 0.26966160 -1.37780789 +H 1.27889520 -0.81548721 -0.22940984""" + + def _species_with_rotor(self, label, success=None, scan_path=None, invalidation_reason=''): + """Build a species with a single (CN) torsion in its rotors_dict.""" + spc = ARCSpecies(label=label, smiles='CN', xyz=self._METHYLAMINE_XYZ) + spc.final_xyz = spc.get_xyz() # a restarted species whose optimization has finished + spc.number_of_rotors = 1 + spc.rotors_dict = {0: {'pivots': [1, 2], 'top': [3, 4, 5], 'scan': [3, 1, 2, 6], 'torsion': [2, 0, 1, 5], + 'success': success, 'scan_path': scan_path, + 'invalidation_reason': invalidation_reason, 'directed_scan_type': 'ess'}} + return spc + + def _restart_scheduler(self, species_list, **patched): + """ + Construct a non-testing Scheduler from a restart dict in which every species already has a geometry + (so the restart loop in ``__init__`` runs), with ``schedule_jobs`` and the given methods patched out. + + Returns: + dict: The mocks, keyed by method name (``schedule_jobs`` included). + """ + job_types = dict(self.job_types1, rotors=True) + kwargs = dict(project='project_test_restart_loop', ess_settings=self.ess_settings, + species_list=species_list, composite_method=None, + conformer_opt_level=Level(repr=default_levels_of_theory['conformer']), + opt_level=Level(repr=default_levels_of_theory['opt']), + freq_level=Level(repr=default_levels_of_theory['freq']), + sp_level=Level(repr=default_levels_of_theory['sp']), + scan_level=Level(repr=default_levels_of_theory['scan']), + ts_guess_level=Level(repr=default_levels_of_theory['ts_guesses']), + project_directory=tempfile.mkdtemp(prefix='arc_test_restart_loop_'), + job_types=job_types, orbitals_level=default_levels_of_theory['orbitals'], + adaptive_levels=None) + self.addCleanup(shutil.rmtree, kwargs['project_directory'], ignore_errors=True) + output = Scheduler(testing=True, **kwargs).output + for spc_output in output.values(): + spc_output['paths'].update({'geo': 'geo.out', 'freq': 'freq.out', 'sp': 'sp.out'}) + mocks = dict() + with patch.object(Scheduler, 'schedule_jobs') as schedule_jobs: + mocks['schedule_jobs'] = schedule_jobs + patchers = {name: patch.object(Scheduler, name) for name in patched} + for name, patcher in patchers.items(): + mocks[name] = patcher.start() + self.addCleanup(patcher.stop) + mocks['sched'] = Scheduler(testing=False, restart_dict={'output': output}, **kwargs) + return mocks + + def test_restart_does_not_scan_irc_endpoint_rotors(self): + """ + IRC endpoint species are diagnostic and the normal flow never scans their rotors, but they carry a + populated rotors_dict with no scan_path. The restart loop used to treat that as pending scan work, + both when the IRC species itself was restarted and when it was found by the scan over all species. + """ + regular = self._species_with_rotor('regular', success=True, scan_path=__file__) + irc = self._species_with_rotor('IRC_regular_1', success=False) + mocks = self._restart_scheduler([regular, irc], run_scan_jobs=None) + self.assertEqual(mocks['run_scan_jobs'].call_args_list, list()) + + def test_restart_still_scans_a_failed_regular_rotor(self): + """The IRC guard must not suppress the restart scan of an ordinary species whose rotor failed""" + regular = self._species_with_rotor('regular', success=False) + irc = self._species_with_rotor('IRC_regular_1', success=False) + mocks = self._restart_scheduler([regular, irc], run_scan_jobs=None) + self.assertEqual([c.args for c in mocks['run_scan_jobs'].call_args_list], [('regular',)]) + def test_check_negative_freq(self): """Test the check_negative_freq() method""" label = 'C2H6' From 84e55d35c0c93ab167872529a9a32f8b37a0f3db Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:21:05 +0200 Subject: [PATCH 04/13] scheduler: restart rotors that were never scanned On restart only rotors with success=False were respawned. A rotor with no result and no scan_path, such as one added to restart.yml by hand or determined after the previous session died, was silently never run and the species was reported as converged while missing a mode. A rotor queued or running in a pipe run carries that same state, and pipe runs are not restored on restart: the coordinator starts with no active pipes and is constructed after the restart loop, so has_pending_pipe_work() cannot see one. Such a rotor is therefore skipped unless its staged pipe task has reached a terminal state, which keeps the restart from submitting a standalone scan that duplicates a job still in the queue. Co-Authored-By: Claude --- arc/job/pipe/pipe_run.py | 25 +++++- arc/job/pipe/pipe_state.py | 21 ++++- arc/job/pipe/pipe_state_test.py | 14 ++++ arc/scheduler.py | 104 +++++++++++++++++++++--- arc/scheduler_test.py | 135 +++++++++++++++++++++++++++++++- 5 files changed, 284 insertions(+), 15 deletions(-) diff --git a/arc/job/pipe/pipe_run.py b/arc/job/pipe/pipe_run.py index 43e0d3c05b..c19a9890c5 100644 --- a/arc/job/pipe/pipe_run.py +++ b/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, diff --git a/arc/job/pipe/pipe_state.py b/arc/job/pipe/pipe_state.py index 7333938d14..f799bd6924 100644 --- a/arc/job/pipe/pipe_state.py +++ b/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: diff --git a/arc/job/pipe/pipe_state_test.py b/arc/job/pipe/pipe_state_test.py index 5b7175977d..2bb09a177c 100644 --- a/arc/job/pipe/pipe_state_test.py +++ b/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): diff --git a/arc/scheduler.py b/arc/scheduler.py index f00285c010..593fe20643 100644 --- a/arc/scheduler.py +++ b/arc/scheduler.py @@ -6,6 +6,7 @@ from __future__ import annotations import datetime +import glob import itertools import os import pprint @@ -50,6 +51,8 @@ from arc.job.local import check_running_jobs_ids from arc.job.pipe.pipe_coordinator import PipeCoordinator from arc.job.pipe.pipe_planner import PipePlanner +from arc.job.pipe.pipe_run import rotor_scan_1d_task_id +from arc.job.pipe.pipe_state import ROTOR_SCAN_1D_TASK_FAMILY, is_terminal_task_state, read_task_state from arc.job.ssh_pool import borrow_ssh_client from arc.job.trsh import (scan_quality_check, trsh_conformer_isomorphism, @@ -539,6 +542,11 @@ def __init__(self, if species.label not in self.job_dict: self.job_dict[species.label] = dict() if species.yml_path is None: + # Evaluated BEFORE determine_rotors() below, and used only for the converged short-circuit. + # determine_rotors() manufactures fresh rotors (success None, no scan_path) for any species + # whose restored number_of_rotors is 0, which is indistinguishable from a rotor genuinely + # owed - so asking afterwards would re-scan every converged species on every restart. + restored_rotor_scans_owed = self.rotor_scans_pending_on_restart(species) if self.job_types['rotors'] and not species.number_of_rotors and species.rotors_dict is not None: # if species.rotors_dict is None, it means the species is marked to not spawn rotor scans species.determine_rotors() @@ -551,6 +559,14 @@ def __init__(self, if species.label not in self.running_jobs: self.running_jobs[species.label if not species.multi_species else species.multi_species] = list() if self.output[species.label]['convergence']: + # A species saved as converged is scheduled no further - except for a rotor scan it + # still owes. ``convergence`` is restored verbatim from restart.yml, so without this + # a rotor added to an already-finished species is skipped on every subsequent restart + # and the species keeps being reported as converged while missing a mode. Convergence + # itself is deliberately left set: clearing it would re-open every other job type for + # a species whose other work is genuinely done. + if restored_rotor_scans_owed: + self.run_scan_jobs(species.label) continue if species.is_monoatomic(): if not self.output[species.label]['job_types']['sp'] \ @@ -614,15 +630,7 @@ def __init__(self, and not self.output[species.label]['paths']['freq'] and self.job_types['freq'] \ and (species.is_ts or species.number_of_atoms > 1) and species.irc_label is None: self.run_freq_job(species.label) - # IRC endpoint species are diagnostic, not thermo/kinetics targets, and the normal - # flow never scans their rotors. They nevertheless carry a populated rotors_dict - # without a scan_path, so they must not trigger (nor be) a restart scan spawn. - if self.output[species.label]['paths']['geo'] and self.job_types['rotors'] \ - and not species.label.startswith('IRC_') and \ - any(spc.rotors_dict is not None and not spc.label.startswith('IRC_') and - any(rotor_dict['success'] is False and not rotor_dict['invalidation_reason'] - for rotor_dict in spc.rotors_dict.values()) - for spc in self.species_list): + if self.rotor_scans_pending_on_restart(species): # Additional restart-related checks are performed within run_scan_jobs(). self.run_scan_jobs(species.label) else: @@ -651,6 +659,84 @@ def __init__(self, if not self.testing: self.schedule_jobs() + def rotor_scans_pending_on_restart(self, species: ARCSpecies) -> bool: + """ + Whether the restart loop still owes ``species`` a rotor scan spawn. + + IRC endpoint species are diagnostic, not thermo/kinetics targets, and the normal flow never scans + their rotors. They nevertheless carry a populated ``rotors_dict`` without a scan_path, so they must + not trigger (nor be) a restart scan spawn. + + A rotor that was never scanned (no scan_path) counts as owed, not only one that failed: it may have + been added to restart.yml by hand, or determined after the previous session died. Otherwise it is + silently skipped and the species is reported as converged while missing a mode. A rotor a pipe run + still owes a result looks identical and is excluded, see ``rotor_scan_owed_to_a_pipe_run()``; a rotor + whose scan failed (``success`` is ``False``) is not, since a pipe that reported a failure is done + with it and re-spawning such a rotor is this loop's long-standing behaviour. + + Args: + species (ARCSpecies): The species being restarted. + + Returns: + bool: Whether ``run_scan_jobs()`` should be called for this species. + """ + if not self.output[species.label]['paths']['geo'] or not self.job_types['rotors'] \ + or species.label.startswith('IRC_') or species.rotors_dict is None: + # A None rotors_dict marks a species deliberately excluded from rotor scans. + return False + return any((rotor_dict['success'] is False and not rotor_dict['invalidation_reason']) + or (rotor_dict['success'] is None and not rotor_dict['scan_path'] + and not self.rotor_scan_owed_to_a_pipe_run(species, rotor_index)) + for rotor_index, rotor_dict in species.rotors_dict.items()) + + def rotor_scan_owed_to_a_pipe_run(self, species: ARCSpecies, rotor_index: int) -> bool: + """ + Whether a pipe run staged by a previous session still owes this rotor a scan. + + Pipe runs do not survive a restart: ``PipeCoordinator`` starts with an empty ``active_pipes``, no + code reloads the runs persisted under ``/tasks/``, and the restart loop in ``__init__`` + runs before the coordinator and planner are constructed at all - so ``has_pending_pipe_work()`` + cannot see a piped rotor, and ``run_scan_jobs()`` falls back to submitting each rotor individually. + A rotor queued or running in a pipe carries exactly the state the caller reads as 'never scanned' + (no result, no scan_path), so without this check the restart spawns a standalone ESS scan + duplicating a job that may still be in the queue. + + The run's staging directory is the only record of it available this early, and it is deterministic + (see ``PipeCoordinator._compute_pipe_root()``) with one task directory per rotor. A task in a + non-terminal state means the pipe still owes this rotor a result. A terminal one does not suppress + the spawn: nothing will ingest its results now that the run is gone, so the rotor does need a fresh + scan and submitting one duplicates nothing. A state file that cannot be read is treated as + non-terminal, erring away from submitting. + + This does not reconcile pipe runs across a restart - a pre-existing gap, and a feature rather than + a review fix. It only keeps the restart loop from submitting work a pipe may already be running. + + Args: + species (ARCSpecies): The species being restarted. + rotor_index (int): The rotor's index in ``species.rotors_dict``. + + Returns: + bool: Whether a pipe run still owes this rotor a scan. + """ + task_id = rotor_scan_1d_task_id(species.label, rotor_index) + task_dirs = glob.glob(os.path.join(self.project_directory, 'calcs', + 'TSs' if species.is_ts else 'Species', species.label, + f'pipe_{ROTOR_SCAN_1D_TASK_FAMILY}_*', 'tasks', task_id)) + for task_dir in task_dirs: + pipe_root = os.path.dirname(os.path.dirname(task_dir)) + try: + status = read_task_state(pipe_root, task_id).status + except (OSError, ValueError, KeyError) as e: + logger.warning(f'Could not read the state of pipe task {task_id} under {pipe_root}: {e}\n' + f'Assuming the pipe run still owes rotor {rotor_index} of {species.label} ' + f'a scan, and not spawning one on restart.') + return True + if not is_terminal_task_state(status): + logger.info(f'Rotor {rotor_index} of {species.label} is {status} in the pipe run at ' + f'{pipe_root}; not spawning a standalone scan for it on restart.') + return True + return False + def has_pending_pipe_work(self, label: str) -> bool: """ Whether a species still has work queued for, or running in, a pipe run. diff --git a/arc/scheduler_test.py b/arc/scheduler_test.py index 29bd47c444..6800332bb1 100644 --- a/arc/scheduler_test.py +++ b/arc/scheduler_test.py @@ -12,6 +12,7 @@ import os import shutil import tempfile +import time from types import SimpleNamespace @@ -24,6 +25,9 @@ REFERENCE_AGNOSTIC_METHOD_TYPES, REFERENCE_CHANGE_AVAILABLE_KEY, ts_adapters_by_rmg_family, ts_adapters_for_unknown_unimolecular) from arc.job.factory import job_factory +from arc.job.pipe.pipe_coordinator import PipeCoordinator +from arc.job.pipe.pipe_run import build_rotor_scan_1d_tasks +from arc.job.pipe.pipe_state import TaskState, initialize_task, update_task_state from arc.level import Level from arc.plotter import save_conformers_file from arc.scheduler import (COLLAPSED_REFERENCE_MESSAGE, INVALID_ANALYTIC_FREQ_MESSAGE, MAX_S_SQUARED_DEVIATION, @@ -338,11 +342,55 @@ def _species_with_rotor(self, label, success=None, scan_path=None, invalidation_ 'invalidation_reason': invalidation_reason, 'directed_scan_type': 'ess'}} return spc - def _restart_scheduler(self, species_list, **patched): + def _stage_rotor_scan_pipe_task(self, sched, label, rotor_index, status=TaskState.PENDING): + """ + Stage a 1D rotor scan pipe task on disk as a previous session's pipe run would have left it. + + This goes through the real ``PipeCoordinator`` and ``pipe_state`` writers, so the directory layout + the restart loop looks for is the one the pipe system actually writes rather than a replica of it: + a change to either side of that convention turns these tests red instead of silently detaching the + restart guard from the pipe runs it exists to find. + + Args: + sched (Scheduler): A scheduler supplying the project directory and the species to stage under. + label (str): The species label. + rotor_index (int): The rotor's index in the species' ``rotors_dict``. + status (TaskState, optional): The state to walk the staged task to. + + Returns: + str: The ``pipe_root`` the task was staged under. + """ + tasks = build_rotor_scan_1d_tasks(sched.species_dict[label], label, [rotor_index], + {'method': 'wb97xd', 'basis': 'def2tzvp'}, 'gaussian', 14000) + pipe_root = PipeCoordinator(sched)._compute_pipe_root(f'{label}_scan_1d', tasks) + initialize_task(pipe_root, tasks[0]) + now = time.time() + lifecycle = [(TaskState.CLAIMED, {'claimed_by': 'worker-0', 'claim_token': 'token', + 'claimed_at': now, 'lease_expires_at': now + 3600}), + (TaskState.RUNNING, {'started_at': now}), + (TaskState.COMPLETED, {'ended_at': now})] + if status != TaskState.PENDING: + statuses = [task_state for task_state, _ in lifecycle] + if status not in statuses: + raise ValueError(f'Cannot stage a task in {status}, only PENDING or one of {statuses}') + for next_status, fields in lifecycle[:statuses.index(status) + 1]: + update_task_state(pipe_root, tasks[0].task_id, next_status, **fields) + return pipe_root + + def _restart_scheduler(self, species_list, converged=(), stage=None, **patched): """ Construct a non-testing Scheduler from a restart dict in which every species already has a geometry (so the restart loop in ``__init__`` runs), with ``schedule_jobs`` and the given methods patched out. + Args: + species_list (list): The species to restart. + converged (tuple, optional): Labels whose saved ``output[label]['convergence']`` is ``True``, + i.e. species the previous session had already finished. + stage (callable, optional): Called with a ``testing=True`` Scheduler before the restarted one is + constructed, to lay down on-disk state that a previous session would + have left in the project directory (e.g. a staged pipe run). + **patched: Scheduler method names to patch out. + Returns: dict: The mocks, keyed by method name (``schedule_jobs`` included). """ @@ -359,9 +407,14 @@ def _restart_scheduler(self, species_list, **patched): job_types=job_types, orbitals_level=default_levels_of_theory['orbitals'], adaptive_levels=None) self.addCleanup(shutil.rmtree, kwargs['project_directory'], ignore_errors=True) - output = Scheduler(testing=True, **kwargs).output + testing_sched = Scheduler(testing=True, **kwargs) + output = testing_sched.output for spc_output in output.values(): spc_output['paths'].update({'geo': 'geo.out', 'freq': 'freq.out', 'sp': 'sp.out'}) + for label in converged: + output[label]['convergence'] = True + if stage is not None: + stage(testing_sched) mocks = dict() with patch.object(Scheduler, 'schedule_jobs') as schedule_jobs: mocks['schedule_jobs'] = schedule_jobs @@ -390,6 +443,84 @@ def test_restart_still_scans_a_failed_regular_rotor(self): mocks = self._restart_scheduler([regular, irc], run_scan_jobs=None) self.assertEqual([c.args for c in mocks['run_scan_jobs'].call_args_list], [('regular',)]) + def test_restart_scans_a_never_scanned_rotor(self): + """A rotor with no result and no scan_path (e.g. added by hand to restart.yml) must be spawned on restart""" + regular = self._species_with_rotor('regular', success=None, scan_path=None) + mocks = self._restart_scheduler([regular], run_scan_jobs=None) + self.assertEqual([c.args for c in mocks['run_scan_jobs'].call_args_list], [('regular',)]) + + def test_restart_scans_a_never_scanned_rotor_of_a_converged_species(self): + """A pending rotor must be spawned even when the species was saved as converged. + + This is the case the fix above misses: ``output[label]['convergence']`` is restored verbatim from + restart.yml, and the `continue` it triggers sits BEFORE the rotor-spawn condition, so a rotor added + by hand to a finished species is skipped exactly as it was before the fix -- and the species is then + reported as converged while missing a mode, which is the failure the fix exists to prevent. + """ + regular = self._species_with_rotor('regular', success=None, scan_path=None) + mocks = self._restart_scheduler([regular], converged=('regular',), run_scan_jobs=None) + self.assertEqual([c.args for c in mocks['run_scan_jobs'].call_args_list], [('regular',)]) + + def test_restart_does_not_rescan_a_scanned_or_invalidated_rotor(self): + """Rotors that already have a result, or are invalidated, are not spawned again on restart""" + done = self._species_with_rotor('done', success=True, scan_path=__file__) + invalidated = self._species_with_rotor('invalidated', success=False, invalidation_reason='not a torsion') + mocks = self._restart_scheduler([done, invalidated], run_scan_jobs=None) + self.assertEqual(mocks['run_scan_jobs'].call_args_list, list()) + + def test_restart_does_not_duplicate_a_rotor_scan_running_in_a_pipe(self): + """ + A rotor whose scan is in flight in a pipe run must not be spawned again as a standalone ESS job. + + A pipe-queued rotor carries exactly the state the restart loop reads as 'never scanned' - no result + and no scan_path - and pipe runs are not restored on restart: the planner does not exist yet when the + restart loop runs, and active_pipes starts empty, so has_pending_pipe_work() cannot see it. Without + the staged-pipe guard, run_scan_jobs() submits a standalone duplicate of a job that is still queued. + """ + regular = self._species_with_rotor('regular', success=None, scan_path=None) + mocks = self._restart_scheduler( + [regular], run_scan_jobs=None, + stage=lambda sched: self._stage_rotor_scan_pipe_task(sched, 'regular', 0, TaskState.RUNNING)) + self.assertEqual(mocks['run_scan_jobs'].call_args_list, list()) + + def test_restart_scans_a_rotor_whose_pipe_task_already_terminated(self): + """ + A terminated pipe task is no reason to skip the rotor: it still owes a scan. + + Nothing ingests a COMPLETED task's results once its pipe run is gone, so the rotor would otherwise + be left unscanned while the species is reported as converged - the very failure this restart spawn + exists to prevent - and a standalone scan duplicates no running job. + """ + regular = self._species_with_rotor('regular', success=None, scan_path=None) + mocks = self._restart_scheduler( + [regular], run_scan_jobs=None, + stage=lambda sched: self._stage_rotor_scan_pipe_task(sched, 'regular', 0, TaskState.COMPLETED)) + self.assertEqual([c.args for c in mocks['run_scan_jobs'].call_args_list], [('regular',)]) + + def test_restart_treats_an_unreadable_pipe_task_state_as_still_owed(self): + """A pipe task whose state cannot be read must not be assumed finished. + + Reading it as terminal would submit a standalone scan of a rotor that may still be queued, which is + the expensive direction to be wrong in: the cheap one leaves the rotor unscanned, as it was before. + """ + def stage(sched): + pipe_root = self._stage_rotor_scan_pipe_task(sched, 'regular', 0) + with open(os.path.join(pipe_root, 'tasks', 'regular_scan_r0', 'state.json'), 'w') as f: + f.write('{"status": ') # truncated, as a session killed mid-write would leave it + regular = self._species_with_rotor('regular', success=None, scan_path=None) + mocks = self._restart_scheduler([regular], run_scan_jobs=None, stage=stage) + self.assertEqual(mocks['run_scan_jobs'].call_args_list, list()) + + def test_restart_pipe_guard_is_per_rotor_not_per_species(self): + """A pipe task staged for one rotor must not suppress the restart scan of the species' other rotor""" + regular = self._species_with_rotor('regular', success=None, scan_path=None) + regular.rotors_dict[1] = dict(regular.rotors_dict[0], pivots=[1, 3], torsion=[2, 0, 1, 6]) + regular.number_of_rotors = 2 + mocks = self._restart_scheduler( + [regular], run_scan_jobs=None, + stage=lambda sched: self._stage_rotor_scan_pipe_task(sched, 'regular', 1, TaskState.RUNNING)) + self.assertEqual([c.args for c in mocks['run_scan_jobs'].call_args_list], [('regular',)]) + def test_check_negative_freq(self): """Test the check_negative_freq() method""" label = 'C2H6' From 345f05dcc32eef33693059030b0cdf8f8092e23e Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:21:21 +0200 Subject: [PATCH 05/13] scheduler: allow rotor scans before the pipe planner exists The restart loop in __init__ calls run_scan_jobs(), but the pipe planner is only constructed after that loop, so a restart with a rotor to spawn died with AttributeError. Pipe batching is an optimization; without a planner, scan each rotor individually. Co-Authored-By: Claude --- arc/scheduler.py | 8 ++++++-- arc/scheduler_test.py | 10 ++++++++++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/arc/scheduler.py b/arc/scheduler.py index 593fe20643..2c7202fd5b 100644 --- a/arc/scheduler.py +++ b/arc/scheduler.py @@ -1905,8 +1905,12 @@ def run_scan_jobs(self, label: str): ess_rotor_indices.append(i) # Attempt to batch ESS scans through pipe mode; fall back per-rotor for the rest. - piped_rotors = self.pipe_planner.try_pipe_rotor_scans_1d(label, ess_rotor_indices) \ - if ess_rotor_indices else set() + # run_scan_jobs() is also called from the restart loop in __init__, which runs before the + # pipe planner is constructed. Batching is an optimization, so without a planner every + # rotor is scanned individually. + pipe_planner = getattr(self, 'pipe_planner', None) + piped_rotors = pipe_planner.try_pipe_rotor_scans_1d(label, ess_rotor_indices) \ + if ess_rotor_indices and pipe_planner is not None else set() for i in ess_rotor_indices: if i in piped_rotors: continue diff --git a/arc/scheduler_test.py b/arc/scheduler_test.py index 6800332bb1..aa993b70f2 100644 --- a/arc/scheduler_test.py +++ b/arc/scheduler_test.py @@ -520,6 +520,16 @@ def test_restart_pipe_guard_is_per_rotor_not_per_species(self): [regular], run_scan_jobs=None, stage=lambda sched: self._stage_rotor_scan_pipe_task(sched, 'regular', 1, TaskState.RUNNING)) self.assertEqual([c.args for c in mocks['run_scan_jobs'].call_args_list], [('regular',)]) + def test_restart_spawns_a_rotor_scan_before_the_pipe_planner_exists(self): + """ + The restart loop in __init__ calls run_scan_jobs() before the pipe planner is constructed, so a restart + with a rotor to spawn used to die with AttributeError. Without a planner each rotor is scanned alone. + """ + regular = self._species_with_rotor('regular', success=None, scan_path=None) + mocks = self._restart_scheduler([regular], run_job=None) + self.assertEqual(mocks['run_job'].call_count, 1) + self.assertEqual(mocks['run_job'].call_args.kwargs['job_type'], 'scan') + self.assertEqual(mocks['run_job'].call_args.kwargs['rotor_index'], 0) def test_check_negative_freq(self): """Test the check_negative_freq() method""" From 78d7932a79b69d5ca4c8783c8f3ca0e3ae3e5821 Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:22:05 +0200 Subject: [PATCH 06/13] scheduler: drop superseded conf_opt jobs when respawning TS conformers run_ts_conformer_jobs() wipes job_dict['conf_opt'], which orphans any conf_opt name restored into running_jobs from a previous session. save_restart_dict() maps each such name back through job_dict and raised KeyError on the first one that was gone. The jobs are superseded by the conformers spawned right there, so stop tracking them. Co-Authored-By: Claude --- arc/scheduler.py | 5 +++++ arc/scheduler_test.py | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+) diff --git a/arc/scheduler.py b/arc/scheduler.py index 2c7202fd5b..8560caaeb4 100644 --- a/arc/scheduler.py +++ b/arc/scheduler.py @@ -1620,6 +1620,11 @@ def run_ts_conformer_jobs(self, label: str): piped_indices = self.pipe_planner.try_pipe_ts_opt(label, xyzs, self.ts_guess_level) if not piped_indices: self.job_dict[label]['conf_opt'] = dict() + # Wiping job_dict['conf_opt'] orphans any conf_opt job name restored into running_jobs from a + # previous session, and save_restart_dict() cannot map such a name back to a job. Those jobs + # are superseded by the conformers spawned below, so stop tracking them. + self.running_jobs[label] = [job_name for job_name in self.running_jobs.get(label, list()) + if 'conf_opt' not in job_name] for i, tsg in enumerate(successful_tsgs): tsg.conformer_index = i # Store the conformer index to match them later. if i in piped_indices: diff --git a/arc/scheduler_test.py b/arc/scheduler_test.py index aa993b70f2..5264da56a9 100644 --- a/arc/scheduler_test.py +++ b/arc/scheduler_test.py @@ -2950,6 +2950,41 @@ def test_run_ts_conformer_jobs_single_success_provenance(self, mock_run_opt): self.assertEqual(sched.output['TS_single']['paths']['gsm'], good.log_path) self.assertEqual(sched.output['TS_single']['paths']['neb'], '') + @patch('arc.scheduler.Scheduler.run_job') + def test_run_ts_conformer_jobs_drops_orphaned_conf_opt_running_jobs(self, mock_run_job): + """Respawning the TS conformers wipes job_dict['conf_opt'], so restored conf_opt names must go as well.""" + xyz_a = str_to_xyz("""N 0.91779059 0.51946178 0.00000000 + H 1.81402049 1.03819414 0.00000000 + H 0.00000000 0.00000000 0.00000000 + H 0.91779059 1.22790192 0.72426890""") + xyz_b = str_to_xyz("""N 0.91779059 0.51946178 0.00000000 + H 2.51402049 1.03819414 0.00000000 + H 0.00000000 0.00000000 0.00000000 + H 0.91779059 1.22790192 0.72426890""") + label = 'TS_respawn' + ts_spc = ARCSpecies(label=label, is_ts=True, multiplicity=1, charge=0, compute_thermo=False) + ts_spc.ts_guesses = [TSGuess(index=0, method='heuristics', success=True, xyz=xyz_a), + TSGuess(index=1, method='heuristics', success=True, xyz=xyz_b)] + project_directory = os.path.join(ARC_PATH, 'Projects', 'arc_project_for_testing_delete_after_usage_ts_respawn') + self.addCleanup(shutil.rmtree, project_directory, ignore_errors=True) + sched = Scheduler(project='project_test_ts_respawn', ess_settings=self.ess_settings, + species_list=[ts_spc], + opt_level=Level(repr=default_levels_of_theory['opt']), + freq_level=Level(repr=default_levels_of_theory['freq']), + sp_level=Level(repr=default_levels_of_theory['sp']), + ts_guess_level=Level(repr=default_levels_of_theory['ts_guesses']), + project_directory=project_directory, + testing=True, + job_types=self.job_types1, + ) + sched.pipe_planner = MagicMock() + sched.pipe_planner.try_pipe_ts_opt.return_value = set() + sched.job_dict[label] = {'conf_opt': {0: MagicMock(), 1: MagicMock()}} + sched.running_jobs[label] = ['conf_opt_0', 'conf_opt_1', 'freq_a7'] + sched.run_ts_conformer_jobs(label=label) + self.assertEqual(sched.running_jobs[label], ['freq_a7']) + self.assertEqual(mock_run_job.call_count, 2) + @patch('arc.scheduler.Scheduler.run_opt_job') def test_switch_ts_cleanup(self, mock_run_opt): """Test that switch_ts resets job_types, convergence, cleans up IRC species, and clears pending pipes.""" From d69f77d1f7e2b1f44aec9f3615d922f39c3a9be4 Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:22:33 +0200 Subject: [PATCH 07/13] scheduler: skip restored jobs of species missing from the project A restored job may reference a species that no longer exists in the session, for example a TS adapter job carrying an auto-generated '_TS' label. Restoring it raised KeyError and killed the run over one stale job; warn and skip it instead. Co-Authored-By: Claude --- arc/scheduler.py | 14 ++++++++++++++ arc/scheduler_test.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/arc/scheduler.py b/arc/scheduler.py index 8560caaeb4..62c6227284 100644 --- a/arc/scheduler.py +++ b/arc/scheduler.py @@ -5272,9 +5272,23 @@ def restore_running_jobs(self): 'Assuming all jobs have finished.') else: for spc_label in jobs.keys(): + if spc_label not in self.species_dict: + # An entry left by a species that is no longer in the project. Creating a + # running_jobs key for it would keep schedule_jobs() looping forever. + logger.warning(f'Not restoring the running jobs of {spc_label}: it is not in this project.') + continue if spc_label not in self.running_jobs.keys(): self.running_jobs[spc_label] = list() for job_description in jobs[spc_label]: + # A job may reference a species that no longer exists in this session, e.g. a TS adapter + # job carrying an auto-generated '_TS' label. Restoring it would raise a KeyError + # below and kill the run over one stale job, so skip it. + if job_description.get('species_labels') \ + and any(label not in self.species_dict for label in job_description['species_labels']): + logger.warning(f"Not restoring job {job_description.get('job_name')} of {spc_label}: it " + f"references species {job_description['species_labels']}, and at least one " + f"of those is not in this project. Assuming the job is no longer running.") + continue if ('conformer' not in job_description or job_description['conformer'] is None) \ and ('tsg' not in job_description or job_description['tsg'] is None): self.running_jobs[spc_label].append(job_description['job_name']) diff --git a/arc/scheduler_test.py b/arc/scheduler_test.py index 5264da56a9..361a9b7366 100644 --- a/arc/scheduler_test.py +++ b/arc/scheduler_test.py @@ -531,6 +531,38 @@ def test_restart_spawns_a_rotor_scan_before_the_pipe_planner_exists(self): self.assertEqual(mocks['run_job'].call_args.kwargs['job_type'], 'scan') self.assertEqual(mocks['run_job'].call_args.kwargs['rotor_index'], 0) + def test_restore_running_jobs_skips_a_job_of_an_unknown_species(self): + """A restored job referencing a species label that is not in this project is skipped, not a KeyError""" + label = 'methylamine' + spc = ARCSpecies(label=label, smiles='CN', xyz=self._METHYLAMINE_XYZ) + project_directory = tempfile.mkdtemp(prefix='arc_test_restore_unknown_') + self.addCleanup(shutil.rmtree, project_directory, ignore_errors=True) + sched = Scheduler(project='project_test_restore_unknown', ess_settings=self.ess_settings, + species_list=[spc], composite_method=None, + conformer_opt_level=Level(repr=default_levels_of_theory['conformer']), + opt_level=Level(repr=default_levels_of_theory['opt']), + freq_level=Level(repr=default_levels_of_theory['freq']), + sp_level=Level(repr=default_levels_of_theory['sp']), + scan_level=Level(repr=default_levels_of_theory['scan']), + ts_guess_level=Level(repr=default_levels_of_theory['ts_guesses']), + project_directory=project_directory, testing=True, + job_types=self.job_types1, + orbitals_level=default_levels_of_theory['orbitals'], adaptive_levels=None) + opt_job = job_factory(job_adapter='gaussian', project='project_test_restore_unknown', + ess_settings=self.ess_settings, species=[spc], xyz=self._METHYLAMINE_XYZ, + job_type='opt', level=Level(repr={'method': 'wb97xd', 'basis': 'def2svp'}), + project_directory=project_directory, job_num=903) + stale = dict(opt_job.as_dict(), job_name='tsg9', species_labels=['rxn_label_TS']) + sched.restart_dict = {'running_jobs': {label: [stale, opt_job.as_dict()], + 'removed_species': [opt_job.as_dict()]}} + sched.running_jobs = dict() + sched.job_dict = dict() + sched.restore_running_jobs() + self.assertEqual(sched.running_jobs[label], [opt_job.job_name]) + self.assertEqual(list(sched.job_dict[label]['opt'].keys()), [opt_job.job_name]) + # No running_jobs key for a species that is not in the project; schedule_jobs() would never clear it. + self.assertNotIn('removed_species', sched.running_jobs) + def test_check_negative_freq(self): """Test the check_negative_freq() method""" label = 'C2H6' From c4b830b74a7eea7f5468b305143c38807202cea4 Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:22:49 +0200 Subject: [PATCH 08/13] scheduler: skip untracked conformer jobs when saving the restart file save_restart_dict() looked up every conf_opt, conf_sp and tsg name in running_jobs directly in job_dict and raised KeyError for a name with no entry, killing the run over one untracked job. Warn and leave it out of the restart file instead. Co-Authored-By: Claude --- arc/scheduler.py | 25 +++++++++++++++++++------ arc/scheduler_test.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 6 deletions(-) diff --git a/arc/scheduler.py b/arc/scheduler.py index 62c6227284..57b9faf27a 100644 --- a/arc/scheduler.py +++ b/arc/scheduler.py @@ -5373,16 +5373,29 @@ def save_restart_dict(self): self.restart_dict['running_jobs'] = dict() for spc in self.species_dict.values(): if spc.label in self.running_jobs: + # A conformer or TS-guess name in running_jobs whose job object is not in job_dict + # cannot be serialized or polled. Raising here would kill the run over one untracked + # job, so warn and drop it from both the restart file and running_jobs. + def _conformer_job_dicts(key, names): + dicts = list() + for job_name in names: + job = self.job_dict[spc.label].get(key, dict()).get(get_i_from_job_name(job_name)) + if job is None: + logger.warning(f'Job {job_name} of species {spc.label} is listed as running ' + f'but has no {key} entry in job_dict; dropping it.') + self.running_jobs[spc.label].remove(job_name) + continue + dicts.append(job.as_dict()) + return dicts self.restart_dict['running_jobs'][spc.label] = \ [self.job_dict[spc.label][job_name.rsplit('_', 1)[0]][job_name].as_dict() for job_name in self.running_jobs[spc.label] if not is_conformer_job(job_name) and 'tsg' not in job_name] \ - + [self.job_dict[spc.label]['conf_opt'][get_i_from_job_name(job_name)].as_dict() - for job_name in self.running_jobs[spc.label] if 'conf_opt' in job_name] \ - + [self.job_dict[spc.label]['conf_sp'][get_i_from_job_name(job_name)].as_dict() - for job_name in self.running_jobs[spc.label] if 'conf_sp' in job_name] \ - + [self.job_dict[spc.label]['tsg'][get_i_from_job_name(job_name)].as_dict() - for job_name in self.running_jobs[spc.label] if 'tsg' in job_name] + + _conformer_job_dicts('conf_opt', [n for n in self.running_jobs[spc.label] + if 'conf_opt' in n]) \ + + _conformer_job_dicts('conf_sp', [n for n in self.running_jobs[spc.label] + if 'conf_sp' in n]) \ + + _conformer_job_dicts('tsg', [n for n in self.running_jobs[spc.label] if 'tsg' in n]) save_yaml_file(path=self.restart_path, content=self.restart_dict) def make_reaction_labels_info_file(self): diff --git a/arc/scheduler_test.py b/arc/scheduler_test.py index 361a9b7366..a0f3e9b3f7 100644 --- a/arc/scheduler_test.py +++ b/arc/scheduler_test.py @@ -563,6 +563,37 @@ def test_restore_running_jobs_skips_a_job_of_an_unknown_species(self): # No running_jobs key for a species that is not in the project; schedule_jobs() would never clear it. self.assertNotIn('removed_species', sched.running_jobs) + def test_save_restart_dict_skips_untracked_conformer_jobs(self): + """A conf_opt/conf_sp/tsg name in running_jobs with no job_dict entry is skipped with a warning""" + label = 'methylamine' + spc = ARCSpecies(label=label, smiles='CN', xyz=self._METHYLAMINE_XYZ) + project_directory = tempfile.mkdtemp(prefix='arc_test_save_restart_') + self.addCleanup(shutil.rmtree, project_directory, ignore_errors=True) + sched = Scheduler(project='project_test_save_restart', ess_settings=self.ess_settings, + species_list=[spc], composite_method=None, + conformer_opt_level=Level(repr=default_levels_of_theory['conformer']), + opt_level=Level(repr=default_levels_of_theory['opt']), + freq_level=Level(repr=default_levels_of_theory['freq']), + sp_level=Level(repr=default_levels_of_theory['sp']), + scan_level=Level(repr=default_levels_of_theory['scan']), + ts_guess_level=Level(repr=default_levels_of_theory['ts_guesses']), + project_directory=project_directory, testing=True, + job_types=self.job_types1, + orbitals_level=default_levels_of_theory['orbitals'], adaptive_levels=None) + tracked = MagicMock() + tracked.as_dict.return_value = {'job_name': 'conf_opt_0'} + sched.job_dict[label] = {'conf_opt': {0: tracked}} + sched.running_jobs[label] = ['conf_opt_0', 'conf_opt_5', 'conf_sp_2', 'tsg3'] + sched.restart_dict = dict() + sched.save_restart = True + with self.assertLogs('arc', level='WARNING') as logs: + sched.save_restart_dict() + self.assertEqual(sched.restart_dict['running_jobs'][label], [{'job_name': 'conf_opt_0'}]) + self.assertEqual(len(logs.output), 3) + # The untracked names leave running_jobs too, so the next poll does not KeyError on them. + self.assertEqual(sched.running_jobs[label], ['conf_opt_0']) + sched.get_completed_incore_jobs() + def test_check_negative_freq(self): """Test the check_negative_freq() method""" label = 'C2H6' From 98eab88d29534084178372d4f9436ddaf94b3a57 Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:21:19 +0200 Subject: [PATCH 09/13] ssh: report an unparsable job ID from a known scheduler as an errored job When a recognized scheduler (OGE/SGE, Slurm, PBS, HTCondor) answered a submission with nothing parsable on stdout, submit_job fell through to ValueError('Unrecognized cluster software: ...'). That killed the whole ARC process over one rejected job, and blamed the scheduler name rather than the submission. Log stdout and stderr and return an errored job instead, so the scheduler's troubleshooting decides whether to resubmit. A cluster software ARC has no parser for still raises. Co-Authored-By: Claude --- arc/job/ssh.py | 11 ++++++++++- arc/job/ssh_test.py | 37 +++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/arc/job/ssh.py b/arc/job/ssh.py index 39b0414b0a..0b477e10a6 100644 --- a/arc/job/ssh.py +++ b/arc/job/ssh.py @@ -501,8 +501,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 diff --git a/arc/job/ssh_test.py b/arc/job/ssh_test.py index dc4e9f1464..2bc4f7faae 100644 --- a/arc/job/ssh_test.py +++ b/arc/job/ssh_test.py @@ -1133,5 +1133,42 @@ def test_the_cleanup_connects_with_a_single_attempt(self): self.assertEqual(attempts, [1]) + +class TestSubmitJob(unittest.TestCase): + """submit_job() turns a scheduler's reply into a (status, job ID) pair without killing the run.""" + + def make_client(self, cluster_soft: str) -> ssh.SSHClient: + """Build an SSHClient for a server running ``cluster_soft``.""" + server = {'address': 'host.example.edu', 'un': 'user', 'key': '/dev/null', 'cluster_soft': cluster_soft} + with patch.object(ssh, 'servers', {'srv': server}): + client = ssh.SSHClient('srv') + patcher = patch.object(ssh, 'servers', {'srv': server}) + patcher.start() + self.addCleanup(patcher.stop) + return client + + def submit(self, client: ssh.SSHClient, replies: list): + """Submit through ``client``, answering each command with the next (stdout, stderr) reply.""" + send = MagicMock(side_effect=replies) + with patch.object(client, '_send_command_to_server', send), \ + patch.object(ssh.time, 'sleep') as sleep: + result = client.submit_job(remote_path='/remote/job') + return result, send, sleep + + def test_unparsable_pbs_reply_is_an_errored_job(self): + """An empty stdout from a known scheduler is a failed submission, not an unknown cluster.""" + client = self.make_client('PBS') + (status, job_id), send, _ = self.submit(client, [([], ['qsub: Job rejected by all possible destinations'])]) + self.assertEqual(status, 'errored') + self.assertFalse(job_id) + send.assert_called_once() + + def test_unknown_cluster_software_still_raises(self): + """A cluster software ARC cannot parse at all is still reported as unrecognized.""" + client = self.make_client('LSF') + with patch.dict(ssh.submit_command, {'LSF': 'bsub'}), patch.dict(ssh.submit_filenames, {'LSF': 'submit.sh'}): + with self.assertRaisesRegex(ValueError, 'Unrecognized cluster software: LSF'): + self.submit(client, [(['Job <1> is submitted'], [])]) + if __name__ == '__main__': unittest.main(testRunner=unittest.TextTestRunner(verbosity=2)) From 1fd6f69567e9fc1efa49eaf6ce9c3e2922940c1c Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:21:38 +0200 Subject: [PATCH 10/13] ssh: retry a submission refused by a full PBS queue, once, and keep its result PBS refuses qsub with "Maximum number of jobs already in queue" when a user is over the queue cap. Treat it like Slurm's AssocMaxSubmitJobLimit: wait five minutes and resubmit. Two defects in that retry path are fixed with it: the resubmission's result was discarded (the outer call parsed its own empty stdout), and a resubmission that was refused again could sleep and recurse again while the whole scheduler waited. Return the retry's (status, job ID), and give up as an errored job after the single retry. Co-Authored-By: Claude --- arc/job/ssh.py | 17 ++++++++++++----- arc/job/ssh_test.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 5 deletions(-) diff --git a/arc/job/ssh.py b/arc/job/ssh.py index 0b477e10a6..0d9ae9af78 100644 --- a/arc/job/ssh.py +++ b/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] diff --git a/arc/job/ssh_test.py b/arc/job/ssh_test.py index 2bc4f7faae..8c5fb73cbb 100644 --- a/arc/job/ssh_test.py +++ b/arc/job/ssh_test.py @@ -1163,6 +1163,37 @@ def test_unparsable_pbs_reply_is_an_errored_job(self): self.assertFalse(job_id) send.assert_called_once() + def test_full_pbs_queue_is_retried_once_and_the_retry_result_returned(self): + """PBS refuses a submission over the user's queue cap; ARC waits once and resubmits.""" + client = self.make_client('PBS') + (status, job_id), send, sleep = self.submit( + client, [([], ['qsub: Maximum number of jobs already in queue for user MSG=total number of jobs in ' + 'queue exceeds the queue limit: user user@host, queue main_q']), + (['123456.pbs-server'], [])]) + self.assertEqual((status, job_id), ('running', '123456')) + self.assertEqual(send.call_count, 2) + sleep.assert_called_once() + + def test_full_slurm_queue_returns_the_retry_result(self): + """A successful resubmission after AssocMaxSubmitJobLimit yields the new job's ID.""" + client = self.make_client('Slurm') + (status, job_id), send, sleep = self.submit( + client, [([], ['sbatch: error: AssocMaxSubmitJobLimit']), + (['Submitted batch job 98765'], [])]) + self.assertEqual((status, job_id), ('running', '98765')) + self.assertEqual(send.call_count, 2) + sleep.assert_called_once() + + def test_a_queue_still_full_after_the_wait_is_an_errored_job(self): + """A second refusal ends the submission as errored instead of sleeping again.""" + client = self.make_client('PBS') + refusal = ([], ['qsub: Maximum number of jobs already in queue for user']) + (status, job_id), send, sleep = self.submit(client, [refusal, refusal]) + self.assertEqual(status, 'errored') + self.assertFalse(job_id) + self.assertEqual(send.call_count, 2) + sleep.assert_called_once() + def test_unknown_cluster_software_still_raises(self): """A cluster software ARC cannot parse at all is still reported as unrecognized.""" client = self.make_client('LSF') From 166548150688cea04c9fc7d2e147d60703c1121b Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:22:57 +0200 Subject: [PATCH 11/13] autotst: exit non-zero with a traceback when AutoTST fails autotst_script.py swallowed any exception from building or labeling the AutoTST reaction with a bare `return None`. The process exited 0 and wrote no output file, so ARC logged nothing and treated the adapter as having found no TS guess: a broken AutoTST setup looked like a chemical failure. Print the traceback to stderr and exit 1, so run_in_conda_env logs it and the adapter's return-code warning fires. Co-Authored-By: Claude --- arc/job/adapters/scripts/autotst_script.py | 10 ++- .../adapters/scripts/autotst_script_test.py | 64 +++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) create mode 100644 arc/job/adapters/scripts/autotst_script_test.py diff --git a/arc/job/adapters/scripts/autotst_script.py b/arc/job/adapters/scripts/autotst_script.py index 6d4fd1fa80..3e07d6187b 100644 --- a/arc/job/adapters/scripts/autotst_script.py +++ b/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() diff --git a/arc/job/adapters/scripts/autotst_script_test.py b/arc/job/adapters/scripts/autotst_script_test.py new file mode 100644 index 0000000000..b02e7d5b50 --- /dev/null +++ b/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)) From f7b801251c802b2d607d3b824a0bd83ae1e488d9 Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:27:59 +0200 Subject: [PATCH 12/13] autotst: run the AutoTST subprocess without the caller's PYTHONPATH An RMG-Py source checkout on ARC's PYTHONPATH, compiled for arc_env's Python, shadows tst_env's installed rmgpy inside the AutoTST subprocess and breaks it. Pass strip_pythonpath=True to run_in_conda_env, as the KinBot adapter already does. AutoTST must then be importable from tst_env itself, via a .pth file or the activation hook that install_autotst.sh --conda writes; a clone that is only on PYTHONPATH through ~/.bashrc is no longer seen. Co-Authored-By: Claude --- arc/job/adapters/ts/autotst_test.py | 58 +++++++++++++++++++++++++++++ arc/job/adapters/ts/autotst_ts.py | 10 ++++- 2 files changed, 66 insertions(+), 2 deletions(-) create mode 100644 arc/job/adapters/ts/autotst_test.py diff --git a/arc/job/adapters/ts/autotst_test.py b/arc/job/adapters/ts/autotst_test.py new file mode 100644 index 0000000000..fbbb223af7 --- /dev/null +++ b/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)) diff --git a/arc/job/adapters/ts/autotst_ts.py b/arc/job/adapters/ts/autotst_ts.py index e4474edcb1..90ee525726 100644 --- a/arc/job/adapters/ts/autotst_ts.py +++ b/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 From 3a5f3591e982f163f22add3b7a9a5625ce54c0f3 Mon Sep 17 00:00:00 2001 From: Alon Grinberg Dana Date: Tue, 27 Oct 2026 22:45:09 +0200 Subject: [PATCH 13/13] autotst installer: register AutoTST with a .pth file in tst_env instead of ~/.bashrc The AutoTST subprocess now runs without the caller's PYTHONPATH, so the bashrc PYTHONPATH line written by the default install mode no longer made AutoTST importable. Write autotst.pth into tst_env's site-packages after the environment exists, leave ~/.bashrc alone and note a leftover line. Co-Authored-By: Claude --- devtools/install_autotst.sh | 59 ++++++++++++++++++++++++++----------- docs/source/TS_search.rst | 5 +++- 2 files changed, 45 insertions(+), 19 deletions(-) diff --git a/devtools/install_autotst.sh b/devtools/install_autotst.sh index 5e3bc35288..01d561dc83 100644 --- a/devtools/install_autotst.sh +++ b/devtools/install_autotst.sh @@ -12,8 +12,13 @@ while true; do cat < +write_pth () { + local env="$1" repo_path="$2" site_dir + $COMMAND_PKG env list | awk '{print $1}' | sed 's/^\*//' | grep -Fxq "$env" || return 0 + site_dir="$($COMMAND_PKG run -n "$env" python -c 'import site; print(site.getsitepackages()[0])')" + write_pth_file "$site_dir" "$repo_path" + echo "🔗 AutoTST registered in $env via $site_dir/autotst.pth" +} + +# Writes /autotst.pth containing . +write_pth_file () { + local site_dir="$1" repo_path="$2" + [[ -d "$site_dir" ]] || { echo "❌ site-packages not found: $site_dir"; return 1; } + printf '%s\n' "$repo_path" > "$site_dir/autotst.pth" +} + + # ── locate folders relative to this script ──────────────────────────────── SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)" if ARC_ROOT=$(git -C "$SCRIPT_DIR" rev-parse --show-toplevel 2>/dev/null); then @@ -113,22 +137,7 @@ else cd AutoTST fi -if [[ $MODE == "path" ]]; then - - AUTO_PATH_LINE="export PYTHONPATH=\"\$PYTHONPATH:$(pwd)\"" - if ! grep -Fqx "$AUTO_PATH_LINE" ~/.bashrc; then - echo "$AUTO_PATH_LINE" >> ~/.bashrc - echo "✔ī¸ Added AutoTST path to ~/.bashrc" - else - echo "ℹī¸ AutoTST path already exists in ~/.bashrc" - fi -elif [[ $MODE == "conda" ]]; then - write_hook tst_env "$(pwd)" - # add to arc_env if that env exists - if $COMMAND_PKG env list | awk '{print $1}' | grep -qx arc_env; then - write_hook arc_env "$(pwd)" - fi -fi +AUTOTST_DIR="$(pwd)" if $COMMAND_PKG env list | awk '{print $1}' | sed 's/^\*//' | grep -Fxq 'tst_env'; then echo ">>> Updating existing environment tst_env..." @@ -154,4 +163,18 @@ else $COMMAND_PKG install -n tst_env -c conda-forge pyyaml -y fi +# Register AutoTST only now that tst_env exists. +write_pth tst_env "$AUTOTST_DIR" +if [[ $MODE == "conda" ]]; then + write_hook tst_env "$AUTOTST_DIR" + # add to arc_env if that env exists + if $COMMAND_PKG env list | awk '{print $1}' | grep -qx arc_env; then + write_hook arc_env "$AUTOTST_DIR" + fi +fi + +if grep -Fqx "export PYTHONPATH=\"\$PYTHONPATH:$AUTOTST_DIR\"" ~/.bashrc 2>/dev/null; then + echo "ℹī¸ ~/.bashrc still has an AutoTST PYTHONPATH line from an earlier install. It is no longer needed and can be removed." +fi + echo "✅ Done installing AutoTST." diff --git a/docs/source/TS_search.rst b/docs/source/TS_search.rst index a2995372fa..9b45611988 100644 --- a/docs/source/TS_search.rst +++ b/docs/source/TS_search.rst @@ -232,7 +232,10 @@ package to generate TS guesses from RMG reaction templates. AutoTST performs systematic conformer searches of the TS using distance-geometry embedding and RDKit force-field optimization, guided by the reaction family template distances. -Runs as a subprocess. Requires the ``autotst`` conda environment. +Runs as a subprocess. Requires the ``autotst`` conda environment. ``make install-autotst`` +registers the AutoTST clone in that environment through an ``autotst.pth`` file in its +``site-packages`` (ARC runs the subprocess without the caller's ``PYTHONPATH``, so a +``PYTHONPATH`` entry in ``~/.bashrc`` is not used). KinBot (``'kinbot'``) ^^^^^^^^^^^^^^^^^^^^^