chore: bump version to 0.9.0

feat(simple_metrics): adaptive state on the pipelines doc (SIENTIAPDE-2191)

README.md

@@ -487,16 +487,18 @@ The **SimpleMetrics** workflow calculates regression metrics (RMSE, MSE, MAE, R
 
 #### Execution Flow
 1. **Data Loading**: Loads prediction vs target data via a JOIN query
-2. **Metrics Calculation**: Calculates configured regression metrics
+2. **Metrics Calculation**: Calculates configured regression metrics; alerts on `simple_metrics_thresholds` breaches (one alert per model, the message names the window) unless the window has fewer rows than `simple_metrics_min_samples`
 3. **Data Export**: Exports metrics to PostgreSQL
+4. **Adaptive State**: when the schedule's `pipelines` doc has `frequency_alert`, escalates `adaptive_state` on a breach and recovers after the window is renewed without one (`min(12, ceil(interval_minutes / frequency_alert))` runs, at least 1; empty, under-sampled or threshold-less windows count as healthy). The orchestrator runs the schedule at `frequency_alert` while escalated. Runs on every schedule (it reads the `pipelines` doc to check for `frequency_alert`), so a MongoDB failure here fails the run; the metrics are already exported by then
 
 #### Architecture Diagram
 ```mermaid
 flowchart LR
-    A[1. load_custom_query] --> B[2. calculate_simple_metrics] --> C[3. export_data_to_postgres]
+    A[1. load_custom_query] --> B[2. calculate_simple_metrics] --> C[3. export_data_to_postgres] --> D[4. update_simple_metrics_adaptive_state]
     
     A -.-> Database[(Database)]
     C -.-> PostgreSQL[(PostgreSQL)]
+    D -.-> MongoDB[(MongoDB pipelines)]
 ```
 
 #### Input Parameters
@@ -510,7 +512,12 @@ flowchart LR
   "data_table_name": "laborious_data",
   "target_table_name": "simple_metrics",
   "interval_minutes": 60,
-  "model_config": { "target": "temperature", "alias": "production" },
+  "model_config": {
+    "target": "temperature",
+    "alias": "production",
+    "simple_metrics_thresholds": { "rmse_max": 1.5 },
+    "simple_metrics_min_samples": 10
+  },
   "metrics": ["rmse", "mse", "mae", "r2"]
 }
 ```
@@ -522,7 +529,7 @@ flowchart LR
 - PostgreSQL database
 - MLFlow server
 - MinIO object storage (for MLFlow artifacts)
-- MongoDB server (for notifications)
+- MongoDB server (notifications, prediction metadata, and read/write on the orchestrator's `pipelines` collection for the simple-metrics adaptive state)
 - OPC server(s) if using OPC export
 - PI Web API server if using PI Web API export
 
@@ -898,7 +905,7 @@ See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, an
 | `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes |
 | `MONGODB_USERNAME` | MongoDB username | `root` | Yes |
 | `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes |
-| `MONGODB_DATABASE_NAME` | MongoDB database name | `sientia` | Yes |
+| `MONGODB_DATABASE_NAME` | MongoDB database name; must be the same database as the orchestrator's `pipelines` | `sientia` | Yes |
 | `MONGODB_TTL_INDEX_HOURS` | MongoDB TTL index hours | `1` | No |
 | `LOG_LEVEL` | Application log level | `INFO` | No |
 | `PROJECT_NAME` | Project name for metrics | `laborious` | No |

e2e/conftest.py

@@ -445,6 +445,7 @@ def _simple_metrics_worker_activity_list(test_activities: Activities):
         test_activities.load_custom_query,
         test_activities.calculate_simple_metrics,
         test_activities.export_data_to_postgres,
+        test_activities.update_simple_metrics_adaptive_state,
     ]
 
 

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/scenarios.md

@@ -413,6 +413,15 @@ four metrics with closed-form expected values.
 #### S.1.2 Subset metrics
 **Summary**: Requesting `metrics=['rmse']` writes only the rmse row.
 
+#### S.1.3 Threshold breach escalates the adaptive state
+**Summary**: With `simple_metrics_thresholds` breached and a `pipelines` doc
+(same `schedule_name`) carrying `frequency_alert`, the workflow sets
+`adaptive_state={escalated: true, healthy_streak: 0}` and stamps `updated_at`.
+
+**Expected Outcome**:
+- `adaptive_state` written on the `pipelines` doc.
+- `updated_at` is a BSON datetime (the orchestrator calls `.replace(tzinfo=UTC)` on it).
+
 ### 6.2 Edge cases
 
 #### S.2.1 Zero-variance target returns r2=0

e2e/test_simple_metrics.py

@@ -15,6 +15,7 @@ from decimal import Decimal
 
 import math
 import pytest
+from pymongo import MongoClient
 from sqlalchemy import text
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
@@ -140,7 +141,8 @@ async def test_simple_metrics_happy_path_persists_all_metrics_and_columns(
     prediction/target pair set and written one row per metric. Every column
     expected by ``sientia_data.simple_metrics`` must be populated (except the
     nullable ``timestamp`` column) and the numerical values must match
-    closed-form expectations.
+    closed-form expectations, rounded to 2 decimals as ``RegressionMetrics``
+    persists them (see ``sientia_model/metrics/regression.py``).
     """
     client = temporal_test_env.client
     model_id = 511
@@ -154,13 +156,13 @@ async def test_simple_metrics_happy_path_persists_all_metrics_and_columns(
     ]
     diffs = [target - prediction for prediction, target in pairs]
     n = len(diffs)
-    expected_rmse = math.sqrt(sum(d * d for d in diffs) / n)
-    expected_mse = sum(d * d for d in diffs) / n
-    expected_mae = sum(abs(d) for d in diffs) / n
+    expected_rmse = round(math.sqrt(sum(d * d for d in diffs) / n), 2)
+    expected_mse = round(sum(d * d for d in diffs) / n, 2)
+    expected_mae = round(sum(abs(d) for d in diffs) / n, 2)
     target_mean = sum(t for _, t in pairs) / n
     ss_res = sum((target - prediction) ** 2 for prediction, target in pairs)
     ss_tot = sum((t - target_mean) ** 2 for _, t in pairs)
-    expected_r2 = 1.0 - (ss_res / ss_tot)
+    expected_r2 = round(1.0 - (ss_res / ss_tot), 2)
 
     _seed_predictions_and_targets(postgres_engine, model_id=model_id, pairs=pairs)
 
@@ -331,3 +333,54 @@ async def test_simple_metrics_no_overlapping_data_short_circuits(
             {'m': str(model_id)},
         ).scalar()
     assert count == 0, 'Empty target data must short-circuit and skip persistence'
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_simple_metrics_breach_escalates_adaptive_state(
+    temporal_test_env: WorkflowEnvironment,
+    temporal_worker_simple_metrics: Worker,
+    test_activities: Activities,
+    postgres_engine,
+    mongo_container,
+):
+    """
+    Scenario S.1.3: a threshold breach on a schedule whose pipelines doc has
+    ``frequency_alert`` escalates ``adaptive_state`` and stamps ``updated_at``
+    as a BSON datetime (the type the orchestrator reads).
+    """
+    model_id = 541
+    schedule_name = 'test-simple-metrics-adaptive'
+    _seed_predictions_and_targets(
+        postgres_engine, model_id=model_id, pairs=[(1.0, 2.0), (2.0, 4.0), (3.0, 5.0)]
+    )
+
+    mongo = MongoClient(f'mongodb://localhost:{mongo_container.get_exposed_port(27017)}')
+    pipelines = mongo['test_db']['pipelines']
+    try:
+        pipelines.delete_many({'schedule_name': schedule_name})
+        pipelines.insert_one(
+            {'schedule_name': schedule_name, 'frequency': '1h', 'frequency_alert': '5m'}
+        )
+
+        input_data = _simple_metrics_input(
+            model_id,
+            schedule_name=schedule_name,
+            model_config={
+                'target': 'sensor_target',
+                'simple_metrics_thresholds': {'rmse_max': 0.01},
+            },
+        )
+        await start_and_await_workflow(
+            temporal_test_env.client,
+            SimpleMetrics.run,
+            input_data,
+            make_workflow_id('test-simple-metrics-adaptive'),
+        )
+
+        doc = pipelines.find_one({'schedule_name': schedule_name})
+    finally:
+        mongo.close()
+
+    assert doc['adaptive_state'] == {'escalated': True, 'healthy_streak': 0}
+    assert isinstance(doc['updated_at'], datetime)

laborious/activities/activities.py

@@ -16,19 +16,22 @@ with workflow.unsafe.imports_passed_through():
     from laborious.activities.gates import Gates
     from laborious.activities.mlflow import MLFlow
     from laborious.activities.model_metrics import ModelMetrics
+    from laborious.activities.mongo import MongoActivities
     from laborious.activities.opc import OPC
     from laborious.activities.storage import Storage
     from laborious.utils.connectors_config import build_mlflow_config
 
 
-class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
+class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API, MongoActivities):
     """
     Central orchestrator for all Temporal activities used by Laborious workflows.
 
     Composes Storage (Postgres + MinIO offload), MLFlow (wrapper-based inference and retrain
     via ``SientiaMLflowRepository``), Gates (data quality and ML response filters), OPC exports,
-    drift/simple metrics, and PI Web API writes. The worker constructs one ``Activities`` instance
-    per process and registers its callables on multiple workers bound to different task queues.
+    drift/simple metrics, PI Web API writes, and MongoActivities (the single write point for
+    gate metadata and the simple metrics adaptive state — see `laborious.activities.mongo`).
+    The worker constructs one ``Activities`` instance per process and registers its callables on
+    multiple workers bound to different task queues.
 
     MLflow connectivity: unless ``mlflow_repository`` is injected (tests only), this class builds
     ``SientiaMLflowRepository`` from ``build_mlflow_config()`` so tracking credentials and URL
@@ -70,7 +73,7 @@ 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
+         - mongo_repository: Optional ``MongoDBRepository`` (prediction metadata, adaptive state) 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.
 
@@ -145,6 +148,13 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
         Gates.__init__(
             self,
             minio_repository=minio_repository,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=mc,
+        )
+
+        MongoActivities.__init__(
+            self,
             mongo_repository=mongo_repository,
             logger=logger,
             notification_handler=notification_handler,
@@ -192,3 +202,4 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
         OPC.close(self)
         ModelMetrics.close(self)
         API.close(self)
+        MongoActivities.close(self)

laborious/activities/gates.py

@@ -5,7 +5,6 @@ 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
@@ -13,7 +12,6 @@ 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
@@ -29,6 +27,7 @@ with workflow.unsafe.imports_passed_through():
 InputFilterFunc = Callable[[DataFrame, dict[str, Any]], bool]
 ResponseFilterFunc = Callable[[dict[str, Any], dict[str, Any]], bool]
 ContentFilterFunc = Callable[[DataFrame, dict[str, Any]], bool]
+InputCounterFunc = Callable[[DataFrame, dict[str, Any]], int]
 
 # Input filter function mappings
 input_filter_functions: dict[str, InputFilterFunc] = {
@@ -36,6 +35,42 @@ input_filter_functions: dict[str, InputFilterFunc] = {
     'EMPTY_DATA': filter_empty_data,
 }
 
+
+def _count_specific_variables_null_values(data: DataFrame, config: dict[str, Any]) -> int:
+    """
+    Count rows with a null value in one of the configured variables.
+
+    Mirrors `filter_specific_variables_null_values`'s own trigger condition, so the count
+    reflects exactly what made that filter fire.
+
+    Return:
+        int: Number of rows with a null value in `config['variables']`.
+    """
+    if data.empty:
+        return 0
+    return int((data['variable'].isin(config['variables']) & data['value'].isna()).sum())
+
+
+def _count_empty_data(_data: DataFrame, _config: dict[str, Any]) -> int:
+    """
+    Count rows for the EMPTY_DATA filter.
+
+    `EMPTY_DATA` only triggers when there are no rows at all, so the count is always 0.
+
+    Return:
+        int: Always 0.
+    """
+    return 0
+
+
+# Per-filter row counts for the `counter` field in prediction metadata (gate-local: mirrors each
+# filter's own trigger condition instead of changing `conditional_filters.py`'s bool contract,
+# which `mlflow_content_gate` also depends on for EMPTY_DATA).
+input_filter_counters: dict[str, InputCounterFunc] = {
+    'SPECIFIC_VARIABLES_NULL_VALUES': _count_specific_variables_null_values,
+    'EMPTY_DATA': _count_empty_data,
+}
+
 # Confidence mappings kept separate from function maps to avoid Union types
 input_path_confidence: Mapping[str, int] = {
     'STOP': -1,
@@ -88,14 +123,11 @@ 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,
@@ -106,14 +138,11 @@ 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,
@@ -131,11 +160,6 @@ 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):
@@ -159,56 +183,6 @@ 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]]:
         """
@@ -226,7 +200,9 @@ class Gates(SientiaMonitoring):
         return policy, filter_config
 
     @activity.defn(name='input_gate')
-    def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
+    def input_gate(
+        self, input_data: dict[str, Any]
+    ) -> tuple[str | None, int, str, dict[str, Any] | None]:
         """
         Apply input data quality filters and validation.
 
@@ -241,6 +217,9 @@ class Gates(SientiaMonitoring):
         3. Determines appropriate path decisions (STOP, CONTINUE, REPEAT)
         4. Provides confidence scores and detailed comments
         5. Handles errors gracefully with notification integration
+        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 `MongoActivities.write_prediction_metadata` for the actual write)
 
         Args:
             input_data: Configuration and data for input validation
@@ -251,10 +230,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 filter execution fails or configuration is invalid
@@ -268,7 +251,8 @@ class Gates(SientiaMonitoring):
         data = payload.retrieve(self.minio_repository, metadata)
         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: {filters}', metadata)
@@ -282,7 +266,7 @@ class Gates(SientiaMonitoring):
             try:
                 if input_filter_functions[fil](data, filter_config):
                     self.debug(f'Data not passed the input filter {fil}:{config}', metadata)
-                    filter_output.append(policy)
+                    triggered_filters.append((fil, policy, filter_config))
             except Exception as e:
                 trace = traceback.format_exc()
                 self.send_notification(
@@ -294,16 +278,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'Input gate result: {path_flag}', metadata)
-                return path_flag, input_path_confidence[path_flag], 'Input data with bad quality'
+
+                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]`: a filter added to
+                            # `input_filter_functions` without a matching counter here degrades to
+                            # counter=0 instead of a raw KeyError crashing this prediction (review
+                            # finding, 2026-09-30 — see test enforcing key parity between the two).
+                            'counter': input_filter_counters.get(fil, lambda *_: 0)(
+                                data, filter_config
+                            ),
+                        }
+                        for fil, filter_config in matched
+                    ],
+                }
+
+                return (
+                    path_flag,
+                    input_path_confidence[path_flag],
+                    'Input data with bad quality',
+                    document,
+                )
 
         self.info('Nothing was filtered by the input gate', metadata)
 
         del data
 
