Updates

Author: calvinp0

arc/main.py

@@ -36,6 +36,7 @@
 from arc.job.ssh import SSHClient
 from arc.output import write_output_yml
 from arc.processor import process_arc_project, resolve_neb_level
+from arc.provenance import DecisionKind, EdgeType
 from arc.reaction import ARCReaction
 from arc.scheduler import Scheduler
 from arc.species.converter import str_to_xyz
@@ -671,6 +672,70 @@ def execute(self) -> dict:
         log_footer(execution_time=self.execution_time)
         return status_dict
 
+    def _add_arkane_provenance_nodes(self):
+        """Add Arkane computation and result nodes to the provenance graph.
+
+        For each converged species with thermo results, creates:
+            convergence_confirmed → calc(statmech_thermo) → data(thermo)
+
+        For each converged reaction with kinetics results, creates:
+            convergence_confirmed → calc(statmech_kinetics) → data(kinetics)
+        """
+        graph = self.scheduler.graph
+        for spc in self.scheduler.species_dict.values():
+            if spc.is_ts or getattr(spc.thermo, 'H298', None) is None:
+                continue
+            spc_nid = graph.find_species_node(spc.label)
+            if spc_nid is None:
+                continue
+            # Insert a CalculationNode for the Arkane thermo computation.
+            calc_nid = graph.add_calculation_node(
+                label=spc.label,
+                job_name='arkane_thermo',
+                job_type='statmech_thermo',
+                job_adapter=self.thermo_adapter,
+                status='done',
+            )
+            graph.add_edge(spc_nid, calc_nid, EdgeType.belongs_to)
+            # Link from convergence gate if it exists.
+            conv_nodes = graph.query(decision_kind=DecisionKind.convergence_confirmed, label=spc.label)
+            for conv_node in conv_nodes:
+                graph.add_edge(conv_node.node_id, calc_nid, EdgeType.triggered_by)
+            thermo_nid = graph.add_data_node(
+                label=spc.label,
+                data_kind='thermo',
+                value=f'H298={spc.thermo.H298:.1f} kJ/mol, S298={spc.thermo.S298:.1f} J/mol/K',
+            )
+            graph.add_edge(calc_nid, thermo_nid, EdgeType.output_of)
+        for rxn in self.scheduler.rxn_list:
+            if rxn.kinetics is None:
+                continue
+            ts_nid = graph.find_species_node(rxn.ts_label)
+            if ts_nid is None:
+                continue
+            # Insert a CalculationNode for the Arkane kinetics computation.
+            calc_nid = graph.add_calculation_node(
+                label=rxn.ts_label,
+                job_name='arkane_kinetics',
+                job_type='statmech_kinetics',
+                job_adapter=self.kinetics_adapter,
+                status='done',
+            )
+            graph.add_edge(ts_nid, calc_nid, EdgeType.belongs_to)
+            # Link from TS convergence gate if it exists.
+            conv_nodes = graph.query(decision_kind=DecisionKind.convergence_confirmed, label=rxn.ts_label)
+            for conv_node in conv_nodes:
+                graph.add_edge(conv_node.node_id, calc_nid, EdgeType.triggered_by)
+            ea = rxn.kinetics.get('Ea')
+            ea_str = f', Ea={ea[0]:.1f} {ea[1]}' if ea else ''
+            kinetics_nid = graph.add_data_node(
+                label=rxn.ts_label,
+                data_kind='kinetics',
+                value=f'{rxn.label}{ea_str}',
+            )
+            graph.add_edge(calc_nid, kinetics_nid, EdgeType.output_of)
+        graph.save(self.scheduler.graph_path)
+
     def save_project_info_file(self):
         """
         Save a project info file.

arc/plotter.py

@@ -120,7 +120,10 @@ def render_provenance_graph(prov_graph, run_label: str = 'ARC run') -> 'graphviz
         'ts_validation_freq': 'lightyellow',
         'ts_validation_nmd': 'lightyellow',
         'ts_validation_irc': 'lightyellow',
+        'ts_validation_e0': 'lightyellow',
+        'ts_validation_e_elect': 'lightyellow',
         'ts_switch': 'mistyrose',
+        'convergence_confirmed': 'palegreen',
     }
 
     # Edge styling lookup

arc/provenance/nodes.py

@@ -55,9 +55,12 @@ class DecisionKind(str, Enum):
     ts_validation_freq = 'ts_validation_freq'
     ts_validation_nmd = 'ts_validation_nmd'
     ts_validation_irc = 'ts_validation_irc'
+    ts_validation_e0 = 'ts_validation_e0'
+    ts_validation_e_elect = 'ts_validation_e_elect'
     ts_switch = 'ts_switch'
     job_troubleshooting = 'job_troubleshooting'
     ts_method_spawning = 'ts_method_spawning'
+    convergence_confirmed = 'convergence_confirmed'
 
 
 class EdgeType(str, Enum):

arc/provenance/provenance_test.py

@@ -33,7 +33,9 @@ def test_data_kind_values(self):
     def test_decision_kind_values(self):
         expected = {'conformer_selection', 'ts_guess_clustering', 'ts_guess_selection',
                     'ts_guess_selection_failed', 'ts_validation_freq', 'ts_validation_nmd',
-                    'ts_validation_irc', 'ts_switch', 'job_troubleshooting', 'ts_method_spawning'}
+                    'ts_validation_irc', 'ts_validation_e0', 'ts_validation_e_elect',
+                    'ts_switch', 'job_troubleshooting', 'ts_method_spawning',
+                    'convergence_confirmed'}
         actual = {dk.value for dk in DecisionKind}
         self.assertEqual(expected, actual)
 

arc/scheduler.py

@@ -2888,12 +2888,12 @@ def check_freq_job(self,
                                     f'Status is:\n{self.species_dict[label].ts_checks}\n'
                                     f'Searching for a better TS conformer...')
                         # ── Graph: record NMD validation failure ──
-                        self.graph.add_decision_node(
+                        nmd_fail_nid = self.graph.add_decision_node(
                             label=label,
                             decision_kind=DecisionKind.ts_validation_nmd,
                             outcome='Failed: normal mode displacement check',
                         )
-                        self.switch_ts(label)
+                        self.switch_ts(label, triggered_by_nid=nmd_fail_nid)
                         switch_ts = True
                 if wrong_freq_message in self.output[label]['warnings']:
                     self.output[label]['warnings'] = ''.join(self.output[label]['warnings'].split(wrong_freq_message))
@@ -2961,14 +2961,14 @@ def check_negative_freq(self,
                                 f'Status is:\n{self.species_dict[label].ts_checks}\n'
                                 f'Searching for a better TS conformer...')
                 # ── Graph: record TS freq validation failure ──
-                self.graph.add_decision_node(
+                freq_fail_nid = self.graph.add_decision_node(
                     label=label,
                     decision_kind=DecisionKind.ts_validation_freq,
                     criteria={'neg_freqs': [float(f) for f in neg_freqs],
                               'expected': 1},
                     outcome=f'Failed: {len(neg_freqs)} imaginary freqs, switching TS',
                 )
-                self.switch_ts(label=label)
+                self.switch_ts(label=label, triggered_by_nid=freq_fail_nid)
                 return False
             else:
                 logger.info(f'TS {label} has exactly one imaginary frequency: {neg_freqs[0]}')
@@ -3030,7 +3030,13 @@ def check_rxn_e0_by_spc(self, label: str):
                 if rxn.ts_species.ts_checks['E0'] is False:
                     logger.info(f'TS {rxn.ts_species.label} of reaction {rxn.label} did not pass the E0 check.\n'
                                 f'Searching for a better TS conformer...\n')
-                    self.switch_ts(rxn.ts_label)
+                    # ── Graph: record E0 validation failure ──
+                    e0_fail_nid = self.graph.add_decision_node(
+                        label=rxn.ts_label,
+                        decision_kind=DecisionKind.ts_validation_e0,
+                        outcome=f'Failed: TS E0 not above both wells for {rxn.label}',
+                    )
+                    self.switch_ts(rxn.ts_label, triggered_by_nid=e0_fail_nid)
                     if self.species_dict[rxn.ts_label].ts_guesses_exhausted \
                             or self.species_dict[rxn.ts_label].chosen_ts is None:
                         logger.warning(f'Could not find a valid TS conformer for {rxn.ts_label} '
@@ -3039,27 +3045,50 @@ def check_rxn_e0_by_spc(self, label: str):
                         # Restore E0 failure flag — switch_ts resets ts_checks via populate_ts_checks().
                         # check_all_done reads this to avoid overwriting convergence back to True.
                         self.species_dict[rxn.ts_label].ts_checks['E0'] = False
+                elif rxn.ts_species.ts_checks['E0'] is True:
+                    # ── Graph: record E0 validation pass ──
+                    self.graph.add_decision_node(
+                        label=rxn.ts_label,
+                        decision_kind=DecisionKind.ts_validation_e0,
+                        outcome=f'Passed: TS E0 above both wells for {rxn.label}',
+                    )
+                # Also record e_elect check if it ran as a fallback.
+                if rxn.ts_species.ts_checks['e_elect'] is True:
+                    self.graph.add_decision_node(
+                        label=rxn.ts_label,
+                        decision_kind=DecisionKind.ts_validation_e_elect,
+                        outcome=f'Passed: TS e_elect above both wells for {rxn.label}',
+                    )
+                elif rxn.ts_species.ts_checks['e_elect'] is False:
+                    self.graph.add_decision_node(
+                        label=rxn.ts_label,
+                        decision_kind=DecisionKind.ts_validation_e_elect,
+                        outcome=f'Warning: TS e_elect not above both wells for {rxn.label}',
+                    )
 
-    def switch_ts(self, label: str):
+    def switch_ts(self, label: str, triggered_by_nid: Optional[str] = None):
         """
         Try the next optimized TS guess in line if a previous TS guess was found to be wrong.
 
         Args:
             label (str): The TS species label.
