Restore live conformer jobs by name on restart, not the fossil 'conformer{i}'

The contract: a running conformer job lives in self.running_jobs under the
same name the live path emits, '{job_type}_{i}' (e.g. 'conf_opt_0'). Every
consumer parses that format -- get_i_from_job_name strips the 'conf_opt'/
'conf_sp' prefix, and get_completed_incore_jobs routes on it into
job_dict[label]['conf_opt'][i].

How it broke: restore_running_jobs emitted the fossil 'conformer{i}' instead.
get_i_from_job_name returns None for it, so get_completed_incore_jobs fell into
its fallback branch, derived an empty job-type from the underscore-less name
('conformer0'.split('_')[:-1] == []), and died with KeyError: '' on the first
scheduling sweep -- crashing every ARC restart that had a live conformer job.

The fix emits '{job_type}_{i}' from the same expression the live path uses,
rather than a second hard-coded literal that must be kept in sync by hand. The
job_dict was already reconstructed correctly (conf_opt keyed by int index), so
the name was the only defect. Also corrects get_i_from_job_name's docstring,
which still advertised the retired 'conformer12' format.

The test drives a restart payload carrying a live conf_opt job through the real
restore_running_jobs + get_completed_incore_jobs path: red with KeyError: '' on
the unfixed code, green after.

Author: alongd

arc/checks/common.py

@@ -15,6 +15,22 @@
 TS_IRC_FAILED_MARKER = 'INVALID TS (failed IRC validation)'
 
 
+def get_conformer_job_name(job_type: str, i: int) -> str:
+    """
+    Get the name of a conformer job, the inverse of ``get_i_from_job_name()``.
+    This is the single definition of the conformer job name format,
+    used both when spawning a job and when restoring one from a restart file.
+
+    Args:
+        job_type (str): The conformer job type, e.g., 'conf_opt' or 'conf_sp'.
+        i (int): The conformer index.
+
+    Returns:
+        str: The conformer job name, e.g., 'conf_opt_3'.
+    """
+    return f'{job_type}_{i}'
+
+
 def is_conformer_job(job_name: str) -> bool:
     """
     Check whether a job name represents a conformer job.
@@ -50,7 +66,7 @@ def get_i_from_job_name(job_name: str) -> int | None:
     Get the conformer or tsg index from the job name.
 
     Args:
-        job_name (str): The job name, e.g., 'conformer12' or 'tsg5'.
+        job_name (str): The job name, e.g., 'conf_opt_12', 'conf_sp_3', or 'tsg5'.
 
     Returns:
         int | None: The corresponding conformer or tsg index.

arc/checks/common_test.py

@@ -57,6 +57,16 @@ def test_get_i_from_job_name(self):
         self.assertEqual(common.get_i_from_job_name('conf_opt_3355'), 3355)
         self.assertEqual(common.get_i_from_job_name('tsg2'), 2)
 
+    def test_get_conformer_job_name(self):
+        """Test the get_conformer_job_name() function"""
+        self.assertEqual(common.get_conformer_job_name('conf_opt', 0), 'conf_opt_0')
+        self.assertEqual(common.get_conformer_job_name('conf_sp', 3), 'conf_sp_3')
+        self.assertEqual(common.get_conformer_job_name('conf_opt', 3355), 'conf_opt_3355')
+        self.assertTrue(common.is_conformer_job(common.get_conformer_job_name('conf_sp', 12)))
+        for job_type in common.CONFORMER_JOB_TYPES:
+            for i in [0, 7, 99999]:
+                self.assertEqual(common.get_i_from_job_name(common.get_conformer_job_name(job_type, i)), i)
+
     def test_is_ts_check_exempt(self):
         """
         Test the is_ts_check_exempt() function.

arc/job/adapter.py

@@ -23,6 +23,7 @@
 
 import numpy as np
 