-        return None, 0, ''
+        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]:
@@ -699,6 +713,9 @@ class Gates(SientiaMonitoring):
                 - model_id (str): Unique identifier for the ML model
                 - prediction_confidence (float): Confidence score (typically low for errors)
                 - comment (str): Error description or operational comment
+                - metadata_id (str | None, optional): Id of the `prediction_metadata` Mongo
+                    document to reference for this row; `None`/absent when there's nothing to
+                    reference
 
         Returns:
             dict: Formatted default prediction data with error indicators
@@ -716,6 +733,7 @@ class Gates(SientiaMonitoring):
                 'prediction_confidence': [input_data['prediction_confidence']],
                 'prediction_status': ['Bad'],
                 'comments': [input_data['comment']],
+                'metadata_id': [input_data.get('metadata_id')],
             }
         )
 

laborious/activities/model_metrics.py

@@ -4,9 +4,9 @@ with workflow.unsafe.imports_passed_through():
     import time
     import traceback
     import warnings
+    from math import ceil
     from typing import Any
 
-    import numpy as np
     import pandas as pd
     from pandas import DataFrame, Index, Series, to_datetime
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
@@ -16,6 +16,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.observability.sientia_monitoring import SientiaMonitoring
     from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
     from sientia_model.analytics.drift_analysis import DriftAnalysis, DriftInsufficientDataError
+    from sientia_model.metrics.regression import RegressionMetrics
 
     from laborious import metrics
     from laborious.utils.dataframe_debug import build_dataframe_debug_message
@@ -26,6 +27,73 @@ warnings.filterwarnings(
 )
 
 
+def threshold_breaches(
+    rows: list[dict[str, Any]], thresholds: dict[str, float] | None
+) -> list[str]:
+    """
+    Breach messages for metric rows outside their configured thresholds.
+
+    Convention: `<metric>_max` breaches when value > threshold, `<metric>_min` when value <
+    threshold. Shared by the breach alert and the adaptive-state rule so both agree on what a
+    breach is.
+    """
+    thresholds = thresholds or {}
+    breaches = []
+    for row in rows:
+        metric_name, value = row['metric'], row['value']
+        max_key, min_key = f'{metric_name}_max', f'{metric_name}_min'
+        if max_key in thresholds and value > thresholds[max_key]:
+            breaches.append(f'{metric_name}={value} exceeds {max_key}={thresholds[max_key]}')
+        elif min_key in thresholds and value < thresholds[min_key]:
+            breaches.append(f'{metric_name}={value} below {min_key}={thresholds[min_key]}')
+    return breaches
+
+
+_MAX_RECOVERY_RUNS = 12
+_FREQUENCY_UNIT_SECONDS = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}
+
+
+def frequency_to_seconds(frequency: str) -> int:
+    """`'5m'` -> 300. Same `'<int><s|m|h|d>'` format the orchestrator schedules with."""
+    # ponytail: copy of orchestrator/utils/converters.py:parse_frequency (no shared helper in
+    # sientia_do), same int() parsing so both repos accept the same strings, plus a > 0 guard;
+    # move it to sientia_do if a third repo needs it.
+    unit = _FREQUENCY_UNIT_SECONDS.get(frequency[-1:])
+    if unit is None:
+        raise ValueError(f'Invalid frequency: {frequency!r}')
+    value = int(frequency[:-1])
+    if value <= 0:
+        raise ValueError(f'Invalid frequency: {frequency!r}')
+    return value * unit
+
+
+def recovery_runs(interval_minutes: int, frequency_alert: str) -> int:
+    """
+    N of the hysteresis rule: healthy runs needed to renew the whole window without a breach,
+    capped at 12 (SIENTIAPDE-1995 decision 3).
+    """
+    runs = ceil(interval_minutes * 60 / frequency_to_seconds(frequency_alert))
+    return min(_MAX_RECOVERY_RUNS, max(1, runs))
+
+
+def next_adaptive_state(state: dict[str, Any], breached: bool, runs: int) -> dict[str, Any]:
+    """
+    Hysteresis rule of the adaptive frequency (SIENTIAPDE-1995): escalate on the first breach,
+    recover after `runs` consecutive non-breach runs.
+    """
+    if breached:
+        return {'escalated': True, 'healthy_streak': 0}
+    if not state.get('escalated'):
+        return {'escalated': False, 'healthy_streak': 0}
+    try:
+        streak = int(state.get('healthy_streak', 0)) + 1
+    except (TypeError, ValueError, OverflowError):
+        streak = 1  # hand-edited streak (null, text, inf): restart the count, stay escalated
+    if streak >= runs:
+        return {'escalated': False, 'healthy_streak': 0}
+    return {'escalated': True, 'healthy_streak': streak}
+
+
 class ModelMetrics(SientiaMonitoring):
     """
     Metrics and statistical analysis activities for the Laborious pipeline.
@@ -365,62 +433,62 @@ class ModelMetrics(SientiaMonitoring):
     @activity.defn(name='calculate_simple_metrics')
     def calculate_simple_metrics(self, input_data: dict[str, Any]) -> list[dict]:
         """
-        Calculate simple metrics for a model. Metrics available are:
-            - rmse
-            - mse
-            - mae
-            - r2
-            - accuracy
-            - precision
-            - recall
-            - f1
+        Calculate simple regression metrics for a model using RegressionMetrics.
+
         Args:
-            input_data (dict[str, Any]): Input data containing:
+            input_data: Input data containing:
                 - metadata (dict): Workflow execution metadata
                 - model_id (str): ID of the MLFlow model
-                - target_data (pd.DataFrame): Target data for calculating metrics, containing target and prediction columns
-                - metrics (list[str]): List of metrics to calculate
+                - target_data (list[dict]): Target data with target, prediction, timestamp columns
+                - metrics (list[str]): Metric names to calculate
+                - interval_minutes (int): Window interval in minutes
+                - model_type (str | None): Model algorithm type (for r2 lock)
         Returns:
