TS Edge Case & Pipe QUEUE bug (#862)

# Two changes:
---
## Change 1:
In the Scheduler, only break after conformer troubleshooting if new jobs
(conf_opt or conf_sp) are actually running for that species. This
ensures the scheduler correctly falls through to the "all conformers
done" check if troubleshooting was attempted but failed to launch new
tasks.


Essentially, this was a race to completion bug. If there were, for
example, 3 TS guesses being troubleshooted and if the 2 out the 2
finished troubleshoot, then the last one, if it failed an exhausted all
attempts would cause a bug where ARC would declare there was no TS that
converged

## Change 2:

There is a bug in ARC when Pipe is active. The issue is that when a
batch job is submited, and let's say it has 20 jobs in it - so 20
workers needed, and then only 10 of those were picked up by workers but
the other 10 are in Q mode, ARC misunderstand this and attempts to
resubmit those 10 as it thinks they were not provided workers properly.
And ARC will continue to do this ad infinitum

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):