+from arc.checks.common import get_conformer_job_name
 from arc.common import ARC_PATH, get_logger, read_yaml_file, save_yaml_file, torsions_to_scans, convert_to_hours
 from arc.exceptions import JobError
 from arc.imports import local_arc_path, settings, submit_scripts
@@ -643,7 +644,7 @@ def _set_job_number(self):
         # 2. Set other related attributes job_name and job_server_name.
         self.job_server_name = self.job_server_name or 'a' + str(self.job_num)
         if self.conformer is not None and self.job_name is None:
-            self.job_name = f'{self.job_type}_{self.conformer}'
+            self.job_name = get_conformer_job_name(self.job_type, self.conformer)
         elif self.tsg is not None and (self.job_name is None or 'tsg_a' in self.job_name):
             if self.job_name is not None:
                 logger.warning(f'Replacing job name {self.job_name} with tsg{self.conformer}')

arc/scheduler.py

@@ -17,7 +17,7 @@
 
 import arc.parser.parser as parser
 from arc import plotter
-from arc.checks.common import get_i_from_job_name, is_conformer_job, sum_time_delta
+from arc.checks.common import get_conformer_job_name, get_i_from_job_name, is_conformer_job, sum_time_delta
 from arc.checks.ts import check_imaginary_frequencies, check_ts, check_irc_species_and_rxn
 from arc.common import (extremum_list,
                         get_angle_in_180_range,
@@ -1098,7 +1098,7 @@ def run_job(self,
         elif conformer is not None:
             # Running a conformer DFT job. Append differently to job_dict.
             self.running_jobs[label] = list() if label not in self.running_jobs else self.running_jobs[label]
-            self.running_jobs[label].append(f'{job_type}_{conformer}')  # mark as a running job
+            self.running_jobs[label].append(get_conformer_job_name(job_type, conformer))  # mark as a running job
             if 'conf_opt' not in self.job_dict[label]:
                 self.job_dict[label]['conf_opt'] = dict()
             if 'conf_sp' not in self.job_dict[label] and job_type == 'conf_sp':
@@ -4131,7 +4131,10 @@ def restore_running_jobs(self):
                             and ('tsg' not in job_description or job_description['tsg'] is None):
                         self.running_jobs[spc_label].append(job_description['job_name'])
                     elif 'conformer' in job_description:
-                        self.running_jobs[spc_label].append(f'conformer{job_description["conformer"]}')
+                        # Emit the same name the live path uses (e.g. 'conf_opt_0'),
+                        # not the fossil 'conformer{i}' that no consumer of running_jobs accepts.
+                        self.running_jobs[spc_label].append(get_conformer_job_name(job_description['job_type'],
+                                                                                   job_description['conformer']))
                     elif 'tsg' in job_description:
                         self.running_jobs[spc_label].append(f'tsg{job_description["tsg"]}')
                     for species in self.species_list:
@@ -4162,9 +4165,16 @@ def restore_running_jobs(self):
                             and ('tsg' not in job_description or job_description['tsg'] is None):
                         self.job_dict[spc_label][job_description['job_type']][job_description['job_name']] = job
                     elif 'conformer' in job_description and job_description['conformer'] is not None:
+                        # File the job under its actual job_type ('conf_opt' or 'conf_sp'), the same
+                        # key the live path uses (see run_job) and the same key get_completed_incore_jobs
+                        # reads back -- filing a conf_sp job under 'conf_opt' would crash the first sweep
+                        # with KeyError: 'conf_sp'.
+                        conf_job_type = job_description['job_type']
                         if 'conf_opt' not in self.job_dict[spc_label].keys():
                             self.job_dict[spc_label]['conf_opt'] = dict()
-                        self.job_dict[spc_label]['conf_opt'][int(job_description['conformer'])] = job
+                        if conf_job_type == 'conf_sp' and 'conf_sp' not in self.job_dict[spc_label].keys():
+                            self.job_dict[spc_label]['conf_sp'] = dict()
+                        self.job_dict[spc_label][conf_job_type][int(job_description['conformer'])] = job
                         # don't generate additional conformers for this species
                         self.dont_gen_confs.append(spc_label)
                     elif 'tsg' in job_description and job_description['tsg'] is not None:

arc/scheduler_test.py

@@ -240,6 +240,73 @@ def test_conformers(self):
         self.assertEqual(lines[11], '\n')
         self.assertEqual(lines[12], 'SMILES: CC\n')
 
+    def test_restore_running_jobs_conformer_reconnects(self):
+        """Restarting with a live conformer job must reconnect to it, not crash.
+
+        Regression for the restore-path job-name contract. During normal operation a running
+        conformer job is stored in ``running_jobs`` as ``'{job_type}_{i}'`` (e.g. ``'conf_opt_0'``),
+        and every consumer of ``running_jobs`` parses that format. ``restore_running_jobs`` used to
+        emit the fossil ``'conformer{i}'`` instead, which ``get_i_from_job_name`` returns ``None``
+        for; the first scheduling sweep after a restart (``get_completed_incore_jobs``) then fell
+        into its fallback branch, derived an empty job-type from the underscore-less name, and died
+        with ``KeyError: ''``. This drives a restart payload carrying a live conformer job through
+        the real restore + sweep path and asserts the reconnection instead of the crash.
+        """
+        label = '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"""
+        spc = ARCSpecies(label=label, smiles='CN', xyz=xyz)
+        sched = Scheduler(project='project_test_restore_conf', 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=self.project_directory, testing=True,
+                          job_types=self.job_types1,
+                          orbitals_level=default_levels_of_theory['orbitals'], adaptive_levels=None)
+        # Two live conformer jobs -- a conf_opt and a conf_sp -- serialized exactly as ARC writes
+        # them into the restart file. conf_sp jobs can equally be in flight during a restart, and are
+        # routed differently on read-back (get_completed_incore_jobs reads job_dict[label]['conf_sp']).
+        conf_opt_job = job_factory(job_adapter='gaussian', project='project_test_restore_conf',
+                                   ess_settings=self.ess_settings, species=[spc], xyz=xyz,
+                                   job_type='conf_opt', conformer=0,
+                                   level=Level(repr={'method': 'wb97xd', 'basis': 'def2svp'}),
+                                   project_directory=self.project_directory, job_num=901)
+        conf_sp_job = job_factory(job_adapter='gaussian', project='project_test_restore_conf',
+                                  ess_settings=self.ess_settings, species=[spc], xyz=xyz,
+                                  job_type='conf_sp', conformer=0,
+                                  level=Level(repr={'method': 'wb97xd', 'basis': 'def2svp'}),
+                                  project_directory=self.project_directory, job_num=902)
+        sched.restart_dict = {'running_jobs': {label: [conf_opt_job.as_dict(), conf_sp_job.as_dict()]}}
+        sched.running_jobs = dict()
+        sched.job_dict = dict()
+
+        sched.restore_running_jobs()
+        # Each conformer job is filed under its own job_type keyed by its integer index -- a conf_sp
+        # job under 'conf_sp', not 'conf_opt'. Filing conf_sp under 'conf_opt' would crash the sweep
+        # below with KeyError: 'conf_sp'.
+        self.assertIn('conf_opt', sched.job_dict[label])
+        self.assertIn(0, sched.job_dict[label]['conf_opt'])
+        self.assertIn('conf_sp', sched.job_dict[label])
+        self.assertIn(0, sched.job_dict[label]['conf_sp'])
+
+        # The first scheduling sweep after a restart reproduces the production crash on the unfixed
+        # code: get_i_from_job_name('conformer0') is None, the fallback derives an empty job-type
+        # from the underscore-less name, and self.job_dict[label][''] raises KeyError: ''.
+        sched.get_completed_incore_jobs()
+        self.assertEqual(sched.completed_incore_jobs, list())
+
+        # 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'])
+
     def test_check_negative_freq(self):
         """Test the check_negative_freq() method"""
         label = 'C2H6'