Merge branch 'main' into xTB

Author: alongd

arc/checks/common.py

@@ -7,6 +7,21 @@
 
 from typing import List, Optional
 
+CONFORMER_JOB_TYPES = ('conf_opt', 'conf_sp')
+
+
+def is_conformer_job(job_name: str) -> bool:
+    """
+    Check whether a job name represents a conformer job.
+
+    Args:
+        job_name (str): The job name, e.g., 'conf_opt_3' or 'conf_sp_0'.
+
+    Returns:
+        bool: ``True`` if the job name starts with a conformer job type prefix.
+    """
+    return job_name.startswith(CONFORMER_JOB_TYPES)
+
 
 def sum_time_delta(timedelta_list: List[datetime.timedelta]) -> datetime.timedelta:
     """
@@ -36,10 +51,10 @@ def get_i_from_job_name(job_name: str) -> Optional[int]:
         Optional[int]: The corresponding conformer or tsg index.
     """
     i = None
-    if 'conf_opt' in job_name:
-        i = int(job_name[9:])
-    elif 'conf_sp' in job_name:
-        i = int(job_name[8:])
-    elif 'tsg' in job_name:
+    for prefix in CONFORMER_JOB_TYPES:
+        if job_name.startswith(prefix):
+            i = int(job_name[len(prefix) + 1:])  # +1 for the '_' separator
+            return i
+    if job_name.startswith('tsg'):
         i = int(job_name[3:])
     return i

arc/job/pipe/pipe_run.py

@@ -363,20 +363,11 @@ def reconcile(self) -> Dict[str, int]:
                     logger.debug(f'Could not promote task {task_id} to FAILED_TERMINAL '
                                  f'(lock contention or concurrent state change): {e}')
 
-        # Only flag resubmission for genuinely retried tasks (attempt_index > 0).
-        # Fresh PENDING tasks (attempt_index == 0) are waiting for the initial
-        # submission's workers to start — don't resubmit for those.
-        # After a resubmission, allow a grace period for workers to start before
-        # flagging again (prevents duplicate submissions).
-        active_after_retry = counts[TaskState.CLAIMED.value] + counts[TaskState.RUNNING.value]
-        resubmit_grace = 120  # seconds
-        time_since_submit = (now - self.submitted_at) if self.submitted_at else float('inf')
-        if retried_pending > 0 and active_after_retry == 0 and time_since_submit > resubmit_grace:
-            self._needs_resubmission = True
-            logger.info(f'Pipe run {self.run_id}: {retried_pending} retried tasks '
-                        f'need workers. Resubmission needed.')
-        else:
-            self._needs_resubmission = False
+        # Never resubmit a new scheduler job for retried tasks.
+        # Workers still in the scheduler queue (PBS Q state) will claim
+        # retried PENDING tasks when they start.  If the scheduler job
+        # was killed, that is a manual intervention issue.
+        self._needs_resubmission = False
 
         terminal = (counts[TaskState.COMPLETED.value]
                     + counts[TaskState.FAILED_ESS.value]

arc/job/pipe/pipe_run_test.py

@@ -267,6 +267,44 @@ def test_terminal_run_not_regressed(self):
         self.assertEqual(run.status, PipeRunState.COMPLETED)
 
 
+class TestPipeRunNoResubmission(unittest.TestCase):
+    """Pipe runs must never flag resubmission — Q-state workers handle retried tasks."""
+
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp(prefix='pipe_run_resub_')
+
+    def tearDown(self):
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def test_never_resubmit_even_with_retried_tasks_and_no_workers(self):
+        """Even when all workers are done and retried tasks remain,
+        needs_resubmission must stay False — no automatic resubmission."""
+        tasks = [_make_spec(f't{i}') for i in range(3)]
+        run = PipeRun(project_directory=self.tmpdir, run_id='resub',
+                      tasks=tasks, cluster_software='slurm', max_attempts=3)
+        run.stage()
+        run.submitted_at = time.time() - 300
+        run.status = PipeRunState.SUBMITTED
+        # All 3 workers started: t0 completed, t1 failed, t2 completed
+        now = time.time()
+        for tid in ('t0', 't2'):
+            update_task_state(run.pipe_root, tid, new_status=TaskState.CLAIMED,
+                              claimed_by='w', claim_token='tok', claimed_at=now,
+                              lease_expires_at=now + 300)
+            update_task_state(run.pipe_root, tid, new_status=TaskState.RUNNING, started_at=now)
+            update_task_state(run.pipe_root, tid, new_status=TaskState.COMPLETED, ended_at=now)
+        update_task_state(run.pipe_root, 't1', new_status=TaskState.CLAIMED,
+                          claimed_by='w', claim_token='tok', claimed_at=now,
+                          lease_expires_at=now + 300)
+        update_task_state(run.pipe_root, 't1', new_status=TaskState.RUNNING, started_at=now)
+        update_task_state(run.pipe_root, 't1', new_status=TaskState.FAILED_RETRYABLE,
+                          ended_at=now + 1, failure_class='timeout')
+
+        run.reconcile()
+        self.assertFalse(run.needs_resubmission,
+                         'Should never resubmit — Q-state workers or manual intervention handle retries')
+
+
 class TestPipeRunHomogeneity(unittest.TestCase):
     """Tests for PipeRun homogeneity validation."""
 

arc/scheduler.py

@@ -15,7 +15,7 @@
 
 import arc.parser.parser as parser
 from arc import plotter
-from arc.checks.common import get_i_from_job_name, sum_time_delta
+from arc.checks.common import 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,
@@ -630,7 +630,15 @@ def schedule_jobs(self):
                                     # Accumulate for deferred pipe batching of conf_sp.
                                     self._pending_pipe_conf_sp.setdefault(label, set()).add(i)
                                 if troubleshooting_conformer:
-                                    break
+                                    # Only break if other conformer jobs are still in flight.
+                                    # When the last conformer exhausts troubleshooting without
+                                    # converging, we must fall through to the "all done" check
+                                    # below so it can call determine_most_likely_ts_conformer
+                                    # on the conformers that already succeeded — otherwise ARC
+                                    # mistakenly concludes no TS guess converged.
+                                    if any(is_conformer_job(j)
+                                           for j in self.running_jobs.get(label, [])):
+                                        break
                             # Just terminated a conformer job.
                             # Are there additional conformer jobs currently running for this species?
                             # Note: end_job already removed the current job from running_jobs,
@@ -3791,7 +3799,7 @@ def save_restart_dict(self):
                     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 all(x not in job_name for x in ['conf_opt', 'conf_sp', 'tsg'])] \
