Merge pull request #61 from Aignosi/rc/SIENTIAPDE-2167

Rc/sientiapde 2167 - Prediction Metadata — Passar metadata_id até o Postgres

e2e/conftest.py

@@ -48,8 +48,22 @@ def postgres_container():
 
 @pytest_asyncio.fixture(scope='session')
 def minio_container():
-    """MinIO testcontainer used by E2E offload and payload retrieval paths."""
-    minio = MinioContainer()
+    """
+    MinIO testcontainer used by E2E offload and payload retrieval paths.
+
+    `MinioContainer`'s default image (`minio/minio:RELEASE.2022-12-02T19-19-22Z`) no longer
+    pulls — Docker Hub removed the `minio/minio` repository entirely after MinIO's licensing
+    change. Production (`sientia-helm-chart`) already moved to Bitnami's post-Broadcom-migration
+    "legacy" namespace; mirroring that image here. Two more adjustments on top of just the image:
+    the image runs as non-root by default and can't write to the container's anonymous volume
+    without `user='root'`; and Bitnami deprecated `MINIO_ACCESS_KEY`/`MINIO_SECRET_KEY` (what
+    `MinioContainer` sets) in favor of `MINIO_ROOT_USER`/`MINIO_ROOT_PASSWORD` — reusing the same
+    values so `get_client()` still works unmodified.
+    """
+    minio = MinioContainer(image='docker.io/bitnamilegacy/minio:2025.7.23-debian-12-r3')
+    minio.with_env('MINIO_ROOT_USER', minio.access_key)
+    minio.with_env('MINIO_ROOT_PASSWORD', minio.secret_key)
+    minio.with_kwargs(user='root')
     minio.start()
     yield minio
     minio.stop()
@@ -368,6 +382,7 @@ def _worker_activity_list(test_activities: Activities):
         test_activities.mlflow_content_gate,
         test_activities.request_predict,
         test_activities.repeat_last_prediction,
+        test_activities.write_prediction_metadata,
         test_activities.format_prediction,
         test_activities.format_transformed_data,
         test_activities.format_default_prediction,
@@ -624,5 +639,3 @@ async def temporal_worker_real_opc(temporal_test_env, test_activities_real_opc):
             activity_executor=activity_executor,
         ) as worker:
             yield worker
-
-

e2e/test_drift.py