-            dict[Hashable, Any]: Dictionary containing the calculated metrics
+            list[dict]: Records with metric, value, model_id, timestamp, data_size, interval_minutes
         """
-
         metadata = input_data['metadata']
         model_id = input_data['model_id']
         target_data = DataFrame(input_data['target_data'])
-        metric_names = input_data['metrics']
+        metric_names = list(input_data['metrics'])
         interval_minutes = input_data['interval_minutes']
+        model_type = input_data.get('model_type')
 
         data_size = target_data.shape[0]
 
-        output_data = []
+        # Filter r2 when model_type is known and unsupported
+        if (
+            model_type
+            and 'r2' in metric_names
+            and not RegressionMetrics.is_r2_supported(model_type)
+        ):
+            metric_names = [m for m in metric_names if m != 'r2']
+            self.warning(
+                f'r2 excluded for model {model_id}: not supported for model_type={model_type}',
+                metadata,
+            )
+            self.send_notification(
+                metadata=metadata,
+                notification_id='SIMPLE_METRICS_R2_UNSUPPORTED',
+                message=f'r2 excluded: not a valid metric for model_type={model_type}',
+                block='model_metrics',
+                level=NotificationLevel.WARNING,
+            )
 
-        diff = target_data['target'] - target_data['prediction']
-        diff_squared = diff**2
+        if not metric_names:
+            self.info(f'No metrics to calculate for model {model_id} after filtering', metadata)
+            return []
 
         self.info(f'Calculating simple metrics for model {model_id}: {metric_names}', metadata)
 
-        for metric in metric_names:
-            if metric == 'rmse':
-                output_data.append({'metric': 'rmse', 'value': np.sqrt(np.mean(diff_squared))})
-            elif metric == 'mse':
-                output_data.append({'metric': 'mse', 'value': np.mean(diff_squared)})
-            elif metric == 'mae':
-                output_data.append({'metric': 'mae', 'value': np.mean(np.abs(diff))})
-            elif metric == 'r2':
-                y_true = target_data['target']
-                y_mean = np.mean(y_true)
+        # Build Series with DatetimeIndex for RegressionMetrics
+        timestamps = pd.to_datetime(target_data['timestamp'])
+        real_data = Series(target_data['target'].values, index=timestamps, dtype=float)
+        predictions = Series(target_data['prediction'].values, index=timestamps, dtype=float)
 
-                ss_res = np.sum(diff_squared)
-                ss_tot = np.sum((y_true - y_mean) ** 2)
-
-                # Evita divisão por zero
-                if ss_tot == 0:
-                    r2_score = 0.0
-                else:
-                    r2_score = 1 - (ss_res / ss_tot)
-
-                output_data.append({'metric': 'r2', 'value': r2_score})
+        regression = RegressionMetrics(real_data, predictions)
+        output_data = regression.calculate(metric_names)
 
+        # Wrap with metadata columns matching the existing output schema
         data = DataFrame(output_data)
         data['model_id'] = model_id
         data['timestamp'] = target_data['timestamp'].max()
@@ -429,4 +497,22 @@ class ModelMetrics(SientiaMonitoring):
 
         self._debug_dataframe(f'Simple metrics dataframe: Size {data.shape}', data, metadata)
 
+        # Threshold alerting (optional — no crash when absent)
+        for breach_msg in threshold_breaches(output_data, input_data.get('thresholds')):
+            message = (
+                f'Threshold breach for model {model_id} '
+                f'(window {interval_minutes} min): {breach_msg}'
+            )
+            self.warning(message, metadata)
+            self.send_notification(
+                # Model-level trigger: the orchestrator dedups alerts on
+                # `<trigger>:<notification_id>`, so one breach seen by several windows
+                # (one schedule each) becomes one alert (SIENTIAPDE-1995).
+                metadata={**metadata, 'schedule_name': f'{model_id}-simple_metrics'},
+                notification_id='SIMPLE_METRICS_THRESHOLD_BREACH',
+                message=message,
+                block='model_metrics',
+                level=NotificationLevel.WARNING,
+            )
+
         return data.to_dict(orient='records')

laborious/activities/mongo.py

@@ -0,0 +1,227 @@
+from temporalio import activity, workflow
+
+with workflow.unsafe.imports_passed_through():
+    from typing import Any
+
+    from bson import ObjectId
+    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
+    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.mongodb_repository_sync import MongoDBRepository
+    from sientia_do.temporal.constants import now
+
+    from laborious.activities.model_metrics import (
+        next_adaptive_state,
+        recovery_runs,
+        threshold_breaches,
+    )
+
+
+class MongoActivities(SientiaMonitoring):
+    """
+    Activities that read from or write to MongoDB, sharing one `MongoDBRepository`.
+
+    - `write_prediction_metadata`: centralized, best-effort writer for prediction gate metadata
+      (triggered filter, policy, counter). Gates only compute and return a metadata document;
+      they never write it themselves. Keeping the write here, instead of in each gate, puts the
+      one real write point in the subworkflow that's about to persist a `predictions` row — the
+      only place that reliably knows whether one is being written at all — and lets other
+      metadata sources reuse the same activity instead of duplicating write logic wherever they
+      originate.
+    - `update_simple_metrics_adaptive_state`: adaptive frequency state on the `pipelines` doc of
+      a simple metrics schedule (SIENTIAPDE-1995).
+    """
+
+    mongo_repository: MongoDBRepository | None = None
+    _PREDICTION_METADATA_COLLECTION = 'prediction_metadata'
+    _PIPELINES_COLLECTION = 'pipelines'
+
+    def __init__(
+        self,
+        mongo_repository: MongoDBRepository | None = None,
+        logger: Logger | None = None,
+        notification_handler: NotificationHandler | None = None,
+        metrics_controller: MetricsController | None = None,
+    ):
+        """
+        Initialize the Mongo activities.
+
+        Args:
+            mongo_repository: MongoDB, where gate metadata (triggered filter, policy, counter) is
+                written best-effort and the simple metrics adaptive state is kept. `None` is a
+                valid state: both activities become no-ops.
+
+        Raises:
+            Exception: If BaseActivity initialization fails
+        """
+        self.mongo_repository = mongo_repository
+        SientiaMonitoring.__init__(
+            self,
+            logger=logger,
+            notification_handler=notification_handler,
+            metrics_controller=metrics_controller,
+        )
+
+    def close(self) -> None:
+        """
+        Close the Mongo repository and clean up resources.
+        """
+        if self.mongo_repository is not None:
+            try:
+                self.mongo_repository.close()
+            finally:
+                self.mongo_repository = None
+        SientiaMonitoring.shutdown(self)
+
+    def __del__(self):
+        self.close()
+
+    @activity.defn(name='write_prediction_metadata')
+    def write_prediction_metadata(self, input_data: dict[str, Any]) -> str | None:
+        """
+        Best-effort write of one or more gate metadata documents to Mongo, combined into a single
+        `prediction_metadata` document.
+
+        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 call-site 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 call.
+        - Nothing is written when there's nothing relevant to report. An empty/missing
+          `documents` list (or one full of falsy/empty entries) 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.
+
+        Takes a *list* of documents, not a single one: each caller contributes its own document
+        (shaped like `{'filters': [...]}`), and this flattens every `filters` entry across all of
+        them into one `prediction_metadata` document before the single insert. Today only one
+        gate ever contributes a document per prediction cycle (the gate chain short-circuits on
+        the first non-`None` `path_flag`), but the list shape means this activity won't need to
+        change again once more than one source contributes metadata for the same cycle.
+
+        `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:
+            input_data: Dict with:
+                - documents (list[dict[str, Any]]): Metadata payloads to combine, each shaped
+                  like `{'filters': [...]}`. Missing/empty, or entries with no `filters`, skip
+                  the write entirely.
+                - metadata (dict): 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.
+        """
+        metadata = input_data['metadata']
+        documents = input_data.get('documents') or []
+
+        filters = [
+            entry for document in documents if document for entry in document.get('filters', [])
+        ]
+
+        if not filters or self.mongo_repository is None:
+            return None
+
+        generated_id = str(ObjectId())
+
+        try:
+            self.mongo_repository.insert(
+                self._PREDICTION_METADATA_COLLECTION,
+                {'filters': filters, '_id': generated_id},
+                metadata,
+            )
+        except Exception as e:
+            self.error(f'Failed to write prediction metadata: {e}', metadata)
+            return None
+
+        return generated_id
+
+    @activity.defn(name='update_simple_metrics_adaptive_state')
+    def update_simple_metrics_adaptive_state(self, input_data: dict[str, Any]) -> dict | None:
+        """
+        Update the adaptive frequency state on the schedule's `pipelines` doc.
+
+        The orchestrator derives the schedule frequency from `adaptive_state.escalated` and
+        `frequency_alert`. State is read from Mongo, not from the workflow input, because the
+        input only refreshes on the orchestrator tick and would be stale while escalated.
+        `updated_at` is bumped only when `escalated` flips: that is what makes the orchestrator
+        rewrite the schedule, and a streak change alone does not change the frequency.
+
+        Args:
+            input_data: Dict with:
+                - metadata (dict): Workflow metadata; `schedule_name` identifies the doc.
+                - simple_metrics (list[dict]): Rows from `calculate_simple_metrics`; empty for an
+                  empty or under-sampled window, which counts as a non-breach run.
+                - thresholds (dict): `simple_metrics_thresholds` from `model_config` (may be None: no breach).
+                - interval_minutes (int): Window size, used to compute the recovery runs.
+
+        Return:
+            The new state, or `None` when adaptive mode is off, Mongo is not configured, the doc
+            is missing or `frequency_alert` is invalid.
+
+        Raises:
+            Exception: Mongo read/write failures, so the workflow run fails explicitly.
+        """
+        metadata = input_data['metadata']
+        if self.mongo_repository is None:
+            return None
+
+        # Mongo errors propagate on purpose: the run must fail explicitly so the problem shows
+        # up in the Temporal metrics (PR #59 review). Metrics were already exported by then.
+        query = {'schedule_name': metadata['schedule_name']}
+        pipeline = self.mongo_repository.find_one(self._PIPELINES_COLLECTION, query, metadata)
+        if not pipeline:
+            # Likely misconfig: MONGODB_DATABASE_NAME differs from the orchestrator's, or the
+            # schedule_name does not match its doc.
+            self.warning(
+                f'No pipelines doc for schedule {metadata["schedule_name"]}: '
+                'adaptive state skipped',
+                metadata,
+            )
+            return None
+        frequency_alert = pipeline.get('frequency_alert')
+        if not frequency_alert:
+            return None
+
+        try:
+            runs = recovery_runs(input_data['interval_minutes'], str(frequency_alert))
+        except (TypeError, ValueError):
+            # Config error, not a failure: the orchestrator ignores an invalid `frequency_alert`
+            # too (keeps `frequency`) and already sends SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR
+            # for it. `interval_minutes` is untyped Mongo config as well (e.g. "60" or null).
+            self.warning(
+                f'Invalid frequency_alert={frequency_alert!r} or '
+                f'interval_minutes={input_data["interval_minutes"]!r}: adaptive state skipped',
+                metadata,
+            )
+            return None
+
+        breached = bool(
+            threshold_breaches(input_data['simple_metrics'], input_data.get('thresholds'))
+        )
+        state = pipeline.get('adaptive_state')
+        if not isinstance(state, dict):
+            # Missing or hand-edited garbage: not escalated, same as the orchestrator reads it.
+            state = {'escalated': False, 'healthy_streak': 0}
+        new_state = next_adaptive_state(state, breached, runs)
+        if new_state == state:
+            return new_state
+
+        update: dict[str, Any] = {'adaptive_state': new_state}
+        if new_state['escalated'] != bool(state.get('escalated')):
+            # BSON datetime, never a string: the orchestrator normalizes `updated_at` with
+            # `.replace(tzinfo=UTC).strftime(...)` and a string would fail its whole tick.
+            update['updated_at'] = now()
+        self.mongo_repository.update_one(
+            self._PIPELINES_COLLECTION, query, {'$set': update}, metadata
+        )
+
+        self.info(f'Adaptive state of {query["schedule_name"]}: {state} -> {new_state}', metadata)
+        return new_state

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 `MongoActivities.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/worker/worker.py

@@ -217,6 +217,7 @@ async def main():
                 activities.load_custom_query,
                 activities.calculate_simple_metrics,
                 activities.export_data_to_postgres,
+                activities.update_simple_metrics_adaptive_state,
             ],
             logger=logger,
             runtime='core',
@@ -249,6 +250,8 @@ async def main():
                 activities.format_transformed_data,
                 activities.format_prediction,
                 activities.format_default_prediction,
+                # Prediction metadata
+                activities.write_prediction_metadata,
                 # OPC
                 activities.write_opc_data,
                 # Postgres / MinIO offload

laborious/workflows/simple_metrics.py

@@ -31,6 +31,17 @@ class SimpleMetrics:
 
         model_config = input_data['model_config']
         target_name = model_config['target']
+        model_type = model_config.get('model_type')
+        thresholds = model_config.get('simple_metrics_thresholds')
+        min_samples = model_config.get('simple_metrics_min_samples')
+        # Untyped Mongo config: anything but a positive int (bool excluded) turns the guard off
+        # instead of raising in workflow code; falsy values mean "unset" and stay silent.
+        if type(min_samples) is not int or min_samples <= 0:
+            if min_samples:
+                workflow.logger.warning(
+                    f'Invalid simple_metrics_min_samples={min_samples!r}: min samples guard off'
+                )
+            min_samples = None
 
         # Built from internal Temporal workflow config, not user input (bandit B608).
         query = f"""
@@ -60,37 +71,63 @@ class SimpleMetrics:
             start_to_close_timeout=timedelta(seconds=300),
         )
 
-        if not target_data:
-            return
+        # Too few rows for a reliable value: metrics are still persisted, but a noisy short
+        # window must neither alert nor escalate (SIENTIAPDE-1995).
+        enough_samples = bool(target_data) and (not min_samples or len(target_data) >= min_samples)
+        if target_data and not enough_samples:
+            workflow.logger.info(
+                f'{len(target_data)} rows < simple_metrics_min_samples={min_samples}: '
+                'threshold alerting and escalation skipped'
+            )
 
-        simple_metrics = await workflow.execute_local_activity_method(
-            Activities.calculate_simple_metrics,
+        simple_metrics = None
+        if target_data:
+            simple_metrics = await workflow.execute_local_activity_method(
+                Activities.calculate_simple_metrics,
+                {
+                    **metadata,
+                    'model_id': model_id,
+                    'target_data': target_data,
+                    'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
+                    'interval_minutes': interval_minutes,
+                    'model_type': model_type,
+                    'thresholds': thresholds if enough_samples else None,
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=300),
+            )
+
+        if simple_metrics:
+            await workflow.execute_activity_method(
+                Activities.export_data_to_postgres,
+                {
+                    **metadata,
+                    'data': simple_metrics,
+                    'schema': input_data['schema'],
+                    'table_name': input_data['target_table_name'],
+                    'timestamp_conversion': {
+                        'column': 'timestamp',
+                        'format': DATETIME_FORMAT_WITH_TZ,
+                    },
+                },
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(seconds=300),
+            )
+
+        # No try/except: a failure here must fail the run so it shows up in the Temporal metrics
+        # (metrics are already persisted above).
+        await workflow.execute_activity_method(
+            Activities.update_simple_metrics_adaptive_state,
             {
                 **metadata,
-                'model_id': model_id,
-                'target_data': target_data,
-                'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
+                # Only a valid breach escalates: an empty or under-sampled window is sent as no
+                # rows, which counts as a healthy run (spec decision 8). Always called, even
+                # without thresholds (no breach), so an escalated doc recovers after they are
+                # removed.
+                'simple_metrics': (simple_metrics or []) if enough_samples else [],
+                'thresholds': thresholds,
                 'interval_minutes': interval_minutes,
             },
             retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=300),
-        )
-
-        if not simple_metrics:
-            return
-
-        await workflow.execute_activity_method(
-            Activities.export_data_to_postgres,
-            {
-                **metadata,
-                'data': simple_metrics,
-                'schema': input_data['schema'],
-                'table_name': input_data['target_table_name'],
-                'timestamp_conversion': {
-                    'column': 'timestamp',
-                    'format': DATETIME_FORMAT_WITH_TZ,
-                },
-            },
-            retry_policy=retry_policy,
-            start_to_close_timeout=timedelta(seconds=300),
+            start_to_close_timeout=timedelta(minutes=1),
         )

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -75,6 +75,11 @@ class FormatAndExportPrediction:
                         Required when path_flag is None
                     - comment (str): Operational comment or error description
                         Required when path_flag is not None
+                    - metadata_id (str | None): Id of the `prediction_metadata` Mongo document to
+                        reference for this row, already written by the caller; `None` when there's
+                        nothing to reference. Only meaningful when path_flag is not None — the
+                        normal (path_flag is None) path never has metadata to reference, so this
+                        key is irrelevant there.
 
         Returns:
             None: The workflow completes successfully when all export operations finish
@@ -154,6 +159,7 @@ class FormatAndExportPrediction:
                     'model_id': input_data['model_id'],
                     'prediction_confidence': prediction_confidence,
                     'comment': input_data['comment'],
+                    'metadata_id': input_data.get('metadata_id'),
                 },
                 retry_policy=retry_policy,
                 start_to_close_timeout=timedelta(seconds=60),

laborious/workflows/sub_workflows/prediction_process.py

@@ -132,7 +132,9 @@ class PredictionProcess:
             'path_priority': input_data['path_priority'],
         }
 
-        path_flag, confidence, comment = await workflow.execute_local_activity_method(
+        # `document` describes which filter(s) produced `path_flag` (`None` when nothing was
+        # filtered); `path_flag_handler` is the one that decides whether and how to persist it.
+        path_flag, confidence, comment, document = await workflow.execute_local_activity_method(
             Activities.input_gate,
             gate_input,
             retry_policy=retry_policy,
@@ -141,7 +143,7 @@ class PredictionProcess:
 
         # Handle path decision based on filter results
         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
 
@@ -247,6 +249,51 @@ class PredictionProcess:
             },
         )
 
+    async def _write_prediction_metadata(
+        self, metadata: dict[str, Any], document: dict[str, Any] | None
+    ) -> str | None:
+        """
+        Write a gate's metadata document via the centralized write activity.
+
+        Shared by the CONTINUE and REPEAT branches of `path_flag_handler` — both persist a new
+        `predictions` row (`REPEAT` reuses the previous row's values but still inserts a new
+        one), so both need a `metadata_id` to reference it.
+
+        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.
+
+        Return:
+            The generated id, or `None` when there's nothing to write, including when the
+            activity call itself fails (see `MongoActivities.write_prediction_metadata`).
+        """
+        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).
+            return None
+
+        try:
+            return await workflow.execute_activity_method(
+                Activities.write_prediction_metadata,
+                {**metadata, 'documents': [document]},
+                retry_policy=retry_policy,
+                start_to_close_timeout=timedelta(minutes=1),
+            )
+        except Exception as e:
+            # Business rule 1: writing metadata must never block or fail the prediction write.
+            # `write_prediction_metadata` already swallows Mongo-specific failures internally and
+            # returns `None`, but that protection doesn't cover the activity call itself failing
+            # to run at all (its `retry_policy` has `maximum_attempts=1` — no retry — so a
+            # timeout, a momentarily unavailable worker, or any other dispatch failure raises
+            # straight into the workflow). Catching it here, instead of just at the Mongo client
+            # level, is what actually makes the rule hold (review finding, 2026-10-01).
+            workflow.logger.warning(f'Failed to write prediction metadata, continuing anyway: {e}')
+            return None
+
     async def path_flag_handler(
         self,
         data: dict[str, Any],
@@ -255,6 +302,7 @@ class PredictionProcess:
         confidence: int,
         last_timestamp: str,
         comment: str,
+        document: dict[str, Any] | None = None,
     ) -> bool:
         """
         Handle path decisions based on filter results and confidence levels.