+                         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()

arc/scheduler_pipe_test.py

@@ -1011,8 +1011,8 @@ def test_scan_not_mixed_with_other_families(self):
             run.stage()
 
 
-class TestResubmissionLifecycle(unittest.TestCase):
-    """Tests for #1: resubmission sets SUBMITTED status and clears flag."""
+class TestNoResubmissionLifecycle(unittest.TestCase):
+    """Pipe runs must never resubmit — Q-state workers handle retried tasks."""
 
     def setUp(self):
         self.tmpdir = tempfile.mkdtemp(prefix='pipe_resub_test_')
@@ -1021,11 +1021,11 @@ def setUp(self):
     def tearDown(self):
         shutil.rmtree(self.tmpdir, ignore_errors=True)
 
-    def test_resubmission_sets_submitted_status(self):
-        """After successful resubmission, pipe status should be SUBMITTED."""
+    def test_no_resubmission_even_with_retried_tasks(self):
+        """Even when all workers are done and retried tasks remain,
+        poll_pipes must not resubmit a new scheduler job."""
         tasks = [_make_task_spec(f'task_{i}') for i in range(3)]
         pipe = self.sched.pipe_coordinator.submit_pipe_run('resub_test', tasks)
-        # Simulate retried tasks (attempt_index > 0) so reconcile flags resubmission
         for task_id in ['task_0', 'task_1', 'task_2']:
             now = time.time()
             update_task_state(pipe.pipe_root, task_id, new_status=TaskState.CLAIMED,
@@ -1038,22 +1038,10 @@ def test_resubmission_sets_submitted_status(self):
                               claimed_at=None, lease_expires_at=None,
                               started_at=None, ended_at=None, failure_class=None)
         pipe.status = PipeRunState.RECONCILING
-        # Mock submit_to_scheduler to succeed
-        with patch.object(pipe, 'submit_to_scheduler', return_value=('submitted', '12345')):
+        with patch.object(pipe, 'submit_to_scheduler', return_value=('submitted', '12345')) as mock_submit:
             self.sched.pipe_coordinator.poll_pipes()
-        self.assertEqual(pipe.status, PipeRunState.SUBMITTED)
-        self.assertEqual(pipe.scheduler_job_id, '12345')
-        self.assertFalse(pipe._needs_resubmission)
-
-    def test_resubmission_clears_flag_on_failure(self):
-        """After failed resubmission, flag should still be cleared to avoid infinite loops."""
-        tasks = [_make_task_spec(f'task_{i}') for i in range(3)]
-        pipe = self.sched.pipe_coordinator.submit_pipe_run('resub_fail', tasks)
-        pipe._needs_resubmission = True
-        pipe.status = PipeRunState.RECONCILING
-        with patch.object(pipe, 'submit_to_scheduler', return_value=('errored', None)):
-            self.sched.pipe_coordinator.poll_pipes()
-        self.assertFalse(pipe._needs_resubmission)
+        mock_submit.assert_not_called()
+        self.assertFalse(pipe.needs_resubmission)
 
 
 class TestShouldUsePipeOwnerType(unittest.TestCase):