scheduler: skip composite sub-job spawn when geometry is missing

get_xyz(generate=False) can return None on a corrupted mid-opt restart;
warn and skip instead of submitting sp sub-jobs with no geometry.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: alongd

arc/scheduler.py

@@ -1638,10 +1638,21 @@ def _spawn_composite_pending(self, label: str) -> None:
         Pending sub_labels that share a Level with an existing pending sub_label
         are de-duplicated here: only one sp job runs, and its eventual output
         path maps to every matching sub_label at completion time.
+
+        Species whose geometry cannot be retrieved (``get_xyz(generate=False)``
+        returns ``None``, e.g. on a corrupted mid-opt restart) are skipped with
+        a warning rather than submitting sp jobs with no geometry.
         """
         pending = self._sp_composite_pending.get(label, {})
         if not pending:
             return
+        xyz = self.species_dict[label].get_xyz(generate=False)
+        if xyz is None:
+            logger.warning(format_log_event(
+                label, "no geometry available — skipping sub-job spawn",
+                {"pending_sub_labels": sorted(pending.keys())},
+            ))
+            return
         unique_levels: list[tuple[Level, list[str]]] = []
         for sub_label, lvl in pending.items():
             for existing_lvl, sub_labels in unique_levels:
@@ -1660,7 +1671,7 @@ def _spawn_composite_pending(self, label: str) -> None:
                                          "sub-job queued",
                                          {"sub_labels": sub_labels, "level": lvl.simple()}))
             self.run_job(label=label,
-                         xyz=self.species_dict[label].get_xyz(generate=False),
+                         xyz=xyz,
                          level_of_theory=lvl,
                          job_type='sp')
 

arc/scheduler_test.py

@@ -1683,6 +1683,73 @@ def test_kick_start_skips_species_with_no_prior_progress(self):
         # No prior progress → kick-start must not spawn anything.
         self.assertEqual(spawned, [])
 
+    def test_spawn_composite_pending_skips_species_without_geometry(self):
+        """A species whose get_xyz(generate=False) returns None must not spawn sub-jobs."""
+        recipe = {"base": {"method": "hf", "basis": "cc-pVTZ"}, "corrections": []}
+        protocol = CompositeProtocol.from_user_input(recipe)
+        spc = ARCSpecies(label='H2', smiles='[H][H]')
+        sched = self._make_scheduler([spc], sp_composite=protocol)
+        self.assertIsNone(spc.get_xyz(generate=False))
+        self.assertTrue(sched._sp_composite_pending['H2'])
+        calls = []
+        sched.run_job = lambda *args, **kwargs: calls.append(kwargs)
+        with self.assertLogs(logger='arc', level=logging.WARNING) as cm:
+            sched._spawn_composite_pending('H2')
+        self.assertEqual(calls, [])
+        joined = '\n'.join(cm.output)
+        self.assertIn('H2', joined)
+        self.assertIn('no geometry', joined)
+
+    def test_restart_kick_start_skips_species_without_geometry(self):
+        """Restart kick-start must not submit sp sub-jobs for a species whose
+        geometry cannot be retrieved (e.g., a corrupted mid-opt restart)."""
+        tmp = os.path.join(self.project_directory, "fx_kickstart_noxyz")
+        os.makedirs(tmp, exist_ok=True)
+        recipe = {
+            "base": {"method": "hf", "basis": "cc-pVTZ"},
+            "corrections": [
+                {"label": "delta_T", "type": "delta",
+                 "high": {"method": "ccsdt",   "basis": "cc-pVDZ"},
+                 "low":  {"method": "ccsd(t)", "basis": "cc-pVDZ"}},
+            ],
+        }
+        protocol = CompositeProtocol.from_user_input(recipe)
+        spc = ARCSpecies(label='H2', smiles='[H][H]')
+        spc.final_xyz = {'symbols': ('H', 'H'), 'coords': ((0, 0, 0), (0, 0, 0.74)),
+                         'isotopes': (1, 1)}
+        sched1 = self._make_scheduler([spc], sp_composite=protocol)
+        paths = self._seed_protocol_fixtures(tmp, protocol,
+                                             {"base": -1.10, "delta_T__high": -1.15,
+                                              "delta_T__low": -1.12})
+        sched1.post_sp_actions('H2', paths["base"], protocol.base.level)
+        output_snapshot = sched1.output
+        # Fresh scheduler from the snapshot, but the species carries no 3D info.
+        spc2 = ARCSpecies(label='H2', smiles='[H][H]')
+        self.assertIsNone(spc2.get_xyz(generate=False))
+        spawned = []
+        with patch.object(Scheduler, "run_job", lambda self, *a, **kw: spawned.append(kw)), \
+                self.assertLogs(logger='arc', level=logging.WARNING) as cm:
+            sched2 = Scheduler(
+                project='sp_composite_orch',
+                ess_settings=self.ess_settings,
+                species_list=[spc2],
+                project_directory=self.project_directory,
+                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']),
+                conformer_opt_level=Level(repr=default_levels_of_theory['conformer']),
+                scan_level=Level(repr=default_levels_of_theory['scan']),
+                ts_guess_level=Level(repr=default_levels_of_theory['ts_guesses']),
+                orbitals_level=default_levels_of_theory['orbitals'],
+                sp_composite=protocol,
+                output=output_snapshot,
+                testing=True,
+            )
+        self.assertEqual(spawned, [])
+        self.assertEqual(set(sched2._sp_composite_pending['H2'].keys()),
+                         {"delta_T__high", "delta_T__low"})
+        self.assertIn('no geometry', '\n'.join(cm.output))
+
     # --- Phase 3.5: corruption recovery ------------------------------------ #
 
     def test_corrupted_recorded_output_is_invalidated_and_requeued(self):