feat(simple_metrics): write adaptive_state to the pipelines doc (SIENTIAPDE-2191)

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