SIENTIAPDE-1646

Harden retrain merge and OPC write against silent failures

- pandas merge now uses validate='one_to_one' so duplicated timestamps
  from wrapper.retrain fail loudly instead of silently multiplying rows
- OPC write_data raises OpcClientNotInitializedError instead of a bare
  assert when client is torn down between validation and write, so
  callers get a reported error instead of an AssertionError
- add regression test covering the torn-down client path

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

laborious/activities/mlflow.py

@@ -529,21 +529,26 @@ class MLFlow(SientiaMonitoring):
             )
 
             # Keep heavy model fitting outside MLflow run timing.
-            prediction_data = wrapper.retrain(data)
-            prediction_data.rename(columns={target: 'prediction'}, inplace=True)
+            prediction_data = wrapper.retrain(data).rename(columns={target: 'prediction'})
 
-            # Merge prediction data with retrain data
+            # Merge prediction data with retrain data. Both sides are indexed by the pivoted
+            # timestamp, so one row per timestamp on each side; `validate` fails loudly instead
+            # of silently multiplying rows if the wrapper ever returns duplicated timestamps.
             evaluation_data = pd.merge(
-                data, prediction_data, left_index=True, right_index=True, how='left'
+                data,
+                prediction_data,
+                left_index=True,
+                right_index=True,
+                how='left',
+                validate='one_to_one',
             )
 
-            # Rename target column to "target"
-            evaluation_data.rename(columns={target: 'target'}, inplace=True)
-
-            # Reset index and put as column "timestamp"
+            # Rename target column to "target", then move the index into a "timestamp" column
+            evaluation_data = evaluation_data.rename(columns={target: 'target'})
             evaluation_data['timestamp'] = evaluation_data.index
-            evaluation_data.reset_index(drop=True, inplace=True)
-            evaluation_data.sort_values(by='timestamp', inplace=True, ascending=True)
+            evaluation_data = evaluation_data.reset_index(drop=True).sort_values(
+                by='timestamp', ascending=True
+            )
 
             run_name = f'{model_name}-retrain-{datetime.now().strftime("%Y%m%d%H%M%S")}'
 

laborious/utils/repository/opc_repository.py

@@ -843,7 +843,10 @@ class OpcRepository(SientiaMonitoring):
         session_id = _opc_authentication_token_str(self.client)
 
         try:
-            assert self.client is not None
+            if self.client is None:
+                raise OpcClientNotInitializedError(
+                    f'OPC client is not initialized for server {self.id}; cannot write {node}'
+                )
             node_obj = self.client.get_node(node)
         except Exception as e:
             if is_reconnectable_opcua_bad(e):