test(e2e): one simple_metrics schedule per window for the same model (SIENTIAPDE-2193)

e2e/scenario_inputs/orchestrator_simple_metrics_windows.json

@@ -0,0 +1,45 @@
+{
+  "workflow_input": {
+    "schedule_name": "orchestrator-e2e-sm-windows",
+    "pipelines_query": {
+      "collection": "pipelines",
+      "aggregation": [{"$match": {"active": true}}]
+    },
+    "opc_servers_query": {
+      "collection": "opc_servers",
+      "filters": {"active": true}
+    }
+  },
+  "pipelines": [
+    {
+      "schedule_name": "model-sm-simple_metrics-60m",
+      "workflow_type": "simple_metrics",
+      "model_id": "model-sm",
+      "model": {
+        "name": "Model model-sm",
+        "model_config": {"target": "y", "simple_metrics_thresholds": {"rmse_max": 1.0}}
+      },
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1h",
+      "frequency_alert": "5m",
+      "interval_minutes": 60
+    },
+    {
+      "schedule_name": "model-sm-simple_metrics-1440m",
+      "workflow_type": "simple_metrics",
+      "model_id": "model-sm",
+      "model": {
+        "name": "Model model-sm",
+        "model_config": {"target": "y", "simple_metrics_thresholds": {"rmse_max": 1.0}}
+      },
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1d",
+      "frequency_alert": "1h",
+      "interval_minutes": 1440
+    }
+  ],
+  "opc_servers": [],
+  "active_ingestors": ["heartbeat:ingestor:1"]
+}

e2e/scenarios.md

@@ -45,6 +45,11 @@ unrelated Temporal schedule present.
 
 First run creates TTL index on notification collection used by scouter pipelines.
 
+### 1.7.1 Simple metrics parallel windows
+
+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.
+
 ---
 
 ## 2. Alerts workflow

e2e/test_orchestrator_main_workflow.py

@@ -1,5 +1,7 @@
 """E2E tests for the Orchestrator main workflow."""
 
+from datetime import timedelta
+
 import pytest
 from pymongo import MongoClient
 from redis import Redis
@@ -26,6 +28,13 @@ async def _schedule_ids(host: str, namespace: str) -> list[str]:
     return [schedule.id async for schedule in await client.list_schedules()]
 
 
+async def _schedule_every(host: str, schedule_id: str) -> timedelta:
+    """Return the interval of a laborious-namespace schedule."""
+    client = await Client.connect(host, namespace='laborious')
+    description = await client.get_schedule_handle(schedule_id).describe()
+    return description.schedule.spec.intervals[0].every
+
+
 def _apply_orchestrator_seeds(
     scenario: dict,
     mongo_uri: str,
@@ -224,3 +233,29 @@ async def test_scenario_1_6_1_ttl_index_bootstrap(
         assert any('expireAfterSeconds' in info for info in indexes.values())
     finally:
         mongo.close()
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_1_7_1_simple_metrics_parallel_windows(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    temporal_host: str,
+    redis_client: Redis,
+):
+    """1.7.1: One pipelines doc per window gives one schedule per window, each with its own interval."""
+    scenario = load_scenario_input('orchestrator_simple_metrics_windows')
+    _apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
+
+    await start_and_await_workflow(
+        temporal_env.client,
+        Orchestrator.run,
+        scenario['workflow_input'],
+        make_workflow_id('orchestrator-sm-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)