SIENTIAPDE-2212

Add e2e scenario for drift window and chunk_period

- Add scenario 1.8.1: two drift pipelines on frequency 1m. One has no
  interval_minutes and an invalid chunk_period='hour', so it expects
  interval=2, chunk_period='min' and one DRIFT_CHUNK_PERIOD_INVALID
  notification. The other sets interval_minutes=60 and chunk_period='s',
  and both are forwarded unchanged with no warning (SIENTIAPDE-2212).
- Add a _schedule_input helper that decodes the workflow input of a
  laborious-namespace schedule.
- Reformat the ensure_namespace_search_attributes signature in
  conftest.py to a single line.

e2e/conftest.py

@@ -92,9 +92,7 @@ async def register_namespace_if_missing(env: WorkflowEnvironment, namespace: str
     raise TimeoutError(f'Namespace {namespace} not ready within 5s')
 
 
-async def ensure_namespace_search_attributes(
-    env: WorkflowEnvironment, namespace: str
-) -> None:
+async def ensure_namespace_search_attributes(env: WorkflowEnvironment, namespace: str) -> None:
     """
     Register the orchestrator search attributes on a namespace, if missing.
 

e2e/scenario_inputs/orchestrator_drift_window.json

@@ -0,0 +1,38 @@
+{
+  "workflow_input": {
+    "schedule_name": "orchestrator-e2e-drift-window",
+    "pipelines_query": {
+      "collection": "pipelines",
+      "aggregation": [{"$match": {"active": true}}]
+    },
+    "opc_servers_query": {
+      "collection": "opc_servers",
+      "filters": {"active": true}
+    }
+  },
+  "pipelines": [
+    {
+      "schedule_name": "model-dr-drift-default",
+      "workflow_type": "drift",
+      "model_id": "model-dr",
+      "model": {"name": "Model model-dr"},
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1m",
+      "chunk_period": "hour"
+    },
+    {
+      "schedule_name": "model-dr-drift-explicit",
+      "workflow_type": "drift",
+      "model_id": "model-dr",
+      "model": {"name": "Model model-dr"},
+      "active": true,
+      "updated_at": "@now",
+      "frequency": "1m",
+      "interval_minutes": 60,
+      "chunk_period": "s"
+    }
+  ],
+  "opc_servers": [],
+  "active_ingestors": ["heartbeat:ingestor:1"]
+}

e2e/scenarios.md

@@ -56,6 +56,14 @@ Laborious-style write (`adaptive_state.escalated=true` + BSON-datetime `updated_
 window → next run switches only that schedule to `frequency_alert`; the recovery write switches it
 back to `frequency`.
 
+### 1.8.1 Drift window and chunk_period
+
+Two `drift` pipelines on `frequency='1m'`: one without `interval_minutes` and with an invalid
+`chunk_period='hour'` → workflow input `interval=2` (frequency + 1-minute margin),
+`chunk_period='min'` and exactly one `DRIFT_CHUNK_PERIOD_INVALID` notification; one with
+`interval_minutes=60` and `chunk_period='s'` → both forwarded unchanged, no warning
+(SIENTIAPDE-2212).
+
 ---
 
 ## 2. Alerts workflow

e2e/test_orchestrator_main_workflow.py

@@ -313,3 +313,54 @@ async def test_scenario_1_7_2_simple_metrics_adaptive_frequency_round_trip(
     _laborious_adaptive_write(mongo_uri, short, escalated=False)
     await run('recover')
     assert await _schedule_every(temporal_host, short) == timedelta(hours=1)
+
+
+async def _schedule_input(host: str, schedule_id: str) -> dict:
+    """Return the decoded workflow input of a laborious-namespace schedule."""
+    client = await Client.connect(host, namespace='laborious')
+    description = await client.get_schedule_handle(schedule_id).describe()
+    action = description.schedule.action
+    assert isinstance(action, ScheduleActionStartWorkflow)
+    [input_data] = await client.data_converter.decode(action.args)
+    return input_data
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_scenario_1_8_1_drift_window_and_chunk_period(
+    temporal_env: WorkflowEnvironment,
+    orchestrator_worker,
+    stub_workers,
+    mongo_uri: str,
+    temporal_host: str,
+    redis_client: Redis,
+    notification_handler,
+):
+    """1.8.1: Drift window defaults to frequency + 1 min margin; invalid chunk_period warns and falls back."""
+    scenario = load_scenario_input('orchestrator_drift_window')
+    _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-drift-window'),
+    )
+
+    default = await _schedule_input(temporal_host, 'model-dr-drift-default')
+    assert default['interval'] == 2
+    assert default['chunk_period'] == 'min'
+    assert default['chunk_period_configured'] == 'hour'
+
+    explicit = await _schedule_input(temporal_host, 'model-dr-drift-explicit')
+    assert explicit['interval'] == 60
+    assert explicit['chunk_period'] == 's'
+    assert 'chunk_period_configured' not in explicit
+
+    warnings = list(
+        notification_handler.mongo_collection.find(
+            {'notification_id': 'DRIFT_CHUNK_PERIOD_INVALID'}
+        )
+    )
+    assert len(warnings) == 1
+    assert warnings[0]['model_id'] == 'model-dr'