@@ -280,6 +328,13 @@ class PredictionProcess:
             confidence: Confidence level from filter validation
             last_timestamp: Last processed timestamp
             comment: Additional information about the filter result
+            document: Metadata describing the filter(s) that produced `path_flag`, as returned by
+                the gate (`None` when nothing was filtered, or when the gate that produced this
+                `path_flag` doesn't build one yet). Written via `write_prediction_metadata` for
+                `CONTINUE` and `REPEAT` alike — both persist a `predictions` row (`REPEAT` reuses
+                the previous row's values but still inserts a new one), so both can reference the
+                resulting `metadata_id`; `STOP` never reaches a write at all, so its `document` is
+                simply never used here.
 
         Returns:
             bool: True if processing should stop, False to continue
@@ -304,7 +359,10 @@ class PredictionProcess:
             # Stop processing and exit workflow
             return True
         elif path_flag == 'REPEAT':
-            # Repeat last prediction if available
+            metadata_id = await self._write_prediction_metadata(metadata, document)
+            # 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,
                 {
@@ -313,12 +371,14 @@ class PredictionProcess:
                     'table_name': table_name,
                     'model': model_id,
                     'last_timestamp': last_timestamp,
+                    'metadata_id': metadata_id,
                 },
                 retry_policy=retry_policy,
                 start_to_close_timeout=timedelta(minutes=1),
             )
             return True
         elif path_flag == 'CONTINUE':
+            metadata_id = await self._write_prediction_metadata(metadata, document)
             # call write workflow
             await workflow.execute_child_workflow(
                 'subworkflow.format_and_export_prediction',
@@ -335,6 +395,7 @@ class PredictionProcess:
                     'table_name': table_name,
                     'transform_table_name': transform_table_name,
                     'comment': comment,
+                    'metadata_id': metadata_id,
                     'opc_output_config': input_data['opc_output_config'],
                     'pi_web_api_output_config': input_data['pi_web_api_output_config'],
                     'prediction_store_policy': input_data['prediction_store_policy'],

requirements-local.txt

@@ -6,7 +6,7 @@ sqlalchemy<2.1
 asyncua==1.0.6
 redis
 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
-git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.10.0
+git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.13.1
 prometheus-client   
 # fsspec/s3fs pinned as a pair: s3fs pins fsspec exactly, so a free `s3fs` can make the
 # resolver fail against evidently's fsspec>=2024.2.0. Versions match the
@@ -22,4 +22,4 @@ hyperopt
 shap
 pycurl
 scipy<1.14.0
-scikit-learn==1.5.2
+scikit-learn==1.8.0

requirements.txt

@@ -6,7 +6,7 @@ sqlalchemy<2.1
 asyncua==1.0.6
 redis
 sientia_do>=1.12.2
-sientia_model>=0.8.2
+sientia_model>=0.12.0
 prometheus-client
 # fsspec/s3fs pinned as a pair: s3fs pins fsspec exactly, so a free `s3fs` can make the
 # resolver fail against evidently's fsspec>=2024.2.0. Versions match the

tests/laborious/activities/test_activities.py

@@ -5,6 +5,7 @@ from laborious.activities.api import API
 from laborious.activities.gates import Gates
 from laborious.activities.mlflow import MLFlow
 from laborious.activities.model_metrics import ModelMetrics
+from laborious.activities.mongo import MongoActivities
 from laborious.activities.opc import OPC
 from laborious.activities.storage import Storage
 
@@ -13,6 +14,7 @@ from laborious.activities.storage import Storage
 @patch('laborious.activities.activities.MLFlow.__init__')
 @patch('laborious.activities.activities.OPC.__init__')
 @patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.MongoActivities.__init__')
 @patch('laborious.activities.activities.ModelMetrics.__init__')
 @patch('laborious.activities.activities.API.__init__')
 @patch('laborious.activities.activities.MinioRepository')
@@ -22,6 +24,7 @@ def test___init__(
     mock_minio_repository,
     mock_api_init,
     mock_model_metrics_init,
+    mock_mongo_activities_init,
     mock_gates_init,
     mock_opc_init,
     mock_mlflow_init,
@@ -84,6 +87,7 @@ def test___init__(
     assert isinstance(activities, Gates)
     assert isinstance(activities, ModelMetrics)
     assert isinstance(activities, API)
+    assert isinstance(activities, MongoActivities)
 
     mock_storage_init.assert_called_once_with(
         ANY,
@@ -122,6 +126,13 @@ def test___init__(
     mock_gates_init.assert_called_once_with(
         ANY,
         minio_repository=mock_minio_repository.return_value,
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )
+
+    mock_mongo_activities_init.assert_called_once_with(
+        ANY,
         mongo_repository=mongo_repository,
         logger=logger,
         notification_handler=notification_handler,
@@ -161,6 +172,7 @@ def test___init__(
 @patch('laborious.activities.activities.MLFlow')
 @patch('laborious.activities.activities.OPC')
 @patch('laborious.activities.activities.Gates')
+@patch('laborious.activities.activities.MongoActivities')
 @patch('laborious.activities.activities.ModelMetrics')
 @patch('laborious.activities.activities.API')
 @patch('laborious.activities.activities.MinioRepository')
@@ -168,6 +180,7 @@ def test_shutdown(
     _mock_minio_repository,
     mock_api_init,
     mock_model_metrics_init,
+    mock_mongo_activities_init,
     mock_gates_init,
     mock_opc_init,
     mock_mlflow_init,
@@ -230,6 +243,7 @@ def test_shutdown(
     mock_gates_init.close.assert_called_once()
     mock_model_metrics_init.close.assert_called_once()
     mock_api_init.close.assert_called_once()
+    mock_mongo_activities_init.close.assert_called_once()
 
 
 @patch('laborious.activities.activities.SientiaMLflowRepository')
@@ -238,6 +252,7 @@ def test_shutdown(
 @patch('laborious.activities.activities.MLFlow.__init__')
 @patch('laborious.activities.activities.OPC.__init__')
 @patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.MongoActivities.__init__')
 @patch('laborious.activities.activities.ModelMetrics.__init__')
 @patch('laborious.activities.activities.API.__init__')
 @patch('laborious.activities.activities.MinioRepository')
@@ -247,6 +262,7 @@ def test___init___builds_mlflow_repository_when_not_provided(
     mock_minio_repository,
     _mock_api_init,
     _mock_model_metrics_init,
+    _mock_mongo_activities_init,
     _mock_gates_init,
     _mock_opc_init,
     _mock_mlflow_init,
@@ -310,6 +326,7 @@ def test___init___builds_mlflow_repository_when_not_provided(
 @patch('laborious.activities.activities.MLFlow.__init__')
 @patch('laborious.activities.activities.OPC.__init__')
 @patch('laborious.activities.activities.Gates.__init__')
+@patch('laborious.activities.activities.MongoActivities.__init__')
 @patch('laborious.activities.activities.ModelMetrics.__init__')
 @patch('laborious.activities.activities.API.__init__')
 @patch('laborious.activities.activities.MinioRepository')
@@ -319,6 +336,7 @@ def test___init___builds_mongo_repository_when_not_provided(
     mock_minio_repository,
     _mock_api_init,
     _mock_model_metrics_init,
+    mock_mongo_activities_init,
     mock_gates_init,
     _mock_opc_init,
     _mock_mlflow_init,
@@ -373,10 +391,17 @@ def test___init___builds_mongo_repository_when_not_provided(
         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,
+        logger=logger,
+        notification_handler=notification_handler,
+        metrics_controller=mock_metrics_controller.return_value,
+    )
+    # The same repository built here is the one handed to MongoActivities, not a second
+    # instance.
+    mock_mongo_activities_init.assert_called_once_with(
+        ANY,
         mongo_repository=mock_mongo_repository_cls.return_value,
         logger=logger,
         notification_handler=notification_handler,

tests/laborious/activities/test_gates.py

@@ -6,7 +6,7 @@ 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
+from laborious.activities.gates import Gates, input_filter_counters, input_filter_functions
 
 
 @fixture(autouse=True)
@@ -35,14 +35,8 @@ def _minio_payload(retrieve_return, status=None):
 
 
 @fixture
-def mongo() -> MagicMock:
-    return MagicMock()
-
-
-@fixture
-def gates_activity(mongo: MagicMock):
+def gates_activity():
     gates = Gates(
-        mongo_repository=mongo,
         logger=MagicMock(),
         notification_handler=MagicMock(),
         metrics_controller=MagicMock(spec=MetricsController),
@@ -67,73 +61,11 @@ 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_filter_counters_covers_every_input_filter_function():
+    # A filter added to input_filter_functions without a matching entry here would degrade to
+    # counter=0 (see the `.get(..., lambda *_: 0)` fallback in input_gate) instead of crashing —
+    # but this test should still catch the gap before it ships (review finding, 2026-09-30).
+    assert set(input_filter_counters) == set(input_filter_functions)
 
 
 def test_input_gate_invalid_filter(gates_activity):
@@ -149,7 +81,7 @@ def test_input_gate_invalid_filter(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.error.assert_called_once_with(
         'Filter INVALID_FILTER not found', metadata['metadata']
     )
@@ -173,7 +105,7 @@ def test_input_gate_filter_exception(mock_input_filter_functions, gates_activity
     result = gates_activity.input_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='INTPUT_GATE_ERROR__EMPTY_DATA',
@@ -197,7 +129,7 @@ def test_input_gate_no_filters(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
@@ -211,10 +143,14 @@ def test_input_gate_with_filter(gates_activity):
     }
 
     # Act
-    result = gates_activity.input_gate(input_data)
+    path_flag, confidence, comment, document = gates_activity.input_gate(input_data)
 
-    # Assert
-    assert result == ('STOP', -1, 'Input data with bad quality')
+    # Assert: the gate only computes the document now — it never writes anywhere, and never
+    # checks path_flag to decide whether to build one (that decision moved to the caller, see
+    # `MongoActivities.write_prediction_metadata` and `prediction_process.py`), so even a
+    # STOP resolution still gets its document built.
+    assert (path_flag, confidence, comment) == ('STOP', -1, 'Input data with bad quality')
+    assert document == {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
     gates_activity.debug.assert_called()
 
 
@@ -231,7 +167,12 @@ def test_input_gate_with_filter_lowercase_keys(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == ('STOP', -1, 'Input data with bad quality')
+    assert result == (
+        'STOP',
+        -1,
+        'Input data with bad quality',
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]},
+    )
 
 
 def test_input_gate_with_filter_capitalized_keys(gates_activity):
@@ -247,7 +188,12 @@ def test_input_gate_with_filter_capitalized_keys(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == ('STOP', -1, 'Input data with bad quality')
+    assert result == (
+        'STOP',
+        -1,
+        'Input data with bad quality',
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]},
+    )
 
 
 def test_input_gate_with_filter_not_caught(gates_activity):
@@ -263,10 +209,94 @@ def test_input_gate_with_filter_not_caught(gates_activity):
     result = gates_activity.input_gate(input_data)
 
     # Assert
-    assert result == (None, 0, '')
+    assert result == (None, 0, '', None)
     gates_activity.debug.assert_called()
 
 
+def test_input_gate_repeat_also_builds_document(gates_activity):
+    # Arrange: REPEAT used to be treated like STOP (no document at all) — it isn't anymore, since
+    # REPEAT does insert a new `predictions` row (reusing the previous row's values), so it needs
+    # a document/metadata_id just like CONTINUE (2026-10-01 redesign).
+    input_data = {
+        **metadata,
+        'filters': {'EMPTY_DATA': {'POLICY': 'REPEAT', 'CONFIG': {}}},
+        'data': _minio_payload(DataFrame({'value': []})),
+        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+    }
+
+    # Act
+    result = gates_activity.input_gate(input_data)
+
+    # Assert
+    assert result == (
+        'REPEAT',
+        -1,
+        'Input data with bad quality',
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]},
+    )
+
+
+def test_input_gate_continue_builds_document_with_counter(gates_activity):
+    # Arrange: two rows with a null in one of the configured variables ('x'); a third row is
+    # clean, to confirm the counter reflects only the rows that actually matched the condition.
+    input_data = {
+        **metadata,
+        'filters': {
+            'SPECIFIC_VARIABLES_NULL_VALUES': {
+                'POLICY': 'CONTINUE',
+                'CONFIG': {'variables': ['x']},
+            }
+        },
+        'data': _minio_payload(
+            DataFrame({'variable': ['x', 'x', 'y'], 'value': [None, None, None]})
+        ),
+        'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+    }
+
+    # Act
+    path_flag, confidence, comment, document = gates_activity.input_gate(input_data)
+
+    # Assert
+    assert (path_flag, confidence, comment) == ('CONTINUE', 2, 'Input data with bad quality')
+    assert document == {
+        'filters': [{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}]
+    }
+
+
+def test_input_gate_multiple_filters_same_policy_lists_each(gates_activity):
+    # Arrange: force both input filters to trigger (not reachable with today's real filter
+    # conditions — EMPTY_DATA and SPECIFIC_VARIABLES_NULL_VALUES are mutually exclusive) to lock
+    # in the list-building behavior approved by Vitor (2026-09-30): one entry per filter sharing
+    # the winning policy, instead of a single joined/aggregated value.
+    with patch('laborious.activities.gates.input_filter_functions') as mock_filters:
+        mock_filters.__contains__.return_value = True
+        mock_filters.__getitem__.return_value = MagicMock(return_value=True)
+
+        input_data = {
+            **metadata,
+            'filters': {
+                'SPECIFIC_VARIABLES_NULL_VALUES': {
+                    'POLICY': 'CONTINUE',
+                    'CONFIG': {'variables': ['x']},
+                },
+                'EMPTY_DATA': {'POLICY': 'CONTINUE', 'CONFIG': {}},
+            },
+            'data': _minio_payload(DataFrame({'variable': ['x'], 'value': [None]})),
+            'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
+        }
+
+        # Act
+        path_flag, _, _, document = gates_activity.input_gate(input_data)
+
+    # Assert
+    assert path_flag == 'CONTINUE'
+    assert {f['name'] for f in document['filters']} == {
+        'SPECIFIC_VARIABLES_NULL_VALUES',
+        'EMPTY_DATA',
+    }
+    assert all(f['policy'] == 'CONTINUE' for f in document['filters'])
+
+
 def test_mlflow_response_gate_invalid_filter(gates_activity):
     # Arrange
     input_data = {

tests/laborious/activities/test_model_metrics.py

@@ -7,7 +7,13 @@ from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
 from sientia_model.analytics.drift_analysis import DriftInsufficientDataError
 
-from laborious.activities.model_metrics import ModelMetrics
+from laborious.activities.model_metrics import (
+    ModelMetrics,
+    frequency_to_seconds,
+    next_adaptive_state,
+    recovery_runs,
+    threshold_breaches,
+)
 
 
 @fixture
@@ -46,6 +52,23 @@ metadata = {
 }
 
 
+def _mock_regression_metrics_class(calculate_return):
+    """Returns a patch context manager that mocks RegressionMetrics."""
+    mock_instance = MagicMock()
+    mock_instance.calculate.return_value = calculate_return
+    mock_class = MagicMock(return_value=mock_instance)
+    mock_class.is_r2_supported = MagicMock(return_value=True)
+    mock_class.supported_metrics = MagicMock(return_value=['rmse', 'mse', 'mae', 'r2'])
+    return (
+        patch(
+            'laborious.activities.model_metrics.RegressionMetrics',
+            mock_class,
+        ),
+        mock_class,
+        mock_instance,
+    )
+
+
 def test_calculate_drift_invalid_chunk_period(model_metrics_activity):
     # Arrange
     input_data = {
@@ -694,7 +717,6 @@ def test_get_drift_metrics_dataframe_error(
 
 
 def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'],
@@ -711,28 +733,28 @@ def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
-
-    # Assert
-    assert len(result['metric']) == 4
-    assert 'rmse' in result['metric'].values
-    assert 'mse' in result['metric'].values
-    assert 'mae' in result['metric'].values
-    assert 'r2' in result['metric'].values
-    assert all(model_id == 'test_model_id' for model_id in result['model_id'].values)
-    assert all(timestamp == '2023-05-26 11:12:29' for timestamp in result['timestamp'].values)
-    assert all(data_size == 3 for data_size in result['data_size'].values)
-    assert all(interval_minutes == 5 for interval_minutes in result['interval_minutes'].values)
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['rmse', 'mse', 'mae', 'r2']",
-        metadata['metadata'],
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'rmse', 'value': 0.1},
+            {'metric': 'mse', 'value': 0.01},
+            {'metric': 'mae', 'value': 0.1},
+            {'metric': 'r2', 'value': 0.99},
+        ]
     )
-    model_metrics_activity.debug.assert_called_once()
+
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 4
+    assert set(result['metric'].values) == {'rmse', 'mse', 'mae', 'r2'}
+    assert all(mid == 'test_model_id' for mid in result['model_id'].values)
+    assert all(ts == '2023-05-26 11:12:29' for ts in result['timestamp'].values)
+    assert all(ds == 3 for ds in result['data_size'].values)
+    assert all(im == 5 for im in result['interval_minutes'].values)
+    mock_instance.calculate.assert_called_once_with(['rmse', 'mse', 'mae', 'r2'])
 
 
 def test_calculate_simple_metrics_success_rmse_only(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -749,23 +771,23 @@ def test_calculate_simple_metrics_success_rmse_only(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [{'metric': 'rmse', 'value': 0.1}]
+    )
 
-    # Assert
-    assert len(result['metric']) == 1
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'rmse'
     assert result['model_id'].values[0] == 'test_model_id'
     assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
     assert result['data_size'].values[0] == 2
     assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['rmse']", metadata['metadata']
-    )
+    mock_instance.calculate.assert_called_once_with(['rmse'])
 
 
 def test_calculate_simple_metrics_success_mse_only(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -782,23 +804,23 @@ def test_calculate_simple_metrics_success_mse_only(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [{'metric': 'mse', 'value': 0.01}]
+    )
 
-    # Assert
-    assert len(result['metric']) == 1
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'mse'
     assert result['model_id'].values[0] == 'test_model_id'
     assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
     assert result['data_size'].values[0] == 2
     assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['mse']", metadata['metadata']
-    )
+    mock_instance.calculate.assert_called_once_with(['mse'])
 
 
 def test_calculate_simple_metrics_success_mae_only(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -815,23 +837,23 @@ def test_calculate_simple_metrics_success_mae_only(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [{'metric': 'mae', 'value': 0.1}]
+    )
 
-    # Assert
-    assert len(result['metric']) == 1
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'mae'
     assert result['model_id'].values[0] == 'test_model_id'
     assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
     assert result['data_size'].values[0] == 2
     assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['mae']", metadata['metadata']
-    )
+    mock_instance.calculate.assert_called_once_with(['mae'])
 
 
 def test_calculate_simple_metrics_success_r2_only(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -848,24 +870,24 @@ def test_calculate_simple_metrics_success_r2_only(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [{'metric': 'r2', 'value': 0.95}]
+    )
 
-    # Assert
-    assert len(result['metric']) == 1
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'r2'
     assert result['model_id'].values[0] == 'test_model_id'
     assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
     assert result['data_size'].values[0] == 2
     assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['r2']", metadata['metadata']
-    )
+    mock_instance.calculate.assert_called_once_with(['r2'])
 
 
-def test_calculate_simple_metrics_r2_zero_ss_tot(model_metrics_activity):
-    # Arrange
-    # All target values are the same, so ss_tot will be 0
+def test_calculate_simple_metrics_r2_zero_variance_delegates_to_lib(model_metrics_activity):
+    """r2 zero-variance is now the lib's responsibility. Activity just passes through."""
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -882,24 +904,18 @@ def test_calculate_simple_metrics_r2_zero_ss_tot(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
-
-    # Assert
-    assert len(result['metric']) == 1
-    assert result['metric'].values[0] == 'r2'
-    assert result['value'].values[0] == 0.0  # Should return 0.0 when ss_tot == 0
-    assert result['model_id'].values[0] == 'test_model_id'
-    assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
-    assert result['data_size'].values[0] == 2
-    assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['r2']", metadata['metadata']
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [{'metric': 'r2', 'value': 0.0}]
     )
 
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert result['value'].values[0] == 0.0
+    mock_instance.calculate.assert_called_once_with(['r2'])
+
 
 def test_calculate_simple_metrics_success_multiple_metrics_subset(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'],
@@ -916,23 +932,26 @@ def test_calculate_simple_metrics_success_multiple_metrics_subset(model_metrics_
         'interval_minutes': 5,
     }
 
-    # Act
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
-
-    # Assert
-    assert len(result['metric']) == 2
-    assert 'rmse' in result['metric'].values
-    assert 'mae' in result['metric'].values
-    assert all(model_id == 'test_model_id' for model_id in result['model_id'].values)
-    assert all(timestamp == '2023-05-26 11:12:29' for timestamp in result['timestamp'].values)
-    assert all(data_size == 3 for data_size in result['data_size'].values)
-    assert all(interval_minutes == 5 for interval_minutes in result['interval_minutes'].values)
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['rmse', 'mae']", metadata['metadata']
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'rmse', 'value': 0.1},
+            {'metric': 'mae', 'value': 0.1},
+        ]
     )
 
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
 
-def test_calculate_simple_metrics_unknown_metric_ignored(model_metrics_activity):
+    assert len(result) == 2
+    assert set(result['metric'].values) == {'rmse', 'mae'}
+    assert all(mid == 'test_model_id' for mid in result['model_id'].values)
+    assert all(ts == '2023-05-26 11:12:29' for ts in result['timestamp'].values)
+    assert all(ds == 3 for ds in result['data_size'].values)
+    assert all(im == 5 for im in result['interval_minutes'].values)
+    mock_instance.calculate.assert_called_once_with(['rmse', 'mae'])
+
+
+def test_calculate_simple_metrics_unknown_metric_raises(model_metrics_activity):
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -949,10 +968,218 @@ def test_calculate_simple_metrics_unknown_metric_ignored(model_metrics_activity)
         'interval_minutes': 5,
     }
 
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+    mock_instance = MagicMock()
+    mock_instance.calculate.side_effect = ValueError('Unknown metric: unknown_metric')
+    mock_class = MagicMock(return_value=mock_instance)
+    mock_class.is_r2_supported = MagicMock(return_value=True)
 
-    assert len(result['metric']) == 1
+    with patch('laborious.activities.model_metrics.RegressionMetrics', mock_class):
+        with raises(ValueError, match='Unknown metric'):
+            model_metrics_activity.calculate_simple_metrics(input_data)
+
+
+def test_calculate_simple_metrics_r2_skipped_for_nonlinear_model(model_metrics_activity):
+    """When model_type is non-linear, r2 is excluded and a warning notification fires."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['rmse', 'r2'],
+        'interval_minutes': 5,
+        'model_type': 'XGBoost',
+    }
+
+    mock_instance = MagicMock()
+    mock_instance.calculate.return_value = [{'metric': 'rmse', 'value': 0.1}]
+    mock_class = MagicMock(return_value=mock_instance)
+    mock_class.is_r2_supported = MagicMock(return_value=False)
+
+    with patch('laborious.activities.model_metrics.RegressionMetrics', mock_class):
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'rmse'
+    mock_class.is_r2_supported.assert_called_once_with('XGBoost')
+    mock_instance.calculate.assert_called_once_with(['rmse'])
+    model_metrics_activity.warning.assert_called_once()
+    model_metrics_activity.send_notification.assert_called_once()
+    call_kwargs = model_metrics_activity.send_notification.call_args.kwargs
+    assert call_kwargs['notification_id'] == 'SIMPLE_METRICS_R2_UNSUPPORTED'
+    assert call_kwargs['level'] == NotificationLevel.WARNING
+
+
+def test_calculate_simple_metrics_no_model_type_includes_r2(model_metrics_activity):
+    """When model_type is None (legacy input), r2 is included without check."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['r2'],
+        'interval_minutes': 5,
+        # no model_type key
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [{'metric': 'r2', 'value': 0.95}]
+    )
+
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert result['metric'].values[0] == 'r2'
+    mock_class.is_r2_supported.assert_not_called()
+
+
+def test_calculate_simple_metrics_threshold_breach_rmse(model_metrics_activity):
+    """When rmse exceeds rmse_max, a WARNING notification fires."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['rmse'],
+        'interval_minutes': 5,
+        'thresholds': {'rmse_max': 0.05},
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'rmse', 'value': 0.1},
+        ]
+    )
+
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
+    model_metrics_activity.send_notification.assert_called_once()
+    call_kwargs = model_metrics_activity.send_notification.call_args.kwargs
+    assert call_kwargs['notification_id'] == 'SIMPLE_METRICS_THRESHOLD_BREACH'
+    assert call_kwargs['level'] == NotificationLevel.WARNING
+    assert 'rmse' in call_kwargs['message']
+    assert '(window 5 min)' in call_kwargs['message']
+    # Model-level trigger so the orchestrator dedup merges windows (spec decision 9).
+    assert call_kwargs['metadata']['schedule_name'] == 'test_model_id-simple_metrics'
+    assert call_kwargs['metadata']['model_name'] == 'test_model'
+
+
+def test_calculate_simple_metrics_threshold_no_breach(model_metrics_activity):
+    """When rmse is below rmse_max, no notification fires."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['rmse'],
+        'interval_minutes': 5,
+        'thresholds': {'rmse_max': 1.0},
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'rmse', 'value': 0.1},
+        ]
+    )
+
+    with patcher:
+        model_metrics_activity.calculate_simple_metrics(input_data)
+
+    model_metrics_activity.send_notification.assert_not_called()
+
+
+def test_calculate_simple_metrics_threshold_r2_below_min(model_metrics_activity):
+    """When r2 drops below r2_min, a WARNING notification fires."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['r2'],
+        'interval_minutes': 5,
+        'thresholds': {'r2_min': 0.95},
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'r2', 'value': 0.8},
+        ]
+    )
+
+    with patcher:
+        model_metrics_activity.calculate_simple_metrics(input_data)
+
+    model_metrics_activity.send_notification.assert_called_once()
+    call_kwargs = model_metrics_activity.send_notification.call_args.kwargs
+    assert 'r2' in call_kwargs['message']
+
+
+def test_calculate_simple_metrics_no_thresholds_no_alert(model_metrics_activity):
+    """When thresholds is None (not configured), no alerting, no crash."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['rmse'],
+        'interval_minutes': 5,
+        # no thresholds key
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'rmse', 'value': 999.0},
+        ]
+    )
+
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
+    model_metrics_activity.send_notification.assert_not_called()
 
 
 @mark.parametrize(
@@ -1000,3 +1227,79 @@ def test_to_naive_utc_keeps_naive_timestamps():
 
     assert result.dt.tz is None
     assert result.tolist() == [Timestamp('2024-01-01 06:00:00')]
+
+
+@mark.parametrize(
+    'rows, thresholds, expected_count',
+    [
+        ([{'metric': 'rmse', 'value': 2.0}], {'rmse_max': 1.0}, 1),
+        ([{'metric': 'rmse', 'value': 0.5}], {'rmse_max': 1.0}, 0),
+        ([{'metric': 'r2', 'value': 0.5}], {'r2_min': 0.9}, 1),
+        ([{'metric': 'r2', 'value': 0.95}], {'r2_min': 0.9}, 0),
+        ([{'metric': 'rmse', 'value': 1.0}], {'rmse_max': 1.0}, 0),  # equal is not a breach
+        ([{'metric': 'r2', 'value': 0.9}], {'r2_min': 0.9}, 0),
+        ([{'metric': 'mae', 'value': 99.0}], {'rmse_max': 1.0}, 0),
+        ([{'metric': 'rmse', 'value': 99.0}], None, 0),
+        ([], {'rmse_max': 1.0}, 0),
+        (
+            [{'metric': 'rmse', 'value': 2.0}, {'metric': 'r2', 'value': 0.1}],
+            {'rmse_max': 1.0, 'r2_min': 0.9},
+            2,
+        ),
+    ],
+)
+def test_threshold_breaches(rows, thresholds, expected_count):
+    assert len(threshold_breaches(rows, thresholds)) == expected_count
+
+
+def _state(escalated: bool, streak: int) -> dict:
+    return {'escalated': escalated, 'healthy_streak': streak}
+
+
+@mark.parametrize(
+    'state, breached, runs, expected',
+    [
+        ({}, True, 3, _state(True, 0)),  # first breach escalates
+        ({}, False, 3, _state(False, 0)),  # missing state = healthy
+        (_state(False, 0), False, 3, _state(False, 0)),  # healthy stays healthy
+        (_state(True, 0), False, 3, _state(True, 1)),  # healthy run while escalated
+        (_state(True, 1), False, 3, _state(True, 2)),
+        (_state(True, 2), False, 3, _state(False, 0)),  # N-th healthy run recovers
+        (_state(True, 2), True, 3, _state(True, 0)),  # breach resets the streak
+        (_state(True, 0), False, 1, _state(False, 0)),  # N=1 recovers on first healthy run
+        (_state(True, None), False, 3, _state(True, 1)),  # hand-edited streak restarts
+        (_state(True, 'x'), False, 3, _state(True, 1)),
+        (_state(True, float('inf')), False, 3, _state(True, 1)),
+    ],
+)
+def test_next_adaptive_state(state, breached, runs, expected):
+    assert next_adaptive_state(state, breached, runs) == expected
+
+
+@mark.parametrize(
+    'frequency, seconds',
+    [('30s', 30), ('5m', 300), ('2h', 7200), ('1d', 86400)],
+)
+def test_frequency_to_seconds(frequency, seconds):
+    assert frequency_to_seconds(frequency) == seconds
+
+
+@mark.parametrize('frequency', ['', '5', 'm', '5x', 'abc', '0m'])
+def test_frequency_to_seconds_rejects_invalid(frequency):
+    with raises(ValueError):
+        frequency_to_seconds(frequency)
+
+
+@mark.parametrize(
+    'interval_minutes, frequency_alert, expected',
+    [
+        (60, '5m', 12),  # 60/5 = 12, at the cap
+        (60, '15m', 4),
+        (60, '7m', 9),  # ceil(8.57)
+        (60, '30s', 12),  # 120 capped at 12
+        (1440, '1h', 12),  # 24 capped at 12
+        (5, '1h', 1),  # window shorter than one run: at least 1
+    ],
+)
+def test_recovery_runs(interval_minutes, frequency_alert, expected):
+    assert recovery_runs(interval_minutes, frequency_alert) == expected

tests/laborious/activities/test_mongo.py

@@ -0,0 +1,357 @@
+from datetime import datetime
+from unittest.mock import MagicMock, call, patch
+
+from pytest import fixture, mark, raises
+from sientia_do.observability.metrics_controller import MetricsController
+from sientia_do.observability.sientia_monitoring import SientiaMonitoring
+
+from laborious.activities.mongo import MongoActivities
+
+
+@fixture
+def mongo() -> MagicMock:
+    return MagicMock()
+
+
+@fixture
+def prediction_metadata(mongo: MagicMock):
+    activity = MongoActivities(
+        mongo_repository=mongo,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.error = MagicMock()
+    return activity
+
+
+metadata = {
+    'metadata': {
+        'model_id': 'test_model',
+        'model_name': 'test_model',
+        'workflow_name': 'test_workflow',
+        'schema_name': 'test_schedule',
+    },
+}
+
+
+def test_write_prediction_metadata_no_documents_skips_mongo(prediction_metadata, mongo: MagicMock):
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': []})
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_missing_documents_key_skips_mongo(
+    prediction_metadata, mongo: MagicMock
+):
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata})
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_documents_with_no_filters_skips_mongo(
+    prediction_metadata, mongo: MagicMock
+):
+    # Arrange: a falsy document (None) and one with an empty `filters` list — neither contributes
+    # anything to write.
+    documents = [None, {'filters': []}]
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents})
+
+    # Assert
+    assert result is None
+    mongo.insert.assert_not_called()
+
+
+def test_write_prediction_metadata_not_configured_returns_none():
+    # Arrange
+    activity = MongoActivities(
+        mongo_repository=None,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.error = MagicMock()
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
+
+    # Act
+    result = activity.write_prediction_metadata({**metadata, 'documents': [document]})
+
+    # Assert
+    assert result is None
+    activity.error.assert_not_called()
+
+
+def test_write_prediction_metadata_success_generates_id_and_inserts(
+    prediction_metadata, mongo: MagicMock
+):
+    # Arrange
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 3}]}
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]})
+
+    # 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 == MongoActivities._PREDICTION_METADATA_COLLECTION
+    assert inserted_document == {'filters': document['filters'], '_id': result}
+    assert inserted_metadata == metadata['metadata']
+
+
+def test_write_prediction_metadata_combines_multiple_documents(
+    prediction_metadata, mongo: MagicMock
+):
+    # Arrange: two documents (e.g. from two different sources contributing metadata for the same
+    # prediction cycle) — their `filters` entries are flattened into one combined document.
+    documents = [
+        {'filters': [{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}]},
+        {
+            'filters': [
+                {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}
+            ]
+        },
+    ]
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': documents})
+
+    # Assert
+    assert result is not None
+    _, inserted_document, _ = mongo.insert.call_args.args
+    assert inserted_document['filters'] == [
+        {'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0},
+        {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2},
+    ]
+
+
+def test_write_prediction_metadata_insert_failure_swallowed(prediction_metadata, mongo: MagicMock):
+    # Arrange
+    mongo.insert.side_effect = RuntimeError('connection reset')
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
+
+    # Act
+    result = prediction_metadata.write_prediction_metadata({**metadata, 'documents': [document]})
+
+    # Assert: the prediction write must never be blocked or failed by this
+    assert result is None
+    prediction_metadata.error.assert_called_once()
+
+
+@patch.object(SientiaMonitoring, 'shutdown')
+def test_close_disposes_mongo_repository(mock_shutdown):
+    """``MongoActivities.close`` should close the Mongo client and clear the reference."""
+    mongo = MagicMock()
+    activity = MongoActivities(
+        mongo_repository=mongo,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.close()
+    mongo.close.assert_called_once()
+    assert activity.mongo_repository is None
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(activity)) == 1
+
+
+@patch.object(SientiaMonitoring, 'shutdown')
+def test_close_without_mongo_repository(mock_shutdown):
+    """When no Mongo repository is configured, ``close`` only shuts down monitoring."""
+    activity = MongoActivities(
+        mongo_repository=None,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.close()
+    # Other instances collected by GC during the test also hit the patched shutdown.
+    assert mock_shutdown.call_args_list.count(call(activity)) == 1
+
+
+_SM_METADATA = {**metadata['metadata'], 'schedule_name': 'm1-simple_metrics-60m'}
+_BREACH_INPUT = {
+    'metadata': _SM_METADATA,
+    'simple_metrics': [{'metric': 'rmse', 'value': 2.0}],
+    'thresholds': {'rmse_max': 1.0},
+    'interval_minutes': 60,
+}
+_HEALTHY_INPUT = {**_BREACH_INPUT, 'simple_metrics': [{'metric': 'rmse', 'value': 0.5}]}
+_EMPTY_INPUT = {**_BREACH_INPUT, 'simple_metrics': []}
+
+
+@fixture
+def mongo_activities(mongo: MagicMock):
+    activity = MongoActivities(
+        mongo_repository=mongo,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    activity.info = MagicMock()
+    activity.warning = MagicMock()
+    return activity
+
+
+def _pipeline(**fields) -> dict:
+    return {'schedule_name': 'm1-simple_metrics-60m', 'frequency': '1h', **fields}
+
+
+def _set_of(mongo: MagicMock) -> dict:
+    collection, query, update, _metadata = mongo.update_one.call_args.args
+    assert collection == 'pipelines'
+    assert query == {'schedule_name': 'm1-simple_metrics-60m'}
+    return update['$set']
+
+
+def test_adaptive_state_no_mongo_is_noop():
+    activity = MongoActivities(
+        mongo_repository=None,
+        logger=MagicMock(),
+        notification_handler=MagicMock(),
+        metrics_controller=MagicMock(spec=MetricsController),
+    )
+    assert activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
+
+
+def test_adaptive_state_missing_pipeline_doc_is_noop(mongo_activities, mongo):
+    mongo.find_one.return_value = None
+    assert mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
+    mongo.update_one.assert_not_called()
+    # Likely misconfig (other MONGODB_DATABASE_NAME or schedule_name mismatch): not silent.
+    mongo_activities.warning.assert_called_once()
+    assert 'm1-simple_metrics-60m' in mongo_activities.warning.call_args.args[0]
+
+
+def test_adaptive_state_without_frequency_alert_is_noop(mongo_activities, mongo):
+    mongo.find_one.return_value = _pipeline()
+    assert mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
+    mongo.update_one.assert_not_called()
+    mongo_activities.warning.assert_not_called()
+
+
+def test_adaptive_state_breach_escalates_and_bumps_updated_at(mongo_activities, mongo):
+    mongo.find_one.return_value = _pipeline(frequency_alert='5m')
+
+    result = mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT)
+
+    assert result == {'escalated': True, 'healthy_streak': 0}
+    update = _set_of(mongo)
+    assert update['adaptive_state'] == {'escalated': True, 'healthy_streak': 0}
+    # BSON datetime, never a string: the orchestrator calls .replace(tzinfo=...) on it.
+    assert isinstance(update['updated_at'], datetime)
+    assert update['updated_at'].tzinfo is not None
+
+
+def test_adaptive_state_healthy_run_increments_streak_without_updated_at(mongo_activities, mongo):
+    # 60 min window / 5m = N 12, so one healthy run only moves the streak.
+    mongo.find_one.return_value = _pipeline(
+        frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 0}
+    )
+
+    result = mongo_activities.update_simple_metrics_adaptive_state(_HEALTHY_INPUT)
+
+    assert result == {'escalated': True, 'healthy_streak': 1}
+    assert _set_of(mongo) == {'adaptive_state': {'escalated': True, 'healthy_streak': 1}}
+
+
+def test_adaptive_state_recovers_after_window_renewed(mongo_activities, mongo):
+    # 60 min window / 30m = N 2: second consecutive healthy run recovers.
+    mongo.find_one.return_value = _pipeline(
+        frequency_alert='30m', adaptive_state={'escalated': True, 'healthy_streak': 1}
+    )
+
+    result = mongo_activities.update_simple_metrics_adaptive_state(_HEALTHY_INPUT)
+
+    assert result == {'escalated': False, 'healthy_streak': 0}
+    update = _set_of(mongo)
+    assert update['adaptive_state'] == {'escalated': False, 'healthy_streak': 0}
+    assert isinstance(update['updated_at'], datetime)
+
+
+def test_adaptive_state_empty_window_counts_as_healthy_run(mongo_activities, mongo):
+    # Spec decision 8: an empty/under-sampled window never escalates but counts toward recovery.
+    mongo.find_one.return_value = _pipeline(
+        frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 3}
+    )
+
+    result = mongo_activities.update_simple_metrics_adaptive_state(_EMPTY_INPUT)
+
+    assert result == {'escalated': True, 'healthy_streak': 4}
+
+
+def test_adaptive_state_empty_window_never_escalates(mongo_activities, mongo):
+    mongo.find_one.return_value = _pipeline(frequency_alert='5m')
+
+    result = mongo_activities.update_simple_metrics_adaptive_state(_EMPTY_INPUT)
+
+    assert result == {'escalated': False, 'healthy_streak': 0}
+    mongo.update_one.assert_not_called()
+
+
+def test_adaptive_state_unchanged_does_not_write(mongo_activities, mongo):
+    mongo.find_one.return_value = _pipeline(
+        frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 0}
+    )
+
+    result = mongo_activities.update_simple_metrics_adaptive_state(_HEALTHY_INPUT)
+
+    assert result == {'escalated': False, 'healthy_streak': 0}
+    mongo.update_one.assert_not_called()
+
+
+@mark.parametrize('frequency_alert', ['5min', '0m', 5])
+def test_adaptive_state_invalid_frequency_alert_is_skipped(
+    mongo_activities, mongo, frequency_alert
+):
+    mongo.find_one.return_value = _pipeline(frequency_alert=frequency_alert)
+
+    assert mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None
+    mongo.update_one.assert_not_called()
+    mongo_activities.warning.assert_called_once()
+
+
+@mark.parametrize('interval_minutes', ['60', None])
+def test_adaptive_state_invalid_interval_minutes_is_skipped(
+    mongo_activities, mongo, interval_minutes
+):
+    # Untyped Mongo config: "60" * 60 is a string, None * 60 a TypeError.
+    mongo.find_one.return_value = _pipeline(frequency_alert='5m')
+
+    result = mongo_activities.update_simple_metrics_adaptive_state(
+        {**_BREACH_INPUT, 'interval_minutes': interval_minutes}
+    )
+
+    assert result is None
+    mongo.update_one.assert_not_called()
+    mongo_activities.warning.assert_called_once()
+
+
+@mark.parametrize('adaptive_state', ['garbage', ['escalated']])
+def test_adaptive_state_malformed_state_is_not_escalated(mongo_activities, mongo, adaptive_state):
+    mongo.find_one.return_value = _pipeline(frequency_alert='5m', adaptive_state=adaptive_state)
+
+    result = mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT)
+
+    assert result == {'escalated': True, 'healthy_streak': 0}
+    mongo.update_one.assert_called_once()
+
+
+@mark.parametrize('failing_call', ['find_one', 'update_one'])
+def test_adaptive_state_mongo_error_propagates(mongo_activities, mongo, failing_call):
+    mongo.find_one.return_value = _pipeline(frequency_alert='5m')
+    getattr(mongo, failing_call).side_effect = RuntimeError('mongo down')
+
+    with raises(RuntimeError, match='mongo down'):
+        mongo_activities.update_simple_metrics_adaptive_state(_BREACH_INPUT)

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/worker/test_worker.py

