test(e2e): adaptive state is escalated in pipelines on threshold breach (SIENTIAPDE-2191)

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
@@ -332,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)