From 34f56c7416c89b441b00f030bf70750191a40700 Mon Sep 17 00:00:00 2001 From: PedroHMCosme Date: Wed, 7 Oct 2026 15:45:38 -0300 Subject: [PATCH] feat(simple_metrics): write adaptive_state to the pipelines doc (SIENTIAPDE-2191) --- laborious/activities/model_metrics.py | 66 ++++++++- .../activities/test_model_metrics.py | 133 ++++++++++++++++++ 2 files changed, 198 insertions(+), 1 deletion(-) diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py index d2271b9..2aa8fa6 100644 --- a/laborious/activities/model_metrics.py +++ b/laborious/activities/model_metrics.py @@ -14,7 +14,8 @@ with workflow.unsafe.imports_passed_through(): 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.temporal.constants import DATETIME_FORMAT_WITH_TZ + from sientia_do.repository.mongodb_repository_sync import MongoDBRepository + from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, now from sientia_model.analytics.drift_analysis import DriftAnalysis, DriftInsufficientDataError from sientia_model.metrics.regression import RegressionMetrics @@ -103,6 +104,12 @@ class ModelMetrics(SientiaMonitoring): _MAX_DEBUG_DATAFRAME_ROWS = 100 + # ponytail: same declaration as PredictionMetadata; the Activities composite sets the + # instance value in PredictionMetadata.__init__. A standalone ModelMetrics (unit tests) keeps + # None and the adaptive-state activity becomes a no-op. + mongo_repository: MongoDBRepository | None = None + _PIPELINES_COLLECTION = 'pipelines' + def __init__( self, logger: Logger, @@ -513,3 +520,60 @@ class ModelMetrics(SientiaMonitoring): ) return data.to_dict(orient='records') + + @activity.defn(name='update_simple_metrics_adaptive_state') + def update_simple_metrics_adaptive_state(self, input_data: dict[str, Any]) -> dict | None: + """ + Best-effort update of 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`. + - 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, or the + read/parse/write failed. + """ + metadata = input_data['metadata'] + if self.mongo_repository is None: + return None + + query = {'schedule_name': metadata['schedule_name']} + try: + pipeline = self.mongo_repository.find_one(self._PIPELINES_COLLECTION, query, metadata) + if not pipeline or not pipeline.get('frequency_alert'): + return None + + runs = recovery_runs(input_data['interval_minutes'], pipeline['frequency_alert']) + breached = bool( + threshold_breaches(input_data['simple_metrics'], input_data.get('thresholds')) + ) + state = pipeline.get('adaptive_state') or {'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 + ) + except Exception as e: + self.error(f'Failed to update simple metrics adaptive state: {e}', metadata) + return None + + self.info(f'Adaptive state of {query["schedule_name"]}: {state} -> {new_state}', metadata) + return new_state diff --git a/tests/laborious/activities/test_model_metrics.py b/tests/laborious/activities/test_model_metrics.py index c6490ad..a6a2919 100644 --- a/tests/laborious/activities/test_model_metrics.py +++ b/tests/laborious/activities/test_model_metrics.py @@ -1,3 +1,4 @@ +from datetime import datetime from unittest.mock import ANY, MagicMock, patch from pandas import DataFrame, Series, Timestamp @@ -1300,3 +1301,135 @@ def test_frequency_to_seconds_rejects_invalid(frequency): ) def test_recovery_runs(interval_minutes, frequency_alert, expected): assert recovery_runs(interval_minutes, frequency_alert) == expected + + +_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(model_metrics_activity): + repository = MagicMock() + model_metrics_activity.mongo_repository = repository + return repository + + +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(model_metrics_activity): + assert model_metrics_activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None + + +def test_adaptive_state_missing_pipeline_doc_is_noop(model_metrics_activity, mongo): + mongo.find_one.return_value = None + assert model_metrics_activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None + mongo.update_one.assert_not_called() + + +def test_adaptive_state_without_frequency_alert_is_noop(model_metrics_activity, mongo): + mongo.find_one.return_value = _pipeline() + assert model_metrics_activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None + mongo.update_one.assert_not_called() + + +def test_adaptive_state_breach_escalates_and_bumps_updated_at(model_metrics_activity, mongo): + mongo.find_one.return_value = _pipeline(frequency_alert='5m') + + result = model_metrics_activity.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( + model_metrics_activity, 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 = model_metrics_activity.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(model_metrics_activity, 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 = model_metrics_activity.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(model_metrics_activity, 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 = model_metrics_activity.update_simple_metrics_adaptive_state(_EMPTY_INPUT) + + assert result == {'escalated': True, 'healthy_streak': 4} + + +def test_adaptive_state_empty_window_never_escalates(model_metrics_activity, mongo): + mongo.find_one.return_value = _pipeline(frequency_alert='5m') + + result = model_metrics_activity.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(model_metrics_activity, mongo): + mongo.find_one.return_value = _pipeline( + frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 0} + ) + + result = model_metrics_activity.update_simple_metrics_adaptive_state(_HEALTHY_INPUT) + + assert result == {'escalated': False, 'healthy_streak': 0} + mongo.update_one.assert_not_called() + + +def test_adaptive_state_invalid_frequency_alert_is_swallowed(model_metrics_activity, mongo): + mongo.find_one.return_value = _pipeline(frequency_alert='5min') + + assert model_metrics_activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None + mongo.update_one.assert_not_called() + model_metrics_activity.error.assert_called_once() + + +def test_adaptive_state_mongo_error_is_swallowed(model_metrics_activity, mongo): + mongo.find_one.side_effect = RuntimeError('mongo down') + + assert model_metrics_activity.update_simple_metrics_adaptive_state(_BREACH_INPUT) is None + model_metrics_activity.error.assert_called_once()