@@ -16,6 +16,7 @@ def _build_fake_activities():
     inst.export_data_to_postgres = MagicMock()
     inst.load_custom_query = MagicMock()
     inst.calculate_simple_metrics = MagicMock()
+    inst.update_simple_metrics_adaptive_state = MagicMock()
     inst.get_reference_data = MagicMock()
     inst.calculate_drift = MagicMock()
     inst.request_predict = MagicMock()
@@ -188,6 +189,11 @@ async def test_main_success_exit_zero(monkeypatch):
     # Drift and SimpleMetrics load no model code, so every runtime shares the `core` queues.
     assert prepare_calls[1].kwargs['runtime'] == 'core'
     assert prepare_calls[2].kwargs['runtime'] == 'core'
+    # prepare_calls[1] is the SimpleMetrics worker (order: MinimalRetrain, SimpleMetrics, Drift, PredictionsBatch).
+    assert (
+        fake_activities.update_simple_metrics_adaptive_state
+        in prepare_calls[1].kwargs['activities']
+    )
 
 
 @mark.asyncio

tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py

@@ -326,6 +326,7 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
                     'model_id': input_data['model_id'],
                     'prediction_confidence': input_data['prediction_confidence'],
                     'comment': input_data['comment'],
+                    'metadata_id': None,
                 },
                 retry_policy=ANY,
                 start_to_close_timeout=ANY,
@@ -389,6 +390,56 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
     assert workflow_mock.execute_local_activity_method.call_count == 1
 
 
+@mark.asyncio
+@patch(
+    'laborious.workflows.sub_workflows.format_and_export_prediction.workflow',
+    new_callable=AsyncMock,
+)
+async def test_run_default_path_flag_passes_metadata_id(
+    workflow_mock, format_and_export_prediction
+):
+    # Arrange: metadata_id reaches format_default_prediction, which is the row that ends up in
+    # Postgres via export_data_to_postgres (CONTINUE/REPEAT path).
+    input_data = {
+        'metadata': metadata,
+        'path_flag': 'default',
+        'data': {'test': 'data'},
+        'timestamp': '2021-01-01',
+        'model_id': 1,
+        'model_name': metadata['metadata']['model_name'],
+        'prediction_confidence': 0,
+        'schema': 'test_schema',
+        'table_name': 'test_table',
+        'opc_output_config': {'test': 'config'},
+        'comment': 'test_comment',
+        'metadata_id': 'a' * 24,
+    }
+
+    workflow_mock.execute_activity_method.side_effect = [
+        (MagicMock(), MagicMock()),
+        MagicMock(),
+        MagicMock(),
+    ]
+
+    # Act
+    await format_and_export_prediction.run(input_data)
+
+    # Assert
+    workflow_mock.execute_local_activity_method.assert_any_call(
+        Activities.format_default_prediction,
+        {
+            **metadata,
+            'timestamp': input_data['timestamp'],
+            'model_id': input_data['model_id'],
+            'prediction_confidence': input_data['prediction_confidence'],
+            'comment': input_data['comment'],
+            'metadata_id': 'a' * 24,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch(
     'laborious.workflows.sub_workflows.format_and_export_prediction.workflow',
@@ -638,6 +689,20 @@ async def test_run_default_path_flag_with_pi_web_api(workflow_mock, format_and_e
 
     await format_and_export_prediction.run(input_data)
 
+    workflow_mock.execute_local_activity_method.assert_any_call(
+        Activities.format_default_prediction,
+        {
+            **metadata,
+            'timestamp': input_data['timestamp'],
+            'model_id': input_data['model_id'],
+            'prediction_confidence': input_data['prediction_confidence'],
+            'comment': input_data['comment'],
+            'metadata_id': None,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
     workflow_mock.execute_activity_method.assert_has_calls(
         [
             call(

tests/laborious/workflows/subworkflows/test_prediction_process.py

@@ -57,7 +57,7 @@ async def test_run(workflow_mock, prediction_process):
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'Input data with bad quality'),  # input_gate
+        ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
         ('continue', 0.95, 'Error'),
         # mlflow_content_gate (transform)
@@ -217,7 +217,7 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process):
 
     # Mock the activity responses
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('stop', 0.95, 'Input data with bad quality'),  # input_gate
+        ('stop', 0.95, 'Input data with bad quality', None),  # input_gate
     ]
 
     # Act
@@ -275,7 +275,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('repeat', 0.95, 'Input data with bad quality'),  # input_gate
+        ('repeat', 0.95, 'Input data with bad quality', None),  # input_gate
         ('continue', 0.95, 'Error'),  # mlflow_response_gate (transform)
     ]
 
@@ -365,7 +365,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'Input data with bad quality'),  # input_gate
+        ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
         ('continue', 0.95, 'Error'),
         # mlflow_content_gate (transform)
@@ -476,7 +476,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'Input data with bad quality'),  # input_gate
+        ('continue', 0.95, 'Input data with bad quality', None),  # input_gate
         # mlflow_response_gate (transform)
         ('continue', 0.95, 'Error'),
         # mlflow_content_gate (transform)
@@ -598,6 +598,7 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process):
     last_timestamp = '2024-01-01'
     model_name = 'test_model_name'
     model_config = {'retention': '30'}
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'STOP', 'counter': 0}]}
 
     # Act
     result = await prediction_process.path_flag_handler(
@@ -616,18 +617,22 @@ async def test_path_flag_handler_stop(workflow_mock, prediction_process):
         confidence,
         last_timestamp,
         '',
+        document,
     )
 
