Merge pull request #50 from Aignosi/fix/SIENTIAPDE-2212-drift-janela-e-chunk-period

SIENTIAPDE-2212: Introduce configurable chunk period and dynamic drift window

.gitignore

@@ -49,4 +49,5 @@ git_log
 .env
 
 openspec/
-.cursor/
+.cursor/
+.claude/

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(
+async def ensure_namespace_search_attributes(env: WorkflowEnvironment, namespace: str) -> None:
-    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=1` (frequency, no overlap),
+`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; 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'] == 1
+    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'

orchestrator/activities/formatters.py

@@ -19,6 +19,7 @@ with workflow.unsafe.imports_passed_through():
 
     from orchestrator.utils.orchestrator_functions import (
         build_tag_config,
+        chunk_period_warning,
         drift,
         frequency_alert_warning,
         gather_read_tags,
@@ -371,7 +372,9 @@ class Formatters(SientiaMonitoring):
         configuration to determine which schedules need to be created, updated, or deleted.
         Uses timestamp comparison to identify schedules that have changed.
         Also sends a SIMPLE_METRICS_FREQUENCY_ALERT_BELOW_FLOOR WARNING for laborious schedules
-        whose frequency_alert is below 1m (the value is kept) or unparseable (it is ignored).
+        whose frequency_alert is below 1m (the value is kept) or unparseable (it is ignored),
+        and a DRIFT_CHUNK_PERIOD_INVALID WARNING for drift schedules whose configured
+        chunk_period is not supported by laborious (replaced by 'min').
 
         Args:
             input_data (dict[str, Any]): The input data containing:
@@ -430,6 +433,20 @@ class Formatters(SientiaMonitoring):
                     block='create_schedule_config',
                     level=NotificationLevel.WARNING,
                 )
+            warning = chunk_period_warning(schedule.get('chunk_period_configured'))
+            if warning:
+                self.send_notification(
+                    metadata={
+                        **metadata,
+                        'schedule_name': schedule_name,
+                        'model_id': schedule.get('model_id'),
+                        'model_name': schedule.get('model_name'),
+                    },
+                    notification_id='DRIFT_CHUNK_PERIOD_INVALID',
+                    message=warning,
+                    block='create_schedule_config',
+                    level=NotificationLevel.WARNING,
+                )
 
         for namespace, schedules in current_schedule_config.items():
             for schedule_name in schedules:

orchestrator/activities/temporal_manager.py

@@ -220,6 +220,9 @@ class TemporalManager(SientiaMonitoring):
                     task_queue_name = self._build_task_queue_name(workflow_type, schedule)
                     schedule['task_queue'] = task_queue_name
 
+                    # No `policy`: schedules rely on Temporal's default SKIP overlap. The
+                    # laborious `DriftAnalysis` cache assumes at most one running drift
+                    # workflow per model; revisit its locking if a policy is ever added here.
                     await client.create_schedule(
                         schedule_name,
                         Schedule(

orchestrator/utils/orchestrator_functions.py

@@ -1,3 +1,4 @@
+from math import ceil
 from typing import Any
 
 from orchestrator.utils.converters import parse_frequency
@@ -43,6 +44,27 @@ def common_config(config: dict[str, Any]):
     }
 
 
+# Kept in sync by hand with the `chunk_period` validation in
+# sientia-dataops-laborious/laborious/activities/model_metrics.py (no shared contract).
+DRIFT_CHUNK_PERIODS = ('min', 's')
+
+
+def _drift_interval_minutes(config: dict[str, Any], frequency: Any) -> int:
+    """
+    Drift window in minutes: `interval_minutes` as configured, otherwise the schedule frequency
+    rounded up to whole minutes (min 1). Consecutive default windows do not overlap, so each
+    drift chunk is stored once.
+    """
+    if 'interval_minutes' in config:
+        return config['interval_minutes']
+    try:
+        seconds = parse_frequency(str(frequency))
+    except ValueError:
+        # create_schedules fails this schedule on its own; don't kill the tick for every model.
+        return 60
+    return max(1, ceil(seconds / 60))
+
+
 def drift(config: dict[str, Any]):
     """
     Build drift configuration from pipeline config.
@@ -52,25 +74,51 @@ def drift(config: dict[str, Any]):
 
     Args:
         config (dict[str, Any]): Pipeline configuration containing:
-            - interval_minutes (int, optional): Time window for data comparison in minutes (default: 60)
+            - interval_minutes (int, optional): Time window for data comparison in minutes,
+                used as-is. Default: `frequency` rounded up to whole minutes (min 1), so
+                consecutive windows do not overlap (1 with the default '1m');
+                `interval_minutes: 60` keeps the legacy window.
+            - chunk_period (str, optional): Drift bucket size, 'min' or 's' (default: 'min').
+                Any other value falls back to 'min' and create_schedule_config warns.
             - drift_metrics (list[str], optional): Statistical metrics to compute (default:
                 ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein'])
             - Additional fields from common_config
 
     Returns:
         dict[str, Any]: Drift detection configuration with source/target tables,
-            time interval, and metrics specifications. Results stored in 'drift_metrics' table
+            time interval, chunk period and metrics specifications. Results stored in
+            'drift_metrics' table
     """
-    return {
+    base = common_config(config)
-        **common_config(config),
+    configured_chunk_period = config.get('chunk_period')
+    chunk_period_valid = configured_chunk_period in DRIFT_CHUNK_PERIODS
+    output = {
+        **base,
         'schema': 'sientia_data',
         'source_table_name': 'laborious_data',
         'target_table_name': 'drift_metrics',
-        'interval': config.get('interval_minutes', 60),
+        'interval': _drift_interval_minutes(config, base['frequency']),
+        'chunk_period': configured_chunk_period if chunk_period_valid else 'min',
         'drift_metrics': config.get(
             'drift_metrics', ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']
         ),
     }
+    if configured_chunk_period is not None and not chunk_period_valid:
+        # Forwarded so create_schedule_config can warn without re-reading the pipeline.
+        output['chunk_period_configured'] = configured_chunk_period
+    return output
+
+
+def chunk_period_warning(chunk_period: Any) -> str | None:
+    """
+    Warning message when a configured drift `chunk_period` is not supported by laborious,
+    otherwise None. Warn only: `drift()` already replaced it with 'min'.
+    """
+    if chunk_period is None or chunk_period in DRIFT_CHUNK_PERIODS:
+        return None
+    return (
+        f"chunk_period={chunk_period!r} is not one of {DRIFT_CHUNK_PERIODS}; falling back to 'min'"
+    )
 
 
 def simple_metrics(config: dict[str, Any]):

tests/orchestrator/activities/test_formatters.py

@@ -1073,3 +1073,46 @@ def test_create_schedule_config_no_warning_at_or_above_floor(formatters):
     formatters.create_schedule_config(_schedule_config_with_frequency_alert('5m'))
 
     formatters.send_notification.assert_not_called()
+
+
+def _drift_schedule_config(chunk_period_configured):
+    return {
+        'metadata': {},
+        'current_schedule_config': {'laborious': {}},
+        'schedule_config': {
+            'laborious': {
+                'm1-drift-1m': {
+                    'workflow_type': 'drift',
+                    'model_id': 'm1',
+                    'model_name': 'm1',
+                    'frequency': '1m',
+                    'chunk_period': 'min',
+                    'chunk_period_configured': chunk_period_configured,
+                    'updated_at': '2026-10-27',
+                }
+            },
+            'scouter': {},
+        },
+    }
+
+
+def test_create_schedule_config_warns_on_invalid_chunk_period(formatters):
+    result = formatters.create_schedule_config(_drift_schedule_config('hour'))
+
+    assert 'm1-drift-1m' in result['to_create']['laborious']
+    formatters.send_notification.assert_called_once()
+    kwargs = formatters.send_notification.call_args.kwargs
+    assert kwargs['notification_id'] == 'DRIFT_CHUNK_PERIOD_INVALID'
+    assert kwargs['level'] == NotificationLevel.WARNING
+    assert kwargs['metadata']['schedule_name'] == 'm1-drift-1m'
+    assert kwargs['metadata']['model_id'] == 'm1'
+    assert kwargs['metadata']['model_name'] == 'm1'
+
+
+def test_create_schedule_config_no_chunk_period_warning_when_unchanged(formatters):
+    input_data = _drift_schedule_config('hour')
+    input_data['current_schedule_config'] = {'laborious': {'m1-drift-1m': '2026-10-27'}}
+
+    formatters.create_schedule_config(input_data)
+
+    formatters.send_notification.assert_not_called()

tests/orchestrator/activities/test_temporal_manager.py

@@ -3,6 +3,7 @@ from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
+from temporalio.client import ScheduleOverlapPolicy
 
 from orchestrator.activities.temporal_manager import TemporalManager
 from orchestrator.utils.converters import parse_frequency
@@ -684,3 +685,41 @@ async def test_create_schedules_tenant_runtime_queue(
         task_timeout=timedelta(seconds=300),
         typed_search_attributes=_mock_typed_search_attributes.return_value,
     )
+
+
+@mark.asyncio
+@patch('orchestrator.activities.temporal_manager.Schedule')
+@patch('orchestrator.activities.temporal_manager.ScheduleActionStartWorkflow')
+@patch('orchestrator.activities.temporal_manager.ScheduleIntervalSpec')
+@patch('orchestrator.activities.temporal_manager.ScheduleSpec')
+@patch('orchestrator.activities.temporal_manager.TypedSearchAttributes')
+@patch('orchestrator.activities.temporal_manager.SearchAttributePair')
+async def test_create_schedules_drift_keeps_default_skip_overlap(
+    _mock_search_attribute_pair,
+    _mock_typed_search_attributes,
+    _mock_schedule_spec,
+    _mock_schedule_interval_spec,
+    _mock_schedule_action_start_workflow,
+    mock_schedule,
+    temporal_manager,
+):
+    # The laborious DriftAnalysis cache assumes no two drift runs of a model overlap.
+    input_data = {
+        'schedules': {
+            'laborious': {
+                'm1-drift-1m': {
+                    'model_id': 'm1',
+                    'model_name': 'm1',
+                    'workflow_type': 'drift',
+                    'frequency': '1m',
+                }
+            }
+        }
+    }
+    temporal_manager.temporal_clients['laborious'].create_schedule = AsyncMock()
+
+    await temporal_manager.create_schedules(input_data)
+
+    mock_schedule.assert_called_once()
+    policy = mock_schedule.call_args.kwargs.get('policy')
+    assert policy is None or policy.overlap == ScheduleOverlapPolicy.SKIP

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -5,6 +5,7 @@ from pytest import mark
 from orchestrator.utils.orchestrator_functions import (
     base_scouter,
     build_tag_config,
+    chunk_period_warning,
     common_config,
     drift,
     frequency_alert_warning,
@@ -79,6 +80,7 @@ def test_drift():
         'source_table_name': 'laborious_data',
         'target_table_name': 'drift_metrics',
         'interval': 120,
+        'chunk_period': 'min',
         'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'],
         'execution_timeout_seconds': 300,
         'task_timeout_seconds': 300,
@@ -87,6 +89,79 @@ def test_drift():
     assert result == expected
 
 
+def _drift_config(**fields):
+    return {
+        'workflow_type': 'drift',
+        'schedule_name': 'm1-drift',
+        'model_id': 'm1',
+        'model': {'name': 'm1'},
+        **fields,
+    }
+
+
+def test_drift_default_window_is_frequency():
+    result = drift(_drift_config())
+
+    assert result['frequency'] == '1m'
+    assert result['interval'] == 1
+
+
+@mark.parametrize(
+    ('frequency', 'interval'),
+    [('1h', 60), ('30s', 1), ('90s', 2)],
+)
+def test_drift_window_derived_from_frequency(frequency, interval):
+    assert drift(_drift_config(frequency=frequency))['interval'] == interval
+
+
+def test_drift_explicit_interval_minutes_wins():
+    result = drift(_drift_config(frequency='1m', interval_minutes=60))
+
+    assert result['interval'] == 60
+
+
+def test_drift_invalid_frequency_falls_back_to_legacy_window():
+    result = drift(_drift_config(frequency='abc'))
+
+    assert result['frequency'] == 'abc'
+    assert result['interval'] == 60
+
+
+def test_drift_chunk_period_defaults_to_min():
+    result = drift(_drift_config())
+
+    assert result['chunk_period'] == 'min'
+    assert 'chunk_period_configured' not in result
+
+
+def test_drift_chunk_period_seconds_kept():
+    result = drift(_drift_config(chunk_period='s'))
+
+    assert result['chunk_period'] == 's'
+    assert 'chunk_period_configured' not in result
+
+
+@mark.parametrize('chunk_period', ['hour', 5])
+def test_drift_invalid_chunk_period_falls_back_to_min(chunk_period):
+    result = drift(_drift_config(chunk_period=chunk_period))
+
+    assert result['chunk_period'] == 'min'
+    assert result['chunk_period_configured'] == chunk_period
+
+
+@mark.parametrize('chunk_period', [None, 'min', 's'])
+def test_chunk_period_warning_none_when_absent_or_valid(chunk_period):
+    assert chunk_period_warning(chunk_period) is None
+
+
+@mark.parametrize('chunk_period', ['hour', 5])
+def test_chunk_period_warning_message_when_invalid(chunk_period):
+    warning = chunk_period_warning(chunk_period)
+
+    assert warning is not None
+    assert repr(chunk_period) in warning
+
+
 def test_simple_metrics():
     config = {
         'workflow_type': 'simple_metrics',