Fix pipe jobs immediately failing due to stale server_job_ids snapshot (#869)

Pipe job IDs submitted during the scheduler label loop weren't in the
server_job_ids list (fetched earlier in the same iteration), causing
poll_pipes to conclude the scheduler job was gone and mark all tasks
FAILED TERMINAL

Append the job ID to server_job_ids on both initial submission and
resubmission so _is_scheduler_job_alive sees it immediately.

Author: alongd

arc/job/pipe/pipe_coordinator.py

@@ -172,6 +172,7 @@ def submit_pipe_run(self, run_id: str, tasks: List[TaskSpec],
                 pipe.status = PipeRunState.SUBMITTED
                 pipe.submitted_at = time.time()
                 pipe._save_run_metadata()
+                self.sched.server_job_ids.append(job_id)
                 logger.info(f'Pipe run {run_id} submitted as job {job_id}.')
             else:
                 logger.warning(f'Pipe run {run_id}: submission returned status={job_status}. '
@@ -234,6 +235,7 @@ def poll_pipes(self) -> None:
                         pipe.submitted_at = time.time()
                         pipe._needs_resubmission = False
                         pipe._save_run_metadata()
+                        self.sched.server_job_ids.append(job_id)
                         logger.info(f'Pipe run {run_id}: resubmitted as job {job_id}.')
                     else:
                         pipe._needs_resubmission = False

arc/job/pipe/pipe_coordinator_test.py

@@ -67,6 +67,7 @@ def _make_mock_sched(project_directory):
     """Create a mock Scheduler with the attributes PipeCoordinator needs."""
     sched = MagicMock()
     sched.project_directory = project_directory
+    sched.server_job_ids = list()
     spc = ARCSpecies(label='H2O', smiles='O')
     spc.conformers = [None] * 5
     spc.conformer_energies = [None] * 5
@@ -157,6 +158,14 @@ def test_submit_uses_explicit_cluster_software(self):
         pipe = self.coord.submit_pipe_run('run_pbs', tasks, cluster_software='pbs')
         self.assertEqual(pipe.cluster_software, 'pbs')
 
+    def test_submit_adds_job_id_to_server_job_ids(self):
+        """Submitted pipe job ID is added to server_job_ids to prevent stale-snapshot race."""
+        tasks = [_make_spec('t_0')]
+        with patch.object(PipeRun, 'submit_to_scheduler', return_value=('submitted', '12345[]')):
+            pipe = self.coord.submit_pipe_run('run_ids', tasks)
+        self.assertIn('12345[]', self.coord.sched.server_job_ids)
+        self.assertEqual(pipe.scheduler_job_id, '12345[]')
+
 
 class TestRegisterFromDir(unittest.TestCase):
     """Tests for PipeCoordinator.register_pipe_run_from_dir()."""
@@ -222,6 +231,19 @@ def test_poll_resets_failure_count_on_success(self):
         self.coord.poll_pipes()  # succeeds this time
         self.assertNotIn('run_flaky', self.coord._pipe_poll_failures)
 
+    def test_resubmission_adds_job_id_to_server_job_ids(self):
+        """Resubmitted pipe job ID is added to server_job_ids."""
+        pipe = self.coord.submit_pipe_run('run_resub', [_make_spec('t_resub')])
+
+        def fake_reconcile():
+            pipe._needs_resubmission = True
+            return {TaskState.PENDING.value: 1}
+
+        with patch.object(pipe, 'reconcile', side_effect=fake_reconcile), \
+             patch.object(pipe, 'submit_to_scheduler', return_value=('submitted', '77777[]')):
+            self.coord.poll_pipes()
+        self.assertIn('77777[]', self.coord.sched.server_job_ids)
+
 
 class TestIngestPipeResults(unittest.TestCase):
     """Tests for PipeCoordinator.ingest_pipe_results()."""