-    # Assert
+    # Assert: STOP never reaches a write, even though it now gets a `document` same as CONTINUE
+    # and REPEAT do — it just never has anywhere to use it.
     assert result is True
     workflow_mock.execute_local_activity_method.assert_not_called()
+    workflow_mock.execute_activity_method.assert_not_called()
     workflow_mock.execute_child_workflow.assert_not_called()
 
 
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
-    # Arrange
+    # Arrange: REPEAT now writes metadata just like CONTINUE — it does insert a new `predictions`
+    # row (reusing the previous row's values), so it needs a `metadata_id` too.
     data = {'test': 'data'}
     path_flag = 'repeat'
     confidence = 0.95
@@ -637,6 +642,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
     last_timestamp = '2024-01-01'
     model_name = 'test_model_name'
     model_config = {'retention': '30'}
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]}
 
     # Act
     result = await prediction_process.path_flag_handler(
@@ -655,11 +661,20 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
         confidence,
         last_timestamp,
         '',
+        document,
     )
 
     # Assert
     assert result is True
-    workflow_mock.execute_activity_method.assert_called_once_with(
+    metadata_id = workflow_mock.execute_activity_method.return_value
+    assert workflow_mock.execute_activity_method.call_count == 2
+    workflow_mock.execute_activity_method.assert_any_call(
+        Activities.write_prediction_metadata,
+        {**metadata, 'documents': [document]},
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+    workflow_mock.execute_activity_method.assert_any_call(
         Activities.repeat_last_prediction,
         {
             **metadata,
@@ -667,6 +682,7 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
             'table_name': table_name,
             'model': model,
             'last_timestamp': last_timestamp,
+            'metadata_id': metadata_id,
         },
         retry_policy=ANY,
         start_to_close_timeout=ANY,
@@ -674,6 +690,56 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process):
     workflow_mock.execute_child_workflow.assert_not_called()
 
 
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+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).
+    data = {'test': 'data'}
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        data,
+        'repeat',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+        },
+        0.95,
+        '2024-01-01',
+        '',
+    )
+
+    # Assert
+    assert result is True
+    called_activities = [c.args[0] for c in workflow_mock.execute_activity_method.call_args_list]
+    assert Activities.write_prediction_metadata not in called_activities
+    # repeat_last_prediction still happens, just with metadata_id=None.
+    workflow_mock.execute_activity_method.assert_any_call(
+        Activities.repeat_last_prediction,
+        {
+            **metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'model': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'metadata_id': None,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_path_flag_handler_continue(workflow_mock, prediction_process):
@@ -688,6 +754,9 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
     model_name = 'test_model_name'
     model_config = {'retention': '30'}
     prediction_store_policy = 'erl:1'
+    document = {
+        'filters': [{'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2}]
+    }
 
     # Act
     result = await prediction_process.path_flag_handler(
@@ -713,11 +782,18 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
         confidence,
         last_timestamp,
         'Prediction Process',
+        document,
     )
 
     # Assert
+    metadata_id = workflow_mock.execute_activity_method.return_value
     assert result is True
-    workflow_mock.execute_activity_method.assert_not_called()
+    workflow_mock.execute_activity_method.assert_called_once_with(
+        Activities.write_prediction_metadata,
+        {**metadata, 'documents': [document]},
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
     workflow_mock.execute_child_workflow.assert_called_once_with(
         'subworkflow.format_and_export_prediction',
         {
@@ -733,6 +809,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
             'table_name': table_name,
             'transform_table_name': 'test_transform_table',
             'comment': 'Prediction Process',
+            'metadata_id': metadata_id,
             'opc_output_config': {'test': 'config'},
             'pi_web_api_output_config': {
                 'endpoint': 'https://test-pi-server.com',
@@ -745,6 +822,153 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process):
     )
 
 
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_continue_no_document_skips_write_activity_call(
+    workflow_mock, prediction_process
+):
+    # Arrange: same gap as the REPEAT case above (review finding, 2026-10-01) — no document must
+    # not dispatch the remote write_prediction_metadata activity.
+    data = {'test': 'data'}
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        data,
+        'continue',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+            'opc_output_config': {'test': 'config'},
+            'pi_web_api_output_config': {
+                'endpoint': 'https://test-pi-server.com',
+                'prediction_tags': {},
+                'confidence_tags': {},
+            },
+            'prediction_store_policy': 'erl:1',
+        },
+        0.95,
+        '2024-01-01',
+        '',
+    )
+
+    # Assert
+    assert result is True
+    workflow_mock.execute_activity_method.assert_not_called()
+    workflow_mock.execute_child_workflow.assert_called_once_with(
+        'subworkflow.format_and_export_prediction',
+        ANY,
+    )
+    _, export_input = workflow_mock.execute_child_workflow.call_args.args
+    assert export_input['metadata_id'] is None
+
+
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_continue_write_metadata_failure_does_not_block_prediction(
+    workflow_mock, prediction_process
+):
+    # Arrange: write_prediction_metadata's retry_policy has maximum_attempts=1 — a timeout, an
+    # unavailable worker, or any dispatch failure raises straight into the workflow with no
+    # retry. Business rule 1 says this must never block the prediction write (review finding,
+    # 2026-10-01) — confirm the CONTINUE branch still proceeds instead of letting it propagate.
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'CONTINUE', 'counter': 0}]}
+    workflow_mock.execute_activity_method.side_effect = RuntimeError('activity dispatch failed')
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        {'test': 'data'},
+        'continue',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+            'opc_output_config': {'test': 'config'},
+            'pi_web_api_output_config': {
+                'endpoint': 'https://test-pi-server.com',
+                'prediction_tags': {},
+                'confidence_tags': {},
+            },
+            'prediction_store_policy': 'erl:1',
+        },
+        0.95,
+        '2024-01-01',
+        '',
+        document,
+    )
+
+    # Assert: the prediction write still happens, with metadata_id=None, instead of raising.
+    assert result is True
+    workflow_mock.execute_child_workflow.assert_called_once_with(
+        'subworkflow.format_and_export_prediction',
+        ANY,
+    )
+    _, export_input = workflow_mock.execute_child_workflow.call_args.args
+    assert export_input['metadata_id'] is None
+
+
+@mark.asyncio
+@patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
+async def test_path_flag_handler_repeat_write_metadata_failure_does_not_block_prediction(
+    workflow_mock, prediction_process
+):
+    # Arrange: only the write_prediction_metadata call (first) fails — repeat_last_prediction
+    # (second) must still run normally; its own failure is a separate, intentional
+    # non-best-effort path, not what this test is about.
+    document = {'filters': [{'name': 'EMPTY_DATA', 'policy': 'REPEAT', 'counter': 0}]}
+    workflow_mock.execute_activity_method.side_effect = [
+        RuntimeError('activity dispatch failed'),
+        MagicMock(),
+    ]
+
+    # Act
+    result = await prediction_process.path_flag_handler(
+        {'test': 'data'},
+        'repeat',
+        {
+            'metadata': metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'transform_table_name': 'test_transform_table',
+            'model_id': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'model_name': 'test_model_name',
+            'model_config': {'retention': '30'},
+        },
+        0.95,
+        '2024-01-01',
+        '',
+        document,
+    )
+
+    # Assert: repeat_last_prediction still runs, with metadata_id=None, instead of raising.
+    assert result is True
+    assert workflow_mock.execute_activity_method.call_count == 2
+    workflow_mock.execute_activity_method.assert_any_call(
+        Activities.repeat_last_prediction,
+        {
+            **metadata,
+            'schema': 'test_schema',
+            'table_name': 'test_table',
+            'model': 'test_model',
+            'last_timestamp': '2024-01-01',
+            'metadata_id': None,
+        },
+        retry_policy=ANY,
+        start_to_close_timeout=ANY,
+    )
+
+
 @mark.asyncio
 @patch('laborious.workflows.sub_workflows.prediction_process.workflow', new_callable=AsyncMock)
 async def test_path_flag_handler_unknown(workflow_mock, prediction_process):
