chore: bump version to 0.9.0

This reverts commit e5cea1ef96.

e2e/conftest.py

@@ -29,6 +29,7 @@ from laborious.workflows.sub_workflows.prediction_process import PredictionProce
 from sientia_do.notifications.handlers import CoreNotificationHandler
 from sientia_do.observability.logger import Logger
 from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
 
 # Single source of truth for the test database schema. Mirrors the production
 # DDL for ``sientia_data`` so any production change can be pasted directly into
@@ -141,6 +142,24 @@ def notification_inserts(notification_handler):
         collection.insert_one = original_insert_one
 
 
+@pytest_asyncio.fixture
+def mongo_repository(mock_logger, notification_handler, metrics_controller, mongo_container):
+    """Real MongoDBRepository for gate metadata (prediction_metadata), same testcontainer as
+    ``notification_handler`` above."""
+    mongo_port = mongo_container.get_exposed_port(27017)
+    repository = MongoDBRepository(
+        connection_string=f'mongodb://localhost:{mongo_port}',
+        database_name='test_db',
+        logger=mock_logger,
+        notification_handler=notification_handler,
+        metrics_controller=metrics_controller,
+    )
+    try:
+        yield repository
+    finally:
+        repository.close()
+
+
 class _FakeModelWrapper:
     """External MLflow wrapper double used by repository stub."""
 
@@ -233,6 +252,7 @@ async def test_activities(
     plugin_store_stub,
     pi_web_api_client_stub,
     opc_repository_stub,
+    mongo_repository,
 ):
     """Activities with real infra and external-system stubs only."""
     minio_client = minio_container.get_client()
@@ -269,6 +289,7 @@ async def test_activities(
         notification_handler=notification_handler,
         metrics_controller=metrics_controller,
         mlflow_repository=mlflow_repository_stub,
+        mongo_repository=mongo_repository,
     )
     activities.pi_web_api_client = pi_web_api_client_stub
     activities.opc_repository = {'1': opc_repository_stub}
@@ -289,6 +310,7 @@ async def test_activities_real_minio(
     plugin_store_stub,
     pi_web_api_client_stub,
     opc_repository_stub,
+    mongo_repository,
 ):
     """Compatibility alias for offload tests."""
     minio_client = minio_container.get_client()
@@ -325,6 +347,7 @@ async def test_activities_real_minio(
         notification_handler=notification_handler,
         metrics_controller=metrics_controller,
         mlflow_repository=mlflow_repository_stub,
+        mongo_repository=mongo_repository,
     )
     activities.pi_web_api_client = pi_web_api_client_stub
     activities.opc_repository = {'1': opc_repository_stub}
@@ -535,6 +558,7 @@ async def test_activities_real_opc(
     plugin_store_stub,
     pi_web_api_client_stub,
     opc_e2e_server: OpcE2ETestServer,
+    mongo_repository,
 ):
     """Activities with real OpcRepository connected to the in-process OPC UA server."""
     minio_client = minio_container.get_client()
@@ -571,6 +595,7 @@ async def test_activities_real_opc(
         notification_handler=notification_handler,
         metrics_controller=metrics_controller,
         mlflow_repository=mlflow_repository_stub,
+        mongo_repository=mongo_repository,
     )
     activities.pi_web_api_client = pi_web_api_client_stub
     await asyncio.to_thread(_connect_activities_to_opc_server, activities, opc_e2e_server.url)

laborious/activities/activities.py

@@ -7,6 +7,8 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.observability.logger import Logger
     from sientia_do.repository.minio_repository_sync import MinioRepository
+    from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
+    from sientia_do.utils.connectors_config import build_mongodb_config
     from sientia_model.model_repository.mlflow_repository import SientiaMLflowRepository
     from sientia_model.model_repository.plugin_store import PluginStore
 
@@ -48,6 +50,7 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
         notification_handler: NotificationHandler,
         metrics_controller: MetricsController | None = None,
         mlflow_repository: SientiaMLflowRepository | None = None,
+        mongo_repository: MongoDBRepository | None = None,
     ):
         """
         Wire Postgres, MinIO, MLflow, OPC, gates, metrics, and PI Web API into a single object.
@@ -67,6 +70,9 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
          - metrics_controller: Optional shared controller; if ``None``, a new one is created.
          - mlflow_repository: Optional ``SientiaMLflowRepository`` for unit/e2e tests; in production
            leave unset so the repository is built from environment via ``build_mlflow_config()``.
+         - mongo_repository: Optional ``MongoDBRepository`` (gate metadata, best-effort) for
+           unit/e2e tests; in production leave unset so the repository is built from environment
+           via ``build_mongodb_config()``, same pattern as ``mlflow_repository`` above.
 
         Raises:
             Exception: If any parent ``__init__`` fails (e.g. invalid config keys).
@@ -89,6 +95,16 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
                 metrics_controller=mc,
             )
 
+        if mongo_repository is None:
+            mongo_cfg = build_mongodb_config()
+            mongo_repository = MongoDBRepository(
+                connection_string=mongo_cfg['connection_string'],
+                database_name=mongo_cfg['database_name'],
+                logger=logger,
+                notification_handler=notification_handler,
+                metrics_controller=mc,
+            )
+
         minio_repository = MinioRepository(
             endpoint=minio_config['endpoint_url'],
             access_key=minio_config['access_key'],
@@ -129,6 +145,7 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
         Gates.__init__(
             self,
             minio_repository=minio_repository,
+            mongo_repository=mongo_repository,
             logger=logger,
             notification_handler=notification_handler,
             metrics_controller=mc,

laborious/activities/gates.py

@@ -5,6 +5,7 @@ with workflow.unsafe.imports_passed_through():
     from collections.abc import Callable, Mapping
     from typing import Any
 
+    from bson import ObjectId
     from pandas import DataFrame
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
@@ -12,6 +13,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.observability.metrics_controller import MetricsController
     from sientia_do.observability.sientia_monitoring import SientiaMonitoring
     from sientia_do.repository.minio_repository_sync import MinioRepository
+    from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
     from sientia_do.utils.formatters import create_sample_dict
 
     from laborious import metrics
@@ -86,11 +88,14 @@ class Gates(SientiaMonitoring):
     """
 
     minio_repository: MinioRepository | None = None
+    mongo_repository: MongoDBRepository | None = None
     _MAX_DEBUG_DATAFRAME_ROWS = 100
+    _PREDICTION_METADATA_COLLECTION = 'prediction_metadata'
 
     def __init__(
         self,
         minio_repository: MinioRepository | None = None,
+        mongo_repository: MongoDBRepository | None = None,
         logger: Logger | None = None,
         notification_handler: NotificationHandler | None = None,
         metrics_controller: MetricsController | None = None,
@@ -101,11 +106,14 @@ class Gates(SientiaMonitoring):
         Args:
             logger: Logger instance for observability and debugging
             notification_handler: Notification handler for alerts and monitoring
+            mongo_repository: MongoDB, where gate metadata (triggered filter, policy, counter) is
+                written best-effort. `None` is a valid state (see `_write_prediction_metadata`).
 
         Raises:
             Exception: If BaseActivity initialization fails
         """
         self.minio_repository = minio_repository
+        self.mongo_repository = mongo_repository
         SientiaMonitoring.__init__(
             self,
             logger=logger,
@@ -123,6 +131,11 @@ class Gates(SientiaMonitoring):
                 self.minio_repository.close()
             finally:
                 self.minio_repository = None
+        if self.mongo_repository is not None:
+            try:
+                self.mongo_repository.close()
+            finally:
+                self.mongo_repository = None
         SientiaMonitoring.shutdown(self)
 
     def __del__(self):
@@ -146,6 +159,56 @@ class Gates(SientiaMonitoring):
             metadata,
         )
 
+    def _write_prediction_metadata(
+        self, document: dict[str, Any] | None, metadata: dict[str, Any]
+    ) -> str | None:
+        """
+        Best-effort write of gate metadata (triggered filter, policy, counter) to Mongo.
+
+        Implements both business rules for this feature in one place:
+        - The prediction write must never be blocked or failed by a metadata write. A failed
+          `insert(...)` is swallowed, never re-raised, logged here with gate-level context (which
+          `mongo_repository.insert(...)` itself can't know) — matching how `storage.py`'s
+          `cleanup_minio_objects_expired` also logs again at the call site even though the
+          repository already logs internally and emits its own Prometheus metric
+          (`MONGODB_WRITE_ERROR_COUNT`); the generic library log is for the metric/trace, this
+          one is for whoever is reading logs for this specific gate.
+        - Nothing is written when there's nothing relevant to report. A falsy `document` (e.g.
+          no filter triggered) skips Mongo entirely, so callers can call this unconditionally
+          and get `None` back for "no metadata this time" instead of deciding upfront whether
+          there's something worth saving.
+
+        `insert(...)` doesn't hand back the inserted document's `_id`, so it is generated here,
+        client-side, before the insert, as a plain string (not a BSON ObjectId) — matching this
+        codebase's convention of string `_id`s (e.g. the `models` collection) and the 24-hex-char
+        size of the `predictions.metadata_id` Postgres column that will reference it.
+
+        Args:
+            document: Metadata payload (filter, policy, counter, etc.); `None`/empty skips the
+                write entirely. No `_id` key expected.
+            metadata: Workflow metadata for contextual logging.
+
+        Return:
+            The generated id, or `None` when there's nothing to write, Mongo isn't configured, or
+            the insert failed.
+        """
+        if not document or self.mongo_repository is None:
+            return None
+
+        generated_id = str(ObjectId())
+
+        try:
+            self.mongo_repository.insert(
+                self._PREDICTION_METADATA_COLLECTION,
+                {**document, '_id': generated_id},
+                metadata,
+            )
+        except Exception as e:
+            self.error(f'Failed to write prediction metadata: {e}', metadata)
+            return None
+
+        return generated_id
+
     @staticmethod
     def _read_filter_entry(config: dict[str, Any]) -> tuple[str, dict[str, Any]]:
         """

laborious/activities/mlflow.py

@@ -529,22 +529,23 @@ 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
-            evaluation_data = pd.merge(
-                data, prediction_data, left_index=True, right_index=True, how='left'
+            # Both frames are indexed by the pivoted timestamp: expose it as an explicit column
+            # and join on it. `validate` fails loudly instead of silently multiplying rows if the
+            # wrapper ever returns duplicated timestamps.
+            evaluation_data = (
+                data.reset_index(names='timestamp')
+                .merge(
+                    prediction_data.reset_index(names='timestamp'),
+                    on='timestamp',
+                    how='left',
+                    validate='one_to_one',
+                )
+                .rename(columns={target: 'target'})
+                .sort_values(by='timestamp', ascending=True, ignore_index=True)
             )
 
-            # Rename target column to "target"
-            evaluation_data.rename(columns={target: 'target'}, inplace=True)
-
-            # Reset index and put as column "timestamp"
-            evaluation_data['timestamp'] = evaluation_data.index
-            evaluation_data.reset_index(drop=True, inplace=True)
-            evaluation_data.sort_values(by='timestamp', inplace=True, ascending=True)
-
             run_name = f'{model_name}-retrain-{datetime.now().strftime("%Y%m%d%H%M%S")}'
 
             with self.mlflow_repository.start_run(

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

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "laborious"
-version = "0.7.2"
+version = "0.9.0"
 description = "Sientia DataOps Laborious - ML Model Orchestration System"
 readme = "README.md"
 requires-python = ">=3.11"

requirements-local.txt

@@ -1,6 +1,8 @@
 temporalio
 psycopg2-binary
-sqlalchemy
+# SQLAlchemy 2.1 makes psycopg (v3) the default driver for postgresql:// URLs; sientia_do
+# builds that URL and ships only psycopg2-binary, so 2.1 fails with "No module named psycopg".
+sqlalchemy<2.1
 asyncua==1.0.6
 redis
 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2

requirements.txt

@@ -1,6 +1,8 @@
 temporalio
 psycopg2-binary
-sqlalchemy
+# SQLAlchemy 2.1 makes psycopg (v3) the default driver for postgresql:// URLs; sientia_do
+# builds that URL and ships only psycopg2-binary, so 2.1 fails with "No module named psycopg".
+sqlalchemy<2.1
 asyncua==1.0.6
 redis
 sientia_do>=1.12.2

tests/laborious/activities/test_activities.py

@@ -47,6 +47,7 @@ def test___init__(
     }
 
     mlflow_repository = MagicMock()
+    mongo_repository = MagicMock()
     plugin_store = MagicMock()
 
     opc_config = {
@@ -73,6 +74,7 @@ def test___init__(
         logger=logger,
         notification_handler=notification_handler,
         mlflow_repository=mlflow_repository,
+        mongo_repository=mongo_repository,
     )
 
     assert isinstance(activities, Activities)
@@ -120,6 +122,7 @@ def test___init__(
     mock_gates_init.assert_called_once_with(
         ANY,
         minio_repository=mock_minio_repository.return_value,
+        mongo_repository=mongo_repository,
         logger=logger,
         notification_handler=notification_handler,
         metrics_controller=mock_metrics_controller.return_value,
@@ -217,6 +220,7 @@ def test_shutdown(
         logger=logger,
         notification_handler=notification_handler,
         mlflow_repository=mlflow_repository,
+        mongo_repository=MagicMock(),
     )
 
     activities.shutdown()
@@ -286,6 +290,7 @@ def test___init___builds_mlflow_repository_when_not_provided(
         pi_web_api_config=pi_web_api_config,
         logger=logger,
         notification_handler=notification_handler,
+        mongo_repository=MagicMock(),
     )
 
     mock_build_mlflow_config.assert_called_once()
@@ -297,3 +302,83 @@ def test___init___builds_mlflow_repository_when_not_provided(
         notification_handler=notification_handler,
         metrics_controller=mock_metrics_controller.return_value,
     )
+
+
+@patch('laborious.activities.activities.MongoDBRepository')
+@patch('laborious.activities.activities.build_mongodb_config')
+@patch('laborious.activities.activities.Storage.__init__')
+@patch('laborious.activities.activities.MLFlow.__init__')
+@patch('laborious.activities.activities.OPC.__init__')
+@patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.ModelMetrics.__init__')
+@patch('laborious.activities.activities.API.__init__')
+@patch('laborious.activities.activities.MinioRepository')
+@patch('laborious.activities.activities.MetricsController')
+def test___init___builds_mongo_repository_when_not_provided(
+    mock_metrics_controller,
+    mock_minio_repository,
+    _mock_api_init,
+    _mock_model_metrics_init,
+    mock_gates_init,
+    _mock_opc_init,
+    _mock_mlflow_init,
+    _mock_storage_init,
+    mock_build_mongodb_config,
+    mock_mongo_repository_cls,
+):
+    postgres_config = {
+        'host': 'localhost',
+        'port': 5432,
+        'user': 'postgres',
+        'password': 'postgres',
+        'dbname': 'postgres',
+        'min_connections': 1,
+        'max_connections': 10,
+    }
+    minio_config = {
+        'endpoint_url': 'localhost:9000',
+        'access_key': 'minio',
+        'secret_key': 'minio123',
+        'default_bucket': 'test',
+        'retention_hours': 24,
+        'secure': False,
+    }
+    opc_config = {'bootstrap_servers': 'localhost:9092', 'polling_time': 1000, 'group_id': 'test'}
+    pi_web_api_config = {'base_url': 'https://pi', 'auth_type': 'bearer', 'auth_token': 'token'}
+    logger = MagicMock()
+    notification_handler = MagicMock()
+    plugin_store = MagicMock()
+    mlflow_repository = MagicMock()
+    mock_build_mongodb_config.return_value = {
+        'connection_string': 'mongodb://root:pass@localhost:27018',
+        'database_name': 'sientia',
+    }
+
+    Activities(
+        postgres_config=postgres_config,
+        plugin_store=plugin_store,
+        minio_config=minio_config,
+        opc_config=opc_config,
+        pi_web_api_config=pi_web_api_config,
+        logger=logger,
+        notification_handler=notification_handler,
+        mlflow_repository=mlflow_repository,
+    )
+
+    mock_build_mongodb_config.assert_called_once()
+    mock_mongo_repository_cls.assert_called_once_with(
+        connection_string='mongodb://root:pass@localhost:27018',
+        database_name='sientia',
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )
+    # The same repository built here is the one handed to Gates, not a second instance.
+    mock_gates_init.assert_called_once_with(
+        ANY,
+        minio_repository=mock_minio_repository.return_value,
+        mongo_repository=mock_mongo_repository_cls.return_value,
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )

tests/laborious/activities/test_api.py

@@ -386,7 +386,8 @@ def test_close(mock_shutdown, api):
     api.close()
 
     api.pi_web_api_client.close.assert_called_once()
-    mock_shutdown.assert_called_once_with(api)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(api)) == 1
 
 
 def test_process_pi_web_api_response_success(api):

tests/laborious/activities/test_gates.py

@@ -35,8 +35,14 @@ def _minio_payload(retrieve_return, status=None):
 
 
 @fixture
-def gates_activity():
+def mongo() -> MagicMock:
+    return MagicMock()
+
+
+@fixture
+def gates_activity(mongo: MagicMock):
     gates = Gates(
+        mongo_repository=mongo,
         logger=MagicMock(),
         notification_handler=MagicMock(),
         metrics_controller=MagicMock(spec=MetricsController),
@@ -61,6 +67,75 @@ metadata = {
 }
 
 
+def test_write_prediction_metadata_no_document_skips_mongo(gates_activity, mongo: MagicMock):
+    # Act
+    result = gates_activity._write_prediction_metadata(None, metadata)
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_empty_document_skips_mongo(gates_activity, mongo: MagicMock):
+    # Act
+    result = gates_activity._write_prediction_metadata({}, metadata)
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_not_configured_returns_none():
+    # Arrange
+    gates = Gates(
+        mongo_repository=None,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    gates.error = MagicMock()
+
+    # Act
+    result = gates._write_prediction_metadata({'filter_triggered': 'EMPTY_DATA'}, metadata)
+
+    # Assert
+    assert result is None
+    gates.error.assert_not_called()
+
+
+def test_write_prediction_metadata_success_generates_id_and_inserts(
+    gates_activity, mongo: MagicMock
+):
+    # Arrange
+    document = {'filter_triggered': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 3}
+
+    # Act
+    result = gates_activity._write_prediction_metadata(document, metadata)
+
+    # Assert: a plain 24-hex-char string id (matches predictions.metadata_id's varchar(24)),
+    # not a BSON ObjectId — `insert(...)` never hands one back, so it's generated client-side.
+    assert result is not None
+    assert len(result) == 24
+    int(result, 16)  # raises ValueError if not hex
+
+    collection, inserted_document, inserted_metadata = mongo.insert.call_args.args
+    assert collection == Gates._PREDICTION_METADATA_COLLECTION
+    assert inserted_document == {**document, '_id': result}
+    assert inserted_metadata == metadata
+
+
+def test_write_prediction_metadata_insert_failure_swallowed(gates_activity, mongo: MagicMock):
+    # Arrange
+    mongo.insert.side_effect = RuntimeError('connection reset')
+
+    # Act
+    result = gates_activity._write_prediction_metadata({'filter_triggered': 'EMPTY_DATA'}, metadata)
+
+    # Assert: the prediction write must never be blocked or failed by this
+    assert result is None
+    gates_activity.error.assert_called_once()
+
+
 def test_input_gate_invalid_filter(gates_activity):
     # Arrange
     input_data = {
@@ -991,7 +1066,8 @@ def test_close_disposes_minio_repository(mock_shutdown):
     gates.close()
     minio.close.assert_called_once()
     assert gates.minio_repository is None
-    mock_shutdown.assert_called_once_with(gates)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(gates)) == 1
 
 
 @patch.object(SientiaMonitoring, 'shutdown')
@@ -1004,4 +1080,5 @@ def test_close_without_minio_repository(mock_shutdown):
         metrics_controller=MagicMock(spec=MetricsController),
     )
     gates.close()
-    mock_shutdown.assert_called_once_with(gates)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(gates)) == 1

tests/laborious/activities/test_mlflow.py

@@ -375,9 +375,9 @@ def test_request_predict_failure(mock_to_datetime, mock_from_dataframe, mlflow):
 @patch('laborious.activities.mlflow.rmtree')
 @patch('laborious.activities.mlflow.to_datetime')
 def test_retrain_model_success_data_success_retrain(
-    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
+    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow, tmp_path
 ):
-    mock_mkdtemp.return_value = 'tmp'
+    mock_mkdtemp.return_value = str(tmp_path)
     mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
 
     mv_alias = MagicMock()
@@ -428,9 +428,9 @@ def test_retrain_model_success_data_success_retrain(
 @patch('laborious.activities.mlflow.rmtree')
 @patch('laborious.activities.mlflow.to_datetime')
 def test_retrain_model_success_with_payload_data(
-    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
+    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow, tmp_path
 ):
-    mock_mkdtemp.return_value = 'tmp'
+    mock_mkdtemp.return_value = str(tmp_path)
     mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
     mv_alias = MagicMock(run_id='src')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias
@@ -473,9 +473,9 @@ def test_retrain_model_success_with_payload_data(
 @patch('laborious.activities.mlflow.rmtree')
 @patch('laborious.activities.mlflow.to_datetime')
 def test_retrain_model_always_uses_retrain_even_with_full_retrain_flag(
-    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
+    mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow, tmp_path
 ):
-    mock_mkdtemp.return_value = 'tmp'
+    mock_mkdtemp.return_value = str(tmp_path)
     mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
     mv_alias = MagicMock(run_id='src')
     mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias
@@ -855,3 +855,22 @@ def test_get_reference_data_exception(mlflow):
     result = mlflow.get_reference_data(input_data)
 
     assert result is None
+
+
+def test_find_downloaded_csv_prefers_the_direct_path(mlflow, tmp_path):
+    direct = tmp_path / 'evaluation_data.csv'
+    direct.write_text('a\n1\n')
+
+    assert mlflow._find_downloaded_csv(str(tmp_path), 'evaluation_data.csv') == direct
+
+
+def test_find_downloaded_csv_searches_nested_directories(mlflow, tmp_path):
+    nested = tmp_path / 'artifacts' / 'data' / 'evaluation_data.csv'
+    nested.parent.mkdir(parents=True)
+    nested.write_text('a\n1\n')
+
+    assert mlflow._find_downloaded_csv(str(tmp_path), 'evaluation_data.csv') == nested
+
+
+def test_find_downloaded_csv_returns_none_when_missing(mlflow, tmp_path):
+    assert mlflow._find_downloaded_csv(str(tmp_path), 'evaluation_data.csv') is None

tests/laborious/activities/test_model_metrics.py

@@ -1,7 +1,7 @@
 from unittest.mock import ANY, MagicMock, patch
 
-from pandas import DataFrame, Timestamp
-from pytest import fixture, raises
+from pandas import DataFrame, Series, Timestamp
+from pytest import fixture, mark, raises
 from sientia_do.notifications.models import NotificationLevel
 from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
@@ -953,3 +953,50 @@ def test_calculate_simple_metrics_unknown_metric_ignored(model_metrics_activity)
 
     assert len(result['metric']) == 1
     assert result['metric'].values[0] == 'rmse'
+
+
+@mark.parametrize(
+    'failing_stage',
+    ['detect_univariate_drift', 'detect_multivariate_drift', 'get_drift_metrics_dataframe'],
+)
+@patch('laborious.activities.model_metrics.DriftAnalysis')
+def test_get_drift_metrics_reraises_insufficient_data_untouched(
+    mock_drift_analysis, failing_stage, model_metrics_activity
+):
+    """Insufficient data is not a stage failure: it propagates without the stage error report."""
+    error = DriftInsufficientDataError('not enough rows', analysis_rows=1)
+    getattr(mock_drift_analysis.return_value, failing_stage).side_effect = error
+    data = DataFrame({'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0]})
+    model_metrics_activity._drift_analyze_stage_error = MagicMock()
+
+    with raises(DriftInsufficientDataError) as exc_info:
+        model_metrics_activity.get_drift_metrics(
+            reference_data=data,
+            target_data=data,
+            target_name='target',
+            reference_columns=data.drop(columns=['target', 'timestamp']).columns,
+            drift_metrics=['ks_test'],
+            chunk_period='min',
+            metadata=metadata['metadata'],
+        )
+
+    assert exc_info.value is error
+    model_metrics_activity._drift_analyze_stage_error.assert_not_called()
+
+
+def test_to_naive_utc_converts_aware_timestamps_to_naive_utc():
+    series = Series(['2024-01-01T03:00:00-03:00', '2024-01-01T09:00:00-03:00'])
+
+    result = ModelMetrics._to_naive_utc(series)
+
+    assert result.dt.tz is None
+    assert result.tolist() == [Timestamp('2024-01-01 06:00:00'), Timestamp('2024-01-01 12:00:00')]
+
+
+def test_to_naive_utc_keeps_naive_timestamps():
+    series = Series(['2024-01-01 06:00:00'])
+
+    result = ModelMetrics._to_naive_utc(series)
+
+    assert result.dt.tz is None
+    assert result.tolist() == [Timestamp('2024-01-01 06:00:00')]

tests/laborious/activities/test_storage.py

@@ -1,6 +1,6 @@
 import datetime
 import os
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import ANY, MagicMock, call, patch
 
 from pytest import fixture, raises
 from sientia_do.notifications.models import NotificationLevel
@@ -134,7 +134,8 @@ def test_close(storage, _patch_monitoring_shutdown):
     storage.close()
 
     assert storage.minio_repository is None
-    _patch_monitoring_shutdown.assert_called_once_with(storage)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert _patch_monitoring_shutdown.call_args_list.count(call(storage)) == 1
 
 
 def test_close_when_minio_repository_already_none(storage, _patch_monitoring_shutdown):
@@ -144,7 +145,8 @@ def test_close_when_minio_repository_already_none(storage, _patch_monitoring_shu
     storage.close()
 
     assert storage.minio_repository is None
-    _patch_monitoring_shutdown.assert_called_once_with(storage)
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert _patch_monitoring_shutdown.call_args_list.count(call(storage)) == 1
 
 
 def test_load_query_with_minio_offload_no_rows(storage):
@@ -323,3 +325,17 @@ def test_cleanup_minio_objects_expired_list_objects_error(mock_now, storage):
         attachment_content=ANY,
     )
     storage.error.assert_called_once()
+
+
+def test_close_without_engine_skips_postgres(storage, _patch_monitoring_shutdown):
+    """A Storage whose Postgres init never ran (no engine) still closes MinIO."""
+    del storage.engine
+    minio = MagicMock()
+    storage.minio_repository = minio
+
+    with patch('laborious.activities.storage.Postgres.close') as mock_postgres_close:
+        storage.close()
+
+    mock_postgres_close.assert_not_called()
+    minio.close.assert_called_once()
+    assert storage.minio_repository is None

tests/laborious/utils/repository/test_opc_repository.py

@@ -444,6 +444,22 @@ def test_write_data_get_node_failed(opc_repository):
     assert error_data['attachment_content'] is not None
 
 
+def test_write_data_without_client_fails_as_get_node_error(opc_repository):
+    """A client torn down between validation and the write is reported, not asserted away."""
+    opc_repository.validate_connection = MagicMock(return_value=(True, {}))
+    opc_repository.client = None
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    assert is_success is False
+    assert error_data['notification_id'] == f'OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}'
+    assert 'OPC client is not initialized' in error_data['message']
+    last_call = opc_repository.emit_metric_sync.call_args_list[-1]
+    assert last_call.args[1]['result'] == 'GetNodeError:OpcClientNotInitializedError'
+
+
 def test_write_data_invalid_data_type(opc_repository, mock_client):
     opc_repository.validate_connection = MagicMock(return_value=(True, {}))
     opc_repository.client = mock_client
@@ -769,3 +785,59 @@ def test_run_reconnect_swallows_an_unexpected_exception(opc_repository):
     opc_repository._run_reconnect('BadSessionIdInvalid', 'tok')
 
     assert opc_repository.error.call_count >= 2
+
+
+def test_create_client_without_watchdog_attribute(opc_repository, mock_client):
+    """Older asyncua clients have no `watchdog_intervall`; it is not invented on them."""
+    opc_repository.cert_path = None
+    mock_client.aio_obj = MagicMock(spec=['name', 'description', 'product_uri', 'uaclient'])
+
+    opc_repository._create_client()
+
+    assert opc_repository.client is mock_client
+    assert not hasattr(mock_client.aio_obj, 'watchdog_intervall')
+
+
+def test_run_reconnect_success_logs_no_error(opc_repository):
+    opc_repository._reconnect_locked = MagicMock(return_value=(True, {}))
+
+    opc_repository._run_reconnect('BadSessionIdInvalid', 'tok')
+
+    opc_repository._reconnect_locked.assert_called_once()
+    opc_repository.error.assert_not_called()
+
+
+@pytest.mark.parametrize('failure', [AttributeError('no write_value'), TypeError('bad args')])
+def test_write_node_value_falls_back_to_set_value(failure):
+    node = MagicMock()
+    node.write_value.side_effect = failure
+
+    OpcRepository._write_node_value(node, 'ua-data', 42.0, 'variant')
+
+    node.write_value.assert_called_once_with('ua-data')
+    node.set_value.assert_called_once_with(42.0, 'variant')
+
+
+def test_write_node_value_uses_set_value_without_write_value():
+    node = MagicMock(spec=['set_value'])
+
+    OpcRepository._write_node_value(node, 'ua-data', 42.0, 'variant')
+
+    node.set_value.assert_called_once_with(42.0, 'variant')
+
+
+def test_write_data_session_not_ready_reports_the_reconnect_it_started(opc_repository):
+    """When the SessionNotReady reconnect starts, the write reports reconnect_in_progress."""
+    opc_repository._session_ready.clear()
+    opc_repository._reconnect_thread_in_progress = MagicMock(side_effect=[False, True])
+    opc_repository._start_reconnect = MagicMock()
+    opc_repository.validate_connection = MagicMock()
+
+    is_success, error_data = opc_repository.write_data(
+        'ns=2;s=TestNode', 42.0, 'float', metadata['metadata']
+    )
+
+    opc_repository._start_reconnect.assert_called_once_with('SessionNotReady', ANY)
+    opc_repository.validate_connection.assert_not_called()
+    assert is_success is False
+    assert error_data['opc_error_kind'] == 'reconnect_in_progress'

tests/laborious/worker/test_worker.py

@@ -248,3 +248,45 @@ async def test_main_worker_gather_error_exits_one(monkeypatch):
             await worker.main()
 
     assert m_logger.return_value.custom_error.called
+
+
+@mark.asyncio
+async def test_main_legacy_runtime_installs_single(monkeypatch):
+    """The `legacy` runtime is served by the `single` store runtime."""
+    monkeypatch.setenv('RUNTIME', 'legacy')
+    fake_plugin = MagicMock()
+    fake_plugin.install_runtime = AsyncMock(side_effect=RuntimeError('stop after install'))
+    with (
+        patch('laborious.worker.worker.start_prometheus_server'),
+        patch('laborious.worker.worker.get_logger'),
+        patch(
+            'laborious.worker.worker.build_mongodb_config',
+            return_value={'connection_string': 'cs', 'database_name': 'db'},
+        ),
+        patch('laborious.worker.worker.NotificationHandler'),
+        patch('laborious.worker.worker.MetricsController'),
+        patch(
+            'laborious.worker.worker.build_plugin_store_config',
+            return_value={
+                'base_url': '',
+                'owner': '',
+                'repo': '',
+                'username': None,
+                'password': None,
+                'branch': None,
+                'cache_ttl_seconds': None,
+                'pypi_index_url': '',
+                'pypi_username': None,
+                'pypi_password': None,
+            },
+        ),
+        patch('laborious.worker.worker.PluginStore', return_value=fake_plugin),
+        patch.object(worker.metrics.APP_UP, 'labels') as labels,
+        patch('laborious.worker.worker.sys.exit', side_effect=SystemExit(1)),
+    ):
+        labels.return_value = MagicMock()
+        with raises(SystemExit):
+            await worker.main()
+
+    fake_plugin.install_runtime.assert_awaited_once()
+    assert fake_plugin.install_runtime.await_args.kwargs['runtime_name'] == 'single'