feat(simple_metrics): use frequency_alert while adaptive_state is escalated (SIENTIAPDE-2190)

orchestrator/utils/orchestrator_functions.py

@@ -85,6 +85,9 @@ def simple_metrics(config: dict[str, Any]):
             - interval_minutes (int, optional): Computation interval in minutes (default: 60)
             - metrics (list[str], optional): List of metrics to compute
                 (default: ['rmse', 'mse', 'mae', 'r2'])
+            - frequency_alert (str, optional): Frequency used while `adaptive_state.escalated`
+                is true; same format as `frequency`. Absent = adaptive frequency off.
+            - adaptive_state (dict, optional): `{escalated, healthy_streak}`, written by laborious.
             - Additional fields from common_config
 
     Returns:
@@ -92,14 +95,23 @@ def simple_metrics(config: dict[str, Any]):
             actual data), target table, time interval, and metrics list. Results stored in
             'simple_metrics' table
     """
+    base = common_config(config)
+    frequency_alert = config.get('frequency_alert')
+    # SIENTIAPDE-1995: laborious flips `adaptive_state.escalated` on a threshold breach (and back
+    # after the window is renewed without one) and bumps `updated_at`, which makes this tick
+    # rewrite the schedule.
+    if frequency_alert and (config.get('adaptive_state') or {}).get('escalated'):
+        base['frequency'] = frequency_alert
     return {
-        **common_config(config),
+        **base,
         'schema': 'sientia_data',
         'predictions_table_name': 'predictions',
         'data_table_name': 'laborious_data',
         'target_table_name': 'simple_metrics',
         'interval_minutes': config.get('interval_minutes', 60),
         'metrics': config.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
+        # Forwarded so create_schedule_config can warn on a value below the floor.
+        'frequency_alert': frequency_alert,
     }
 
 

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -110,6 +110,7 @@ def test_simple_metrics():
         'target_table_name': 'simple_metrics',
         'interval_minutes': 120,
         'metrics': ['rmse', 'mse'],
+        'frequency_alert': None,
         'execution_timeout_seconds': 300,
         'task_timeout_seconds': 300,
         'runtime': 'legacy',
@@ -117,6 +118,43 @@ def test_simple_metrics():
     assert result == expected
 
 
+def _simple_metrics_config(**fields):
+    return {
+        'workflow_type': 'simple_metrics',
+        'schedule_name': 'm1-simple_metrics-60m',
+        'model_id': 'm1',
+        'model': {'name': 'm1'},
+        'frequency': '1h',
+        **fields,
+    }
+
+
+def test_simple_metrics_escalated_uses_frequency_alert():
+    config = _simple_metrics_config(
+        frequency_alert='5m', adaptive_state={'escalated': True, 'healthy_streak': 0}
+    )
+    result = simple_metrics(config)
+    assert result['frequency'] == '5m'
+    assert result['frequency_alert'] == '5m'
+
+
+def test_simple_metrics_not_escalated_keeps_frequency():
+    config = _simple_metrics_config(
+        frequency_alert='5m', adaptive_state={'escalated': False, 'healthy_streak': 2}
+    )
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_simple_metrics_escalated_without_frequency_alert_keeps_frequency():
+    # Removing frequency_alert turns adaptive mode off even if a stale state says escalated.
+    config = _simple_metrics_config(adaptive_state={'escalated': True, 'healthy_streak': 0})
+    assert simple_metrics(config)['frequency'] == '1h'
+
+
+def test_simple_metrics_frequency_alert_without_state_keeps_frequency():
+    assert simple_metrics(_simple_metrics_config(frequency_alert='5m'))['frequency'] == '1h'
+
+
 def test_minimal_retrain():
     config = {
         'workflow_type': 'minimal_retrain',