@@ -850,7 +1074,7 @@ async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process):
         MagicMock(),
     ]
     workflow_mock.execute_local_activity_method.side_effect = [
-        ('continue', 0.95, 'ok'),
+        ('continue', 0.95, 'ok', None),  # input_gate
         ('continue', 0.95, ''),
         ('continue', 0.95, ''),
         ('continue', 0.95, ''),

tests/laborious/workflows/test_simple_metrics.py

@@ -1,6 +1,6 @@
-from unittest.mock import ANY, AsyncMock, call, patch
+from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
-from pytest import fixture, mark
+from pytest import fixture, mark, raises
 from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
 
 from laborious.activities.activities import Activities
@@ -42,12 +42,17 @@ async def test_run(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics):
     target_data = {'data': 'test_target_data'}
     simple_metrics_data = {'metrics': 'test_simple_metrics_data'}
 
-    workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None])
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None, None])
     workflow_mock.execute_local_activity_method = AsyncMock(return_value=simple_metrics_data)
 
     # Act
     await simple_metrics.run(input_data)
 
+    # Assert - load, export and adaptive state (no thresholds still updates it)
+    assert workflow_mock.execute_activity_method.call_count == 3
+    workflow_mock.logger.warning.assert_not_called()
+
     # Assert - Check load_custom_query call
     # Query format matches psycopg2.sql output (identifiers with double quotes, literals with single quotes)
     expected_query = f"""
@@ -103,6 +108,8 @@ async def test_run(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics):
             'target_data': target_data,
             'metrics': input_data['metrics'],
             'interval_minutes': input_data['interval_minutes'],
+            'model_type': None,
+            'thresholds': None,
         },
         retry_policy=ANY,
         start_to_close_timeout=ANY,
@@ -128,13 +135,22 @@ async def test_run_empty_target_data(workflow_mock: AsyncMock, simple_metrics: S
 
     target_data = None
 
+    workflow_mock.logger = MagicMock()
     workflow_mock.execute_activity_method = AsyncMock(return_value=target_data)
+    workflow_mock.execute_local_activity_method = AsyncMock()
 
     # Act
     await simple_metrics.run(input_data)
 
-    # Assert - Should not call calculate_simple_metrics or export
-    assert workflow_mock.execute_activity_method.call_count == 1
+    # Assert - No calculate or export; load + adaptive state (empty window = healthy run)
+    workflow_mock.execute_local_activity_method.assert_not_called()
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert [c.args[0] for c in calls] == [
+        Activities.load_custom_query,
+        Activities.update_simple_metrics_adaptive_state,
+    ]
+    assert calls[1].args[1]['simple_metrics'] == []
+    workflow_mock.logger.warning.assert_not_called()
 
 
 @mark.asyncio
@@ -157,15 +173,22 @@ async def test_run_empty_simple_metrics(workflow_mock: AsyncMock, simple_metrics
     target_data = {'data': 'test_target_data'}
     simple_metrics_data = None
 
+    workflow_mock.logger = MagicMock()
     workflow_mock.execute_activity_method = AsyncMock(return_value=target_data)
     workflow_mock.execute_local_activity_method = AsyncMock(return_value=simple_metrics_data)
 
     # Act
     await simple_metrics.run(input_data)
 
-    # Assert - Should call calculate_simple_metrics but not export
-    workflow_mock.execute_activity_method.assert_called_once()
+    # Assert - calculate returns None: no export, adaptive state gets [] (the `or []` is live)
     workflow_mock.execute_local_activity_method.assert_called_once()
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert [c.args[0] for c in calls] == [
+        Activities.load_custom_query,
+        Activities.update_simple_metrics_adaptive_state,
+    ]
+    assert calls[1].args[1]['simple_metrics'] == []
+    workflow_mock.logger.warning.assert_not_called()
 
 
 @mark.asyncio
@@ -188,12 +211,17 @@ async def test_run_default_metrics(workflow_mock: AsyncMock, simple_metrics: Sim
     target_data = {'data': 'test_target_data'}
     simple_metrics_data = {'metrics': 'test_simple_metrics_data'}
 
-    workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None])
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[target_data, None, None])
     workflow_mock.execute_local_activity_method = AsyncMock(return_value=simple_metrics_data)
 
     # Act
     await simple_metrics.run(input_data)
 
+    # Assert - all three activities ran
+    assert workflow_mock.execute_activity_method.call_count == 3
+    workflow_mock.logger.warning.assert_not_called()
+
     # Assert - Check calculate_simple_metrics call with default metrics
     workflow_mock.execute_activity_method.assert_any_call(
         Activities.load_custom_query,
@@ -209,7 +237,182 @@ async def test_run_default_metrics(workflow_mock: AsyncMock, simple_metrics: Sim
             'target_data': target_data,
             'metrics': ['rmse', 'mse', 'mae', 'r2'],  # Default value
             'interval_minutes': input_data['interval_minutes'],
+            'model_type': None,
+            'thresholds': None,
         },
         retry_policy=ANY,
         start_to_close_timeout=ANY,
     )
+
+
+def _input(**model_config) -> dict:
+    return {
+        'schedule_name': 'test_schedule',
+        'model_name': 'test_model',
+        'model_id': 'test_model_id',
+        'interval_minutes': 60,
+        'model_config': {'target': 'test_target', **model_config},
+        'schema': 'test_schema',
+        'predictions_table_name': 'test_predictions_table',
+        'data_table_name': 'test_data_table',
+        'target_table_name': 'test_target_table',
+        'metrics': ['rmse'],
+    }
+
+
+_ROW = {'timestamp': '2026-10-06 10:00:00+0000', 'prediction': 1.0, 'target': 2.0}
+
+
+_METRICS = [{'metric': 'rmse', 'value': 9.0}]
+
+
+@mark.asyncio
+@mark.parametrize('n_rows', [2, 4])
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_below_min_samples_drops_thresholds(workflow_mock, simple_metrics, n_rows):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * n_rows, None, None])
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_METRICS)
+
+    await simple_metrics.run(
+        _input(simple_metrics_thresholds={'rmse_max': 1.0}, simple_metrics_min_samples=5)
+    )
+
+    activity_input = workflow_mock.execute_local_activity_method.call_args.args[1]
+    assert activity_input['thresholds'] is None
+    # Persist, don't alert: load, export, adaptive state.
+    assert workflow_mock.execute_activity_method.call_count == 3
+    export_call = workflow_mock.execute_activity_method.call_args_list[1]
+    assert export_call.args[0] == Activities.export_data_to_postgres
+    assert export_call.args[1]['data'] == _METRICS
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_at_min_samples_keeps_thresholds(workflow_mock, simple_metrics):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 5, None, None])
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_METRICS)
+
+    await simple_metrics.run(
+        _input(simple_metrics_thresholds={'rmse_max': 1.0}, simple_metrics_min_samples=5)
+    )
+
+    activity_input = workflow_mock.execute_local_activity_method.call_args.args[1]
+    assert activity_input['thresholds'] == {'rmse_max': 1.0}
+    assert workflow_mock.execute_activity_method.call_count == 3
+
+
+_BREACH_ROWS = [{'metric': 'rmse', 'value': 9.0}]
+_THRESHOLDS = {'rmse_max': 1.0}
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_updates_adaptive_state_after_export(workflow_mock, simple_metrics):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(
+        side_effect=[[_ROW] * 3, None, {'escalated': True, 'healthy_streak': 0}]
+    )
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
+
+    await simple_metrics.run(_input(simple_metrics_thresholds=_THRESHOLDS))
+
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert calls[1].args[0] is Activities.export_data_to_postgres
+    assert calls[2].args[0] is Activities.update_simple_metrics_adaptive_state
+    assert calls[2].args[1] == {
+        **metadata,
+        'simple_metrics': _BREACH_ROWS,
+        'thresholds': _THRESHOLDS,
+        'interval_minutes': 60,
+    }
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_without_thresholds_still_updates_adaptive_state(workflow_mock, simple_metrics):
+    # No thresholds = never a breach; still calling lets an escalated doc recover after the
+    # thresholds are removed.
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 3, None, None])
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
+
+    await simple_metrics.run(_input())
+
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert len(calls) == 3
+    assert calls[2].args[0] is Activities.update_simple_metrics_adaptive_state
+    assert calls[2].args[1]['thresholds'] is None
+    assert calls[2].args[1]['simple_metrics'] == _BREACH_ROWS
+    workflow_mock.logger.warning.assert_not_called()
+
+
+@mark.asyncio
+@mark.parametrize(
+    ('min_samples', 'warns'), [(None, False), (0, False), ('5', True), (5.0, True), (True, True)]
+)
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_guard_off_keeps_thresholds(workflow_mock, simple_metrics, min_samples, warns):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 2, None, None])
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_METRICS)
+
+    await simple_metrics.run(
+        _input(simple_metrics_thresholds={'rmse_max': 1.0}, simple_metrics_min_samples=min_samples)
+    )
+
+    activity_input = workflow_mock.execute_local_activity_method.call_args.args[1]
+    assert activity_input['thresholds'] == {'rmse_max': 1.0}
+    assert workflow_mock.logger.warning.call_count == int(warns)
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_empty_window_counts_as_healthy_run(workflow_mock, simple_metrics):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[], None])
+    workflow_mock.execute_local_activity_method = AsyncMock()
+
+    await simple_metrics.run(_input(simple_metrics_thresholds=_THRESHOLDS))
+
+    workflow_mock.execute_local_activity_method.assert_not_called()
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert len(calls) == 2  # load + adaptive state, no export
+    assert calls[1].args[0] is Activities.update_simple_metrics_adaptive_state
+    assert calls[1].args[1]['simple_metrics'] == []
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_below_min_samples_sends_no_rows_to_adaptive_state(workflow_mock, simple_metrics):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 2, None, None])
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
+
+    await simple_metrics.run(
+        _input(simple_metrics_thresholds=_THRESHOLDS, simple_metrics_min_samples=5)
+    )
+
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert calls[1].args[0] is Activities.export_data_to_postgres  # metrics still persisted
+    assert calls[2].args[0] is Activities.update_simple_metrics_adaptive_state
+    assert calls[2].args[1]['simple_metrics'] == []  # never a breach
+
+
+@mark.asyncio
+@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock)
+async def test_run_adaptive_state_failure_fails_workflow(workflow_mock, simple_metrics):
+    workflow_mock.logger = MagicMock()
+    workflow_mock.execute_activity_method = AsyncMock(
+        side_effect=[[_ROW] * 3, None, RuntimeError('dispatch failed')]
+    )
+    workflow_mock.execute_local_activity_method = AsyncMock(return_value=_BREACH_ROWS)
+    input_data = _input(simple_metrics_thresholds=_THRESHOLDS)
+
+    with raises(RuntimeError, match='dispatch failed'):
+        await simple_metrics.run(input_data)
+
+    # Metrics were exported before the failing call
+    calls = workflow_mock.execute_activity_method.call_args_list
+    assert calls[1].args[0] is Activities.export_data_to_postgres