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/db_schema.sql

@@ -40,6 +40,7 @@ CREATE TABLE IF NOT EXISTS sientia_data.predictions (
     "timestamp"           timestamptz NOT NULL,
     created_at            timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL,
     "comments"            text NULL,
+    metadata_id           varchar(24) NULL,
     CONSTRAINT unique_model_id_timestamp
         UNIQUE (model_id, "timestamp")
 );

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/gates.py

@@ -78,6 +78,7 @@ input_path_confidence: Mapping[str, int] = {
     'REPEAT': -1,
 }
 
+
 # MLFlow response filter function mappings
 mlflow_response_filter_functions: dict[str, ResponseFilterFunc] = {
     'API_ERROR': api_error_filter,
@@ -89,12 +90,34 @@ mlflow_response_path_confidence: Mapping[str, int] = {
     'REPEAT': -1,
 }
 
+
+def _count_nan_values(data: DataFrame, _config: dict[str, Any]) -> int:
+    """
+    Count rows for the NAN_VALUES filter.
+
+    `NAN_VALUES` only triggers when every value in the DataFrame is NaN, so every row in it
+    matched the condition — same convention as `_count_specific_variables_null_values`:
+    counting rows that satisfied the filter's own trigger condition.
+
+    Return:
+        int: Number of rows in the evaluated DataFrame.
+    """
+    return len(data)
+
+
 # MLFlow content filter function mappings
 mlflow_content_filter_functions: dict[str, ContentFilterFunc] = {
     'NAN_VALUES': nan_values_filter,
     'EMPTY_DATA': filter_empty_data,
 }
 
+# `EMPTY_DATA` reuses `_count_empty_data` — same function (`filter_empty_data`) and same
+# trigger condition (absence of rows) as `input_gate`'s own EMPTY_DATA.
+mlflow_content_filter_counters: dict[str, InputCounterFunc] = {
+    'NAN_VALUES': _count_nan_values,
+    'EMPTY_DATA': _count_empty_data,
+}
+
 mlflow_content_path_confidence: Mapping[str, int] = {
     'STOP': -1,
     'CONTINUE': 18,
@@ -320,7 +343,9 @@ class Gates(SientiaMonitoring):
         return None, 0, '', None
 
     @activity.defn(name='mlflow_response_gate')
-    def mlflow_response_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
+    def mlflow_response_gate(
+        self, input_data: dict[str, Any]
+    ) -> tuple[str | None, int, str, dict[str, Any] | None]:
         """
         Validate MLFlow API response quality and integrity.
 
@@ -334,6 +359,9 @@ class Gates(SientiaMonitoring):
         3. Determines path decisions based on response validation results
         4. Provides confidence scores and detailed validation comments
         5. Handles API errors and response validation failures
+        6. Builds a metadata document describing which filter(s) triggered, with their policy
+           and a counter, whenever one does — this activity only computes it, it never writes it
+           (see `PredictionMetadata.write_prediction_metadata` for the actual write)
 
         Args:
             input_data: Configuration and data for response validation
@@ -345,10 +373,14 @@ class Gates(SientiaMonitoring):
                     - path_priority (list[str]): Priority order for path decisions
 
         Returns:
-            tuple: (path_flag, confidence, comment)
+            tuple: (path_flag, confidence, comment, document)
                 - path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
                 - confidence (int): Confidence score for the decision
                 - comment (str): Detailed explanation of the decision
+                - document (dict | None): Metadata describing the filter(s) that produced
+                    `path_flag`, shaped `{'filters': [{'name', 'policy', 'counter'}, ...]}`;
+                    `None` when nothing was filtered. Not written anywhere by this activity —
+                    the caller decides whether and how to persist it.
 
         Raises:
             Exception: If response validation fails or configuration is invalid
@@ -369,10 +401,11 @@ class Gates(SientiaMonitoring):
         gate_type = input_data['type']
         path_priority = input_data['path_priority']
 
-        filter_output = []
-
         comments = []
 
+        # (filter_name, policy, filter_config) per filter that triggered, in configured order
+        triggered_filters: list[tuple[str, str, dict[str, Any]]] = []
+
         status = payload.status or {}
 
         for fil, config in filters.items():
@@ -381,15 +414,21 @@ class Gates(SientiaMonitoring):
             policy, filter_config = self._read_filter_entry(config)
             try:
                 if mlflow_response_filter_functions[fil](status, filter_config):
-                    filter_output.append(policy)
-                    comments.append(status.get('message', 'Unknown MLFlow API error'))
+                    triggered_filters.append((fil, policy, filter_config))
+                    # The real error lives one level in, under `content` — `request_transform`/
+                    # `request_predict` nest `message`/`traceback` there on failure (mlflow.py),
+                    # reusing the same `content` key the success path uses for the DataFrame
+                    # (review finding, 2026-10-08: reading `status['message']` directly always
+                    # fell back to the generic default, in every real production occurrence).
+                    error_content = status.get('content') or {}
+                    comments.append(error_content.get('message', 'Unknown MLFlow API error'))
                     self.send_notification(
                         metadata=metadata,
                         notification_id=f'{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}',
-                        message=status.get('message', 'Unknown MLFlow API error'),
+                        message=error_content.get('message', 'Unknown MLFlow API error'),
                         block='mlflow_gate',
                         level=NotificationLevel.ERROR,
-                        attachment_content=status.get('traceback'),
+                        attachment_content=error_content.get('traceback'),
                     )
             except Exception as e:
                 trace = traceback.format_exc()
@@ -402,19 +441,47 @@ class Gates(SientiaMonitoring):
                     attachment_content=trace,
                 )
 
+        triggered_policies = [policy for _, policy, _ in triggered_filters]
+
         for path_flag in path_priority:
-            if path_flag in filter_output:
+            if path_flag in triggered_policies:
                 self.info(f'Mlflow response gate result: {path_flag}', metadata)
-                return path_flag, mlflow_response_path_confidence[path_flag], ', '.join(comments)
+
+                matched = [fil for fil, policy, _ in triggered_filters if policy == path_flag]
+                document = {
+                    'filters': [
+                        {
+                            'name': fil,
+                            'policy': path_flag,
+                            # No per-row/per-item unit to count here — `api_error_filter`
+                            # evaluates a single API response as a whole, not tabular data —
+                            # so, unlike `input_gate`'s per-filter counters, this is a fixed
+                            # value: 1 (one API error event). Inlined rather than a counter
+                            # function/dict, since there's only one response filter today and
+                            # this isn't an override of anything (review finding).
+                            'counter': 1,
+                        }
+                        for fil in matched
+                    ],
+                }
+
+                return (
+                    path_flag,
+                    mlflow_response_path_confidence[path_flag],
+                    ', '.join(comments),
+                    document,
+                )
 
         self.info('Nothing was filtered by the mlflow response gate', metadata)
 
         del data
 
-        return None, 0, ''
+        return None, 0, '', None
 
     @activity.defn(name='mlflow_content_gate')
-    def mlflow_content_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
+    def mlflow_content_gate(
+        self, input_data: dict[str, Any]
+    ) -> tuple[str | None, int, str, dict[str, Any] | None]:
         """
         Validate MLFlow prediction content quality and integrity.
 
@@ -428,6 +495,9 @@ class Gates(SientiaMonitoring):
         3. Determines path decisions based on content validation results
         4. Provides confidence scores and detailed validation comments
         5. Handles content validation failures and quality issues
+        6. Builds a metadata document describing which filter(s) triggered, with their policy
+           and a counter, whenever one does — this activity only computes it, it never writes it
+           (see `PredictionMetadata.write_prediction_metadata` for the actual write)
 
         Args:
             input_data: Configuration and data for content validation
@@ -439,10 +509,14 @@ class Gates(SientiaMonitoring):
                     - path_priority (list[str]): Priority order for path decisions
 
         Returns:
-            tuple: (path_flag, confidence, comment)
+            tuple: (path_flag, confidence, comment, document)
                 - path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
                 - confidence (int): Confidence score for the decision
                 - comment (str): Detailed explanation of the decision
+                - document (dict | None): Metadata describing the filter(s) that produced
+                    `path_flag`, shaped `{'filters': [{'name', 'policy', 'counter'}, ...]}`;
+                    `None` when nothing was filtered. Not written anywhere by this activity —
+                    the caller decides whether and how to persist it.
 
         Raises:
             Exception: If content validation fails or configuration is invalid
@@ -458,7 +532,8 @@ class Gates(SientiaMonitoring):
         gate_type = input_data['type']
         path_priority = input_data['path_priority']
 
-        filter_output = []
+        # (filter_name, policy, filter_config) per filter that triggered, in configured order
+        triggered_filters: list[tuple[str, str, dict[str, Any]]] = []
 
         self._debug_dataframe('Input data:', data, metadata)
         self.debug(f'Filters: \n {filters}', metadata)
@@ -469,7 +544,7 @@ class Gates(SientiaMonitoring):
             policy, filter_config = self._read_filter_entry(config)
             try:
                 if mlflow_content_filter_functions[fil](data, filter_config):
-                    filter_output.append(policy)
+                    triggered_filters.append((fil, policy, filter_config))
                     self.send_notification(
                         metadata=metadata,
                         notification_id=f'{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}',
@@ -489,20 +564,46 @@ class Gates(SientiaMonitoring):
                     attachment_content=trace,
                 )
 
+        triggered_policies = [policy for _, policy, _ in triggered_filters]
+
         for path_flag in path_priority:
-            if path_flag in filter_output:
+            if path_flag in triggered_policies:
                 self.info(f'Mlflow content gate result: {path_flag}', metadata)
+
+                matched = [
+                    (fil, filter_config)
+                    for fil, policy, filter_config in triggered_filters
+                    if policy == path_flag
+                ]
+                document = {
+                    'filters': [
+                        {
+                            'name': fil,
+                            'policy': path_flag,
+                            # `.get(..., lambda *_: 0)`, not `[fil]`: same defensive fallback as
+                            # `input_gate` (review finding, 2026-09-30) — a filter added to
+                            # `mlflow_content_filter_functions` without a matching counter here
+                            # degrades to counter=0 instead of a raw KeyError.
+                            'counter': mlflow_content_filter_counters.get(fil, lambda *_: 0)(
+                                data, filter_config
+                            ),
+                        }
+                        for fil, filter_config in matched
+                    ],
+                }
+
                 return (
                     path_flag,
                     mlflow_content_path_confidence[path_flag],
                     'Transformed data not passed the content filter',
+                    document,
                 )
 
         self.info('Nothing was filtered by the mlflow content gate', metadata)
 
         del data
 
-        return None, 0, ''
+        return None, 0, '', None
 
     def get_prediction_store_policy(
         self, prediction_store_policy: str, metadata: dict[str, Any]

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:

laborious/activities/storage.py

@@ -2,6 +2,7 @@ from temporalio import activity, workflow
 
 with workflow.unsafe.imports_passed_through():
     # Extend the Temporal Postgres activities for convenient query -> MinIO export
+    import time
     import traceback
     from datetime import timedelta
     from typing import Any
@@ -9,12 +10,14 @@ with workflow.unsafe.imports_passed_through():
     import pandas as pd
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
     from sientia_do.notifications.models import NotificationLevel
+    from sientia_do.observability import metrics
     from sientia_do.observability.logger import Logger
     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.temporal.activities.postgres_sync import Postgres
     from sientia_do.temporal.constants import now
+    from sqlalchemy import text
 
     from laborious.utils.messages import MISSING_MINIO_REPOSITORY
     from laborious.utils.models.minio_dataframe_payload import MinioDataFramePayload
@@ -26,7 +29,9 @@ _LOAD_QUERY_OFFLOAD_SKIP_KEYS = frozenset({'model_name', 'key_prefix', 'size_thr
 class Storage(Postgres, SientiaMonitoring):
     """
     Extensions for Postgres activities with a helper to export query results
-    directly to MinIO as Parquet and return the object name.
+    directly to MinIO as Parquet and return the object name, and with
+    ``repeat_last_prediction``, a query specific to this repo's REPEAT path
+    (moved here from the shared library — nothing else consumes it).
     """
 
     minio_repository: MinioRepository | None = None
@@ -141,6 +146,110 @@ class Storage(Postgres, SientiaMonitoring):
             }
         )
 
+    def _build_repeat_last_prediction_query(self, schema: str, table_name: str):
+        table_identifier = f'"{schema}".{table_name}'
+        return text(  # noqa: S608
+            f"""
+        INSERT INTO {table_identifier} (
+            model_id,
+            prediction,
+            timestamp,
+            response_time,
+            prediction_status,
+            prediction_confidence,
+            metadata_id,
+            created_at
+        )
+        SELECT
+            model_id,
+            prediction,
+            :last_timestamp,
+            response_time,
+            prediction_status,
+            prediction_confidence,
+            :metadata_id,
+            NOW()
+        FROM {table_identifier}
+        WHERE model_id = :model
+        ORDER BY timestamp DESC
+        LIMIT 1;
+        """  # nosec B608
+        )
+
+    @activity.defn(name='repeat_last_prediction')
+    def repeat_last_prediction(self, query_items: dict[str, Any]):
+        """
+        Insert a new `predictions` row that copies the last one for a model forward in time.
+
+        REPEAT persists a new row (unlike STOP, which persists nothing), reusing the previous
+        row's `prediction`/`response_time`/`prediction_status`/`prediction_confidence` values with
+        a fresh `timestamp`/`created_at`. `metadata_id` references the `prediction_metadata`
+        document for this cycle, if any (see `PredictionMetadata.write_prediction_metadata`);
+        `None` is a valid value — the column is nullable.
+
+        Args (query_items):
+            metadata (dict): Workflow metadata for logging and metrics.
+            schema (str): Postgres schema of the `predictions` table.
+            table_name (str): `predictions` table name.
+            model (str): `model_id` to repeat the last prediction for.
+            last_timestamp: Timestamp to stamp the new row with.
+            metadata_id (str | None): `prediction_metadata` document id, or `None`.
+
+        Raises:
+            Exception: Re-raised after logging/metrics if the insert fails.
+        """
+        metadata = query_items['metadata']
+        schema = query_items['schema']
+        table_name = query_items['table_name']
+        model = query_items['model']
+        last_timestamp = query_items['last_timestamp']
+        metadata_id = query_items.get('metadata_id')
+
+        repeat_query = self._build_repeat_last_prediction_query(schema, table_name)
+        self.info(f'Repeating last prediction for model {model}', metadata)
+        self.debug(
+            f'Query parameters: schema={schema}, model={model}, last_timestamp={last_timestamp}',
+            metadata,
+        )
+        core_labels = {
+            **self.get_core_labels(metadata, 'execute_sql'),
+            'conflict_mode': 'error',
+        }
+
+        with self.session_factory.begin() as session:
+            try:
+                start_time = time.time()
+                session.execute(
+                    repeat_query,
+                    {
+                        'last_timestamp': last_timestamp,
+                        'model': model,
+                        'metadata_id': metadata_id,
+                    },
+                )
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id='ERROR_REPEATING_LAST_PREDICTION',
+                    message=f'Error repeating last prediction: {e}',
+                    block='repeat_last_prediction',
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace,
+                )
+                self.error(trace, metadata)
+                self.emit_metric_sync(
+                    metric_object=metrics.POSTGRES_WRITE_ERROR_COUNT,
+                    tags=core_labels,
+                )
+                raise
+            else:
+                self.observe_lag_sync(start_time, metrics.POSTGRES_WRITE_LAG, core_labels)
+                self.emit_metric_sync(
+                    metric_object=metrics.POSTGRES_WRITE_COUNT,
+                    tags=core_labels,
+                )
+
     @activity.defn(name='cleanup_minio_objects_expired')
     def cleanup_minio_objects_expired(self, input_data: dict[str, Any]) -> dict[str, Any]:
         """

laborious/workflows/sub_workflows/prediction_process.py

@@ -156,7 +156,7 @@ class PredictionProcess:
         )
 
         # Validate MLFlow transform response
-        path_flag, confidence, comment = await workflow.execute_local_activity_method(
+        path_flag, confidence, comment, document = await workflow.execute_local_activity_method(
             Activities.mlflow_response_gate,
             {
                 **metadata,
@@ -171,11 +171,11 @@ class PredictionProcess:
 
         # Handle path decision based on transform validation
         if await self.path_flag_handler(
-            data, path_flag, input_data, confidence, last_timestamp, comment
+            data, path_flag, input_data, confidence, last_timestamp, comment, document
         ):
             return
 
-        path_flag, confidence, comment = await workflow.execute_local_activity_method(
+        path_flag, confidence, comment, document = await workflow.execute_local_activity_method(
             Activities.mlflow_content_gate,
             {
                 **metadata,
@@ -189,7 +189,7 @@ class PredictionProcess:
         )
 
         if await self.path_flag_handler(
-            data, path_flag, input_data, confidence, last_timestamp, comment
+            data, path_flag, input_data, confidence, last_timestamp, comment, document
         ):
             return
 
@@ -206,7 +206,7 @@ class PredictionProcess:
         )
 
         # Validate MLFlow prediction response
-        path_flag, confidence, comment = await workflow.execute_local_activity_method(
+        path_flag, confidence, comment, document = await workflow.execute_local_activity_method(
             Activities.mlflow_response_gate,
             {
                 **metadata,
@@ -221,7 +221,7 @@ class PredictionProcess:
 
         # Handle path decision based on prediction validation
         if await self.path_flag_handler(
-            data, path_flag, input_data, confidence, last_timestamp, comment
+            data, path_flag, input_data, confidence, last_timestamp, comment, document
         ):
             return
 
@@ -262,8 +262,7 @@ class PredictionProcess:
         Args:
             metadata: Workflow metadata for contextual logging.
             document: Metadata produced by the gate that resolved this `path_flag`; `None` when
-                nothing was filtered, or when the gate that produced this `path_flag` doesn't
-                build one yet.
+                nothing was filtered.
 
         Return:
             The generated id, or `None` when there's nothing to write, including when the
@@ -271,9 +270,11 @@ class PredictionProcess:
         """
         if not document:
             # Skip the remote activity call entirely rather than dispatching it just to have
-            # `write_prediction_metadata` decide there's nothing to write — `mlflow_response_gate`
-            # and `mlflow_content_gate` don't build a `document` yet, so this is the common case
-            # today, not a rare edge case (review finding, 2026-10-01).
+            # `write_prediction_metadata` decide there's nothing to write (review finding,
+            # 2026-10-01). All 3 gates now build a `document` whenever they return a non-`None`
+            # `path_flag`, so this branch is a defensive fallback today, not the common case it
+            # was when this was first written — kept in case a future caller passes a `path_flag`
+            # without one.
             return None
 
         try:
@@ -360,10 +361,9 @@ class PredictionProcess:
             return True
         elif path_flag == 'REPEAT':
             metadata_id = await self._write_prediction_metadata(metadata, document)
-            # Repeat last prediction if available. `metadata_id` isn't consumed by
-            # `repeat_last_prediction` yet (its INSERT...SELECT has no `metadata_id` column —
-            # pending a change in `sientia-dataops-library`); passed through already so no further
-            # change is needed here once that lands.
+            # Repeat last prediction, copying the previous row's values forward with a new
+            # timestamp. `repeat_last_prediction` now lives in `Storage` (laborious), not the
+            # shared library — its INSERT...SELECT references `metadata_id` directly.
             await workflow.execute_activity_method(
                 Activities.repeat_last_prediction,
                 {

pyproject.toml

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

tests/laborious/activities/test_gates.py

@@ -6,7 +6,13 @@ from sientia_do.notifications.models import NotificationLevel
 from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.observability.sientia_monitoring import SientiaMonitoring
 
-from laborious.activities.gates import Gates, input_filter_counters, input_filter_functions
+from laborious.activities.gates import (
+    Gates,
+    input_filter_counters,
+    input_filter_functions,
+    mlflow_content_filter_counters,
+    mlflow_content_filter_functions,
+)
 
 
 @fixture(autouse=True)
@@ -314,7 +320,7 @@ def test_mlflow_response_gate_invalid_filter(gates_activity):
     result = gates_activity.mlflow_response_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
 
 
 @patch('laborious.activities.gates.mlflow_response_filter_functions')
@@ -341,7 +347,7 @@ def test_mlflow_response_gate_filter_exception(
     result = gates_activity.mlflow_response_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.send_notification.assert_called_once_with(
         metadata=metadata['metadata'],
         notification_id='MLFLOW_GATE_RESPONSE_FILTER__INVALID_FILTER',
@@ -369,7 +375,7 @@ def test_mlflow_response_gate_no_filters(gates_activity):
     result = gates_activity.mlflow_response_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
@@ -380,19 +386,30 @@ def test_mlflow_response_gate_with_filter(gates_activity):
         'filters': {'API_ERROR': {'POLICY': 'STOP'}},
         'data': _minio_payload(
             {'content': {'message': 'API error occurred', 'traceback': 'error trace'}},
-            status={'success': False, 'message': 'API error occurred'},
+            status={
+                'success': False,
+                'content': {'message': 'API error occurred', 'traceback': 'error trace'},
+            },
         ),
         'type': 'test',
         'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
     }
 
     # Act
-    result = gates_activity.mlflow_response_gate(input_data)
+    path_flag, confidence, comment, document = gates_activity.mlflow_response_gate(input_data)
 
     # Assert
-    assert result == ('STOP', -1, 'API error occurred')
+    assert (path_flag, confidence, comment) == ('STOP', -1, 'API error occurred')
+    assert document == {'filters': [{'name': 'API_ERROR', 'policy': 'STOP', 'counter': 1}]}
     gates_activity.debug.assert_called()
-    gates_activity.send_notification.assert_called()
+    gates_activity.send_notification.assert_called_once_with(
+        metadata=metadata['metadata'],
+        notification_id='TEST_GATE_RESPONSE_FILTER__API_ERROR',
+        message='API error occurred',
+        block='mlflow_gate',
+        level=NotificationLevel.ERROR,
+        attachment_content='error trace',
+    )
 
 
 def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity):
@@ -402,7 +419,10 @@ def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity):
         'filters': {'API_ERROR': {'Policy': 'STOP'}},
         'data': _minio_payload(
             {'content': {'message': 'API error occurred', 'traceback': 'error trace'}},
-            status={'success': False, 'message': 'API error occurred'},
+            status={
+                'success': False,
+                'content': {'message': 'API error occurred', 'traceback': 'error trace'},
+            },
         ),
         'type': 'test',
         'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
@@ -412,7 +432,12 @@ def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity):
     result = gates_activity.mlflow_response_gate(input_data)
 
     # Assert
-    assert result == ('STOP', -1, 'API error occurred')
+    assert result == (
+        'STOP',
+        -1,
+        'API error occurred',
+        {'filters': [{'name': 'API_ERROR', 'policy': 'STOP', 'counter': 1}]},
+    )
 
 
 def test_mlflow_response_gate_with_filter_not_caught(gates_activity):
@@ -432,7 +457,7 @@ def test_mlflow_response_gate_with_filter_not_caught(gates_activity):
     result = gates_activity.mlflow_response_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
@@ -450,7 +475,7 @@ def test_mlflow_content_gate_invalid_filter(gates_activity):
     result = gates_activity.mlflow_content_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
 
 
 @patch('laborious.activities.gates.mlflow_content_filter_functions')
@@ -472,7 +497,7 @@ def test_mlflow_content_gate_filter_exception(mock_mlflow_content_filter_functio
     result = gates_activity.mlflow_content_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
     gates_activity.send_notification.assert_called_once_with(
         metadata=metadata['metadata'],
@@ -498,12 +523,12 @@ def test_mlflow_content_gate_no_filters(gates_activity):
     result = gates_activity.mlflow_content_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
 def test_mlflow_content_gate_with_filter(gates_activity):
-    # Arrange
+    # Arrange: 3 rows, all-NaN after the filter's own normalization — triggers NAN_VALUES.
     input_data = {
         **metadata,
         'filters': {'NAN_VALUES': {'POLICY': 'STOP', 'CONFIG': {}}},
@@ -513,10 +538,15 @@ def test_mlflow_content_gate_with_filter(gates_activity):
     }
 
     # Act
-    result = gates_activity.mlflow_content_gate(input_data)
+    path_flag, confidence, comment, document = gates_activity.mlflow_content_gate(input_data)
 
     # Assert
-    assert result == ('STOP', -1, 'Transformed data not passed the content filter')
+    assert (path_flag, confidence, comment) == (
+        'STOP',
+        -1,
+        'Transformed data not passed the content filter',
+    )
+    assert document == {'filters': [{'name': 'NAN_VALUES', 'policy': 'STOP', 'counter': 3}]}
     gates_activity.debug.assert_called()
     gates_activity.send_notification.assert_called()
 
@@ -535,7 +565,7 @@ def test_mlflow_content_gate_with_filter_not_caught(gates_activity):
     result = gates_activity.mlflow_content_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
@@ -550,10 +580,18 @@ def test_mlflow_content_gate_filter_returns_false(gates_activity):
 
     result = gates_activity.mlflow_content_gate(input_data)
 
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
+def test_mlflow_content_filter_counters_covers_every_mlflow_content_filter_function():
+    # Same parity guard as `test_input_filter_counters_covers_every_input_filter_function`
+    # (review finding, 2026-09-30): a filter added to `mlflow_content_filter_functions` without
+    # a matching counter here degrades to counter=0 instead of crashing, but this test still
+    # catches the gap before it ships.
+    assert set(mlflow_content_filter_counters) == set(mlflow_content_filter_functions)
+
+
 def test_get_prediction_store_policy_invalid_policy(gates_activity):
     # Arrange
     prediction_store_policy = 'INVALID_POLICY'

tests/laborious/activities/test_storage.py

@@ -257,6 +257,86 @@ def test_export_payload_to_postgres(storage):
     assert result == {'success': True}
 
 
+def test_build_repeat_last_prediction_query(storage):
+    query = storage._build_repeat_last_prediction_query('public', 'predictions')
+    query_str = str(query)
+
+    assert 'INSERT INTO "public".predictions' in query_str
+    assert 'metadata_id' in query_str
+    assert ':last_timestamp' in query_str
+    assert ':metadata_id' in query_str
+    assert ':model' in query_str
+
+
+@patch('laborious.activities.storage.text')
+def test_repeat_last_prediction_success(text_mock, storage):
+    storage.emit_metric_sync = MagicMock()
+    storage.observe_lag_sync = MagicMock()
+    query_items = {
+        **metadata,
+        'schema': 'public',
+        'table_name': 'predictions',
+        'model': 1,
+        'last_timestamp': '2022-01-01',
+        'metadata_id': 'abc123',
+    }
+
+    with patch('sqlalchemy.orm.session.Session.execute') as mock_execute:
+        storage.repeat_last_prediction(query_items)
+
+    text_mock.assert_called_once()
+    mock_execute.assert_called_once_with(
+        text_mock.return_value,
+        {'last_timestamp': '2022-01-01', 'model': 1, 'metadata_id': 'abc123'},
+    )
+    storage.observe_lag_sync.assert_called_once()
+    storage.emit_metric_sync.assert_called_once()
+
+
+@patch('laborious.activities.storage.text')
+def test_repeat_last_prediction_without_metadata_id(text_mock, storage):
+    storage.emit_metric_sync = MagicMock()
+    storage.observe_lag_sync = MagicMock()
+    query_items = {
+        **metadata,
+        'schema': 'public',
+        'table_name': 'predictions',
+        'model': 1,
+        'last_timestamp': '2022-01-01',
+    }
+
+    with patch('sqlalchemy.orm.session.Session.execute') as mock_execute:
+        storage.repeat_last_prediction(query_items)
+
+    mock_execute.assert_called_once_with(
+        text_mock.return_value,
+        {'last_timestamp': '2022-01-01', 'model': 1, 'metadata_id': None},
+    )
+
+
+@patch('laborious.activities.storage.text')
+def test_repeat_last_prediction_error(text_mock, storage):
+    storage.emit_metric_sync = MagicMock()
+    storage.send_notification = MagicMock()
+    query_items = {
+        **metadata,
+        'schema': 'public',
+        'table_name': 'predictions',
+        'model': 1,
+        'last_timestamp': '2022-01-01',
+        'metadata_id': 'abc123',
+    }
+
+    with (
+        patch('sqlalchemy.orm.session.Session.execute', side_effect=ValueError('db')),
+        raises(ValueError),
+    ):
+        storage.repeat_last_prediction(query_items)
+
+    storage.send_notification.assert_called_once()
+    storage.emit_metric_sync.assert_called()
+
+
 def test_cleanup_minio_objects_expired_minio_not_initialized(storage):
     storage.minio_repository = None
 

tests/laborious/workflows/subworkflows/test_prediction_process.py

@@ -59,11 +59,11 @@ async def test_run(workflow_mock, prediction_process):
     workflow_mock.execute_local_activity_method.side_effect = [
         ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
-        ('continue', 0.95, 'Error'),
+        ('continue', 0.95, 'Error', None),
         # mlflow_content_gate (transform)
-        ('continue', 0.95, 'Transformed data not passed the content filter'),
+        ('continue', 0.95, 'Transformed data not passed the content filter', None),
         # mlflow_response_gate (predict)
-        ('continue', 0.95, 'Error'),
+        ('continue', 0.95, 'Error', None),
     ]
 
     # Act
@@ -276,7 +276,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
         ('repeat', 0.95, 'Input data with bad quality', None),  # input_gate
-        ('continue', 0.95, 'Error'),  # mlflow_response_gate (transform)
+        ('continue', 0.95, 'Error', None),  # mlflow_response_gate (transform)
     ]
 
     # Act
@@ -367,9 +367,9 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process
     workflow_mock.execute_local_activity_method.side_effect = [
         ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
-        ('continue', 0.95, 'Error'),
+        ('continue', 0.95, 'Error', None),
         # mlflow_content_gate (transform)
-        ('continue', 0.95, 'Transformed data not passed the content filter'),
+        ('continue', 0.95, 'Transformed data not passed the content filter', None),
     ]
 
     # Act
@@ -478,10 +478,10 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
     workflow_mock.execute_local_activity_method.side_effect = [
         ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
-        ('continue', 0.95, 'Error'),
+        ('continue', 0.95, 'Error', None),
         # mlflow_content_gate (transform)
-        ('continue', 0.95, 'Transformed data not passed the content filter'),
-        ('continue', 0.95, 'Error'),  # mlflow_response_gate (predict)
+        ('continue', 0.95, 'Transformed data not passed the content filter', None),
+        ('continue', 0.95, 'Error', None),  # mlflow_response_gate (predict)
     ]
 
     # Act
@@ -695,10 +695,10 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
 async def test_path_flag_handler_repeat_no_document_skips_write_activity_call(
     workflow_mock, prediction_process
 ):
-    # Arrange: no document (e.g. REPEAT triggered by a gate that doesn't build one yet — today,
-    # `mlflow_response_gate`/`mlflow_content_gate`) must not even dispatch the remote
-    # write_prediction_metadata activity — that would be a wasted Temporal round-trip on every
-    # REPEAT caused by those two gates, not a rare edge case (review finding, 2026-10-01).
+    # Arrange: no document must not even dispatch the remote write_prediction_metadata
+    # activity — that would be a wasted Temporal round-trip (review finding, 2026-10-01).
+    # Defensive case: all 3 gates now always build a document alongside a non-`None`
+    # `path_flag`, so this guards a caller that doesn't, not today's common case.
     data = {'test': 'data'}
 
     # Act
@@ -1075,9 +1075,9 @@ async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process):
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
         ('continue', 0.95, 'ok', None),  # input_gate
-        ('continue', 0.95, ''),
-        ('continue', 0.95, ''),
-        ('continue', 0.95, ''),
+        ('continue', 0.95, '', None),  # mlflow_response_gate (transform)
+        ('continue', 0.95, '', None),  # mlflow_content_gate (transform)
+        ('continue', 0.95, '', None),  # mlflow_response_gate (predict)
     ]
 
     await prediction_process.run(input_data)