@@ -142,7 +142,7 @@ def _recent_minute_timestamps(count: int, offset_minutes: int = 6) -> list[str]:
 def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFrame) -> None:
     """
     Wire ``mlflow_repository_stub`` so ``get_reference_data`` returns
-    ``reference_rows`` by writing them to ``dst_path/retrain_input.csv``.
+    ``reference_rows`` by writing them to ``dst_path/evaluation_data.csv``.
 
     Args:
      - mlflow_repository_stub: External MLflow repository fixture.
@@ -158,7 +158,8 @@ def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFram
         run_id='fake-reference-run'
     )
     file_info = MagicMock()
-    file_info.path = 'retrain_input.csv'
+    # Matches `MLFlow._REFERENCE_ARTIFACT_CANDIDATES` (SIENTIAPDE-1646).
+    file_info.path = 'evaluation_data.csv'
     mlflow_repository_stub._client.list_artifacts.return_value = [file_info]
     mlflow_repository_stub.download_artifacts.side_effect = _download
 

e2e/test_predictions_batch_prediction_process.py

@@ -8,6 +8,7 @@ from unittest.mock import MagicMock, patch
 import numpy as np
 import pandas as pd
 import pytest
+from pymongo import MongoClient
 from sqlalchemy import text
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
@@ -86,6 +87,76 @@ async def test_scenario_2_1_1_input_gate_triggers_continue(
     mlflow_repository_stub.stub_wrapper.predict.assert_not_called()
 
 
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_e2e_metadata_id_reaches_postgres_when_gate_triggers(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mongo_container,
+    mlflow_repository_stub,
+):
+    """
+    input_gate CONTINUE: the gate's document is written to Mongo, and the resulting
+    metadata_id lands on the predictions row — independently verified on both sides.
+    """
+    client = temporal_test_env.client
+    model_id = 261
+    # sensor_1 NULL, sensor_2 present — matches `input_filters.SPECIFIC_VARIABLES_NULL_VALUES`
+    # (`variables: ["sensor_1"]`, POLICY CONTINUE) in prediction_process_base.json.
+    insert_sample_data(postgres_engine, model_id, ['NULL', 78.2])
+    input_data = get_base_input_data(model_id)
+    await start_and_await_workflow(
+        client, PredictionsBatch.run, input_data, make_workflow_id('test-metadata-id-continue')
+    )
+    assert_continue(postgres_engine, model_id)
+
+    with postgres_engine.connect() as conn:
+        metadata_id = conn.execute(
+            text(f'SELECT metadata_id FROM sientia_data.predictions WHERE model_id = {model_id}')
+        ).scalar()
+    assert metadata_id is not None, 'Expected a metadata_id on the predictions row'
+
+    # Independent readback: a fresh client against the same testcontainer, not the
+    # `mongo_repository`/`PredictionMetadata` instances the workflow itself wrote through.
+    mongo_port = mongo_container.get_exposed_port(27017)
+    verify_client = MongoClient(f'mongodb://localhost:{mongo_port}')
+    try:
+        document = verify_client['test_db']['prediction_metadata'].find_one({'_id': metadata_id})
+    finally:
+        verify_client.close()
+    assert document is not None, f'Expected a prediction_metadata document for id {metadata_id}'
+    assert document['filters'] == [
+        {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 1}
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_e2e_metadata_id_is_null_when_nothing_triggers(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker: Worker,
+    test_activities: Activities,
+    postgres_engine,
+):
+    """Clean prediction, no gate triggers: predictions.metadata_id stays NULL."""
+    client = temporal_test_env.client
+    model_id = 262
+    insert_sample_data(postgres_engine, model_id, [60.0, 78.2])
+    input_data = get_base_input_data(model_id)
+    await start_and_await_workflow(
+        client, PredictionsBatch.run, input_data, make_workflow_id('test-metadata-id-none')
+    )
+    assert_prediction(postgres_engine, model_id, prediction=0.5, prediction_confidence=0)
+
+    with postgres_engine.connect() as conn:
+        metadata_id = conn.execute(
+            text(f'SELECT metadata_id FROM sientia_data.predictions WHERE model_id = {model_id}')
+        ).scalar()
+    assert metadata_id is None, f'Expected metadata_id to be NULL, got {metadata_id!r}'
+
+
 @pytest.mark.asyncio
 @pytest.mark.integration
 async def test_scenario_2_1_2_input_gate_triggers_stop(
@@ -204,7 +275,7 @@ async def test_scenario_2_2_1_transform_gate_triggers_continue(
         postgres_engine=postgres_engine,
         model_id=model_id,
         prediction_confidence=Decimal(10),
-        comments='Unknown MLFlow API error',
+        comments='Bad data model',
     )
 
 
@@ -332,7 +403,7 @@ async def test_scenario_2_3_1_predict_gate_triggers_continue(
         postgres_engine=postgres_engine,
         model_id=model_id,
         prediction_confidence=Decimal(10),
-        comments='Unknown MLFlow API error',
+        comments='Bad predict model',
     )
 
 
@@ -450,7 +521,7 @@ async def test_scenario_2_4_1_priority_conflict_resolution(
         postgres_engine=postgres_engine,
         model_id=model_id,
         prediction_confidence=Decimal(10),
-        comments='Unknown MLFlow API error',
+        comments='Bad data model',
     )
 
 

laborious/activities/mlflow.py

@@ -664,7 +664,7 @@ class MLFlow(SientiaMonitoring):
         """
         Pick the first available reference CSV artifact path from the MLflow run.
 
-        Candidates are checked in priority order: ``retrain_input.csv``, then ``train_data.csv``.
+        Candidates are checked in priority order: ``evaluation_data.csv``, then ``test_data.csv``.
         A path matches when it equals the candidate or ends with ``/<candidate>`` for nested layouts.
 
         Args:
@@ -703,9 +703,9 @@ class MLFlow(SientiaMonitoring):
         """
         Download reference training CSV from the MLflow run linked to the production alias.
 
-        Resolves ``retrain_input.csv`` or ``train_data.csv`` via artifact listing before download.
-        ``retrain_input.csv`` is preferred when both exist (most recent retrain snapshot). Used by
-        drift workflows to compare live data against the reference distribution logged with the model.
+        Resolves ``evaluation_data.csv`` or ``test_data.csv`` via artifact listing before download.
+        ``evaluation_data.csv`` is preferred when both exist. Used by drift workflows to compare
+        live data against the reference distribution logged with the model.
         Timestamps are normalized to ``DATETIME_FORMAT`` string columns before returning records.
 
         Args: