SIENTIAPDE-2212

SIENTIAPDE-2212-drift-janela-e-chunk-period: Add configurable chunk period and dynamic window calculation for drift, with warnings for invalid periods.

.gitignore

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

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,28 @@ 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')
+# Re-reads the tail of the previous window so rows ingested up to 1 minute late still count.
+DRIFT_WINDOW_MARGIN_MINUTES = 1
+
+
+def _drift_interval_minutes(config: dict[str, Any], frequency: Any) -> int:
+    """
+    Drift window in minutes: `interval_minutes` as configured (no margin), otherwise the
+    schedule frequency rounded up to whole minutes (min 1) plus DRIFT_WINDOW_MARGIN_MINUTES.
+    """
+    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)) + DRIFT_WINDOW_MARGIN_MINUTES
+
+
 def drift(config: dict[str, Any]):
     """
     Build drift configuration from pipeline config.
@@ -52,25 +75,52 @@ 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 (no margin). Default: `frequency` rounded up to whole minutes
+                (min 1) plus a 1-minute margin for late data (2 with the default '1m').
+                Pipelines whose ingestion lag exceeds 1 minute should set it explicitly
+                (frequency + lag); `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_plus_margin():
+    result = drift(_drift_config())
+
+    assert result['frequency'] == '1m'
+    assert result['interval'] == 2
+
+
+@mark.parametrize(
+    ('frequency', 'interval'),
+    [('1h', 61), ('30s', 2), ('90s', 3)],
+)
+def test_drift_window_derived_from_frequency(frequency, interval):
+    assert drift(_drift_config(frequency=frequency))['interval'] == interval
+
+
+def test_drift_explicit_interval_minutes_wins_without_margin():
+    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',