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

@@ -43,7 +43,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/scheduler.py

@@ -4060,7 +4060,9 @@ 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 '{job_type}_{conformer}' 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(f'{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:
@@ -4091,9 +4093,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'