test(e2e): adaptive frequency escalation and recovery round trip (SIENTIAPDE-2193)

e2e/scenarios.md

@@ -50,6 +50,12 @@ First run creates TTL index on notification collection used by scouter pipelines
 Two `simple_metrics` pipelines for the same `model_id` (60 min and 1440 min windows) → one
 Temporal schedule per window, each with its own `frequency` as interval.
 
+### 1.7.2 Adaptive frequency escalation/recovery round trip
+
+Laborious-style write (`adaptive_state.escalated=true` + BSON-datetime `updated_at`) on one
+window → next run switches only that schedule to `frequency_alert`; the recovery write switches it
+back to `frequency`.
+
 ---
 
 ## 2. Alerts workflow

e2e/test_orchestrator_main_workflow.py

@@ -1,6 +1,6 @@
 """E2E tests for the Orchestrator main workflow."""
 
-from datetime import timedelta
+from datetime import UTC, datetime, timedelta
 
 import pytest
 from pymongo import MongoClient
@@ -259,3 +259,57 @@ async def test_scenario_1_7_1_simple_metrics_parallel_windows(
     short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m'
     assert await _schedule_every(temporal_host, short) == timedelta(hours=1)
     assert await _schedule_every(temporal_host, long) == timedelta(days=1)
+
+
+def _laborious_adaptive_write(mongo_uri: str, schedule_name: str, escalated: bool) -> None:
+    """Replicates laborious `update_simple_metrics_adaptive_state` on an escalated flip."""
+    mongo = MongoClient(mongo_uri)
+    try:
+        mongo[E2E_DATABASE]['pipelines'].update_one(
+            {'schedule_name': schedule_name},
+            {
+                '$set': {
+                    'adaptive_state': {'escalated': escalated, 'healthy_streak': 0},
+                    # BSON datetime, as laborious writes it; a string would break the tick.
+                    'updated_at': datetime.now(UTC),
+                }
+            },
+        )
+    finally:
+        mongo.close()
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_1_7_2_simple_metrics_adaptive_frequency_round_trip(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    temporal_host: str,
+    redis_client: Redis,
+):
+    """1.7.2: Escalation written by laborious switches only that window to frequency_alert; recovery switches it back."""
+    scenario = load_scenario_input('orchestrator_simple_metrics_windows')
+    _apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
+    short, long = 'model-sm-simple_metrics-60m', 'model-sm-simple_metrics-1440m'
+
+    async def run(tag: str) -> None:
+        await start_and_await_workflow(
+            temporal_env.client,
+            Orchestrator.run,
+            scenario['workflow_input'],
+            make_workflow_id(f'orchestrator-sm-adaptive-{tag}'),
+        )
+
+    await run('create')
+    assert await _schedule_every(temporal_host, short) == timedelta(hours=1)
+
+    _laborious_adaptive_write(mongo_uri, short, escalated=True)
+    await run('escalate')
+    assert await _schedule_every(temporal_host, short) == timedelta(minutes=5)
+    assert await _schedule_every(temporal_host, long) == timedelta(days=1)
+
+    _laborious_adaptive_write(mongo_uri, short, escalated=False)
+    await run('recover')
+    assert await _schedule_every(temporal_host, short) == timedelta(hours=1)