+            triggered_by_nid (str, optional): Node ID of the validation decision that triggered this switch.
         """
         logger.info(f'Switching a TS guess for {label}...')
         old_chosen = self.species_dict[label].chosen_ts
         self.determine_most_likely_ts_conformer(label=label)  # Look for a different TS guess.
         new_chosen = self.species_dict[label].chosen_ts
         # ── Graph: record TS switch decision ──
-        self.graph.add_decision_node(
+        switch_nid = self.graph.add_decision_node(
             label=label,
             decision_kind=DecisionKind.ts_switch,
             criteria={'old_chosen': old_chosen, 'new_chosen': new_chosen,
                       'exhausted': self.species_dict[label].ts_guesses_exhausted},
             outcome=f'Switched from TSG #{old_chosen} to #{new_chosen}'
             if new_chosen is not None else 'All TS guesses exhausted',
         )
+        if triggered_by_nid is not None:
+            self.graph.add_edge(triggered_by_nid, switch_nid, EdgeType.triggered_by)
         self.delete_all_species_jobs(label=label)  # Delete other currently running jobs for this TS.
         freq_path = os.path.join(self.project_directory, 'output', 'rxns', label, 'geometry', 'freq.out')
         if os.path.isfile(freq_path):
@@ -3239,17 +3268,25 @@ def check_irc_species(self, label: str):
         if len(self.output[ts_label]['paths']['irc']) == 2:
             irc_species_labels = self.species_dict[ts_label].irc_label.split()
             if all(self.output[irc_label]['paths']['geo'] for irc_label in irc_species_labels):
+                rxn = self.rxn_dict.get(self.species_dict[ts_label].rxn_index, None)
                 check_irc_species_and_rxn(
                     xyz_1=self.output[irc_species_labels[0]]['paths']['geo'],
                     xyz_2=self.output[irc_species_labels[1]]['paths']['geo'],
-                    rxn=self.rxn_dict.get(self.species_dict[ts_label].rxn_index, None),
+                    rxn=rxn,
                 )
-                # ── Graph: record IRC validation decision ──
+                # ── Graph: record IRC validation decision with actual outcome ──
+                irc_result = rxn.ts_species.ts_checks['IRC'] if rxn is not None else None
+                if irc_result is True:
+                    irc_outcome = 'Passed: IRC endpoints match expected R/P'
+                elif irc_result is False:
+                    irc_outcome = 'Failed: IRC endpoints do not match expected R/P'
+                else:
+                    irc_outcome = 'Inconclusive: could not determine IRC match'
                 self.graph.add_decision_node(
                     label=ts_label,
                     decision_kind=DecisionKind.ts_validation_irc,
-                    criteria={'irc_species': irc_species_labels},
-                    outcome='IRC validation completed',
+                    criteria={'irc_species': irc_species_labels, 'result': irc_result},
+                    outcome=irc_outcome,
                 )
 
     def check_scan_job(self,
@@ -3513,6 +3550,12 @@ def check_all_done(self, label: str):
                 all_converged = False
         if label in self.output and all_converged:
             self.output[label]['convergence'] = True
+            # ── Graph: record convergence gate ──
+            self.graph.add_decision_node(
+                label=label,
+                decision_kind=DecisionKind.convergence_confirmed,
+                outcome=f'All required calculations converged',
+            )
             if self.species_dict[label].is_ts:
                 self.species_dict[label].make_ts_report()
                 logger.info(self.species_dict[label].ts_report + '\n')