From 0742078faddc8399ec696ab76b09cd211e19003c Mon Sep 17 00:00:00 2001 From: Carla Pinto Date: Mon, 5 Oct 2026 22:14:18 +0200 Subject: [PATCH 1/9] rc/SIENTIAPDE-2165- feat: rich metadata (filter/policy/counter) for mlflow_response_gate --- laborious/activities/gates.py | 73 +++++++++++++++++-- .../sub_workflows/prediction_process.py | 14 ++-- tests/laborious/activities/test_gates.py | 36 +++++++-- .../subworkflows/test_prediction_process.py | 24 +++--- 4 files changed, 112 insertions(+), 35 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index c6fece9..0c981fa 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -28,6 +28,7 @@ InputFilterFunc = Callable[[DataFrame, dict[str, Any]], bool] ResponseFilterFunc = Callable[[dict[str, Any], dict[str, Any]], bool] ContentFilterFunc = Callable[[DataFrame, dict[str, Any]], bool] InputCounterFunc = Callable[[DataFrame, dict[str, Any]], int] +ResponseCounterFunc = Callable[[dict[str, Any], dict[str, Any]], int] # Input filter function mappings input_filter_functions: dict[str, InputFilterFunc] = { @@ -78,11 +79,31 @@ input_path_confidence: Mapping[str, int] = { 'REPEAT': -1, } + +def _count_api_error(_response: dict[str, Any], _config: dict[str, Any]) -> int: + """ + Count for the API_ERROR filter. + + There's no per-row/per-item unit to count here — `api_error_filter` evaluates a single + API response as a whole, not tabular data — so this mirrors `_count_empty_data`'s + approach to a structural (not countable) trigger: a fixed value, here 1 (one API error + event), instead of 0 (zero rows present) like `_count_empty_data`'s own case. + + Return: + int: Always 1. + """ + return 1 + + # MLFlow response filter function mappings mlflow_response_filter_functions: dict[str, ResponseFilterFunc] = { 'API_ERROR': api_error_filter, } +mlflow_response_filter_counters: dict[str, ResponseCounterFunc] = { + 'API_ERROR': _count_api_error, +} + mlflow_response_path_confidence: Mapping[str, int] = { 'STOP': -1, 'CONTINUE': 10, @@ -320,7 +341,9 @@ class Gates(SientiaMonitoring): return None, 0, '', None @activity.defn(name='mlflow_response_gate') - def mlflow_response_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]: + def mlflow_response_gate( + self, input_data: dict[str, Any] + ) -> tuple[str | None, int, str, dict[str, Any] | None]: """ Validate MLFlow API response quality and integrity. @@ -334,6 +357,9 @@ class Gates(SientiaMonitoring): 3. Determines path decisions based on response validation results 4. Provides confidence scores and detailed validation comments 5. Handles API errors and response validation failures + 6. Builds a metadata document describing which filter(s) triggered, with their policy + and a counter, whenever one does — this activity only computes it, it never writes it + (see `PredictionMetadata.write_prediction_metadata` for the actual write) Args: input_data: Configuration and data for response validation @@ -345,10 +371,14 @@ class Gates(SientiaMonitoring): - path_priority (list[str]): Priority order for path decisions Returns: - tuple: (path_flag, confidence, comment) + tuple: (path_flag, confidence, comment, document) - path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None) - confidence (int): Confidence score for the decision - comment (str): Detailed explanation of the decision + - document (dict | None): Metadata describing the filter(s) that produced + `path_flag`, shaped `{'filters': [{'name', 'policy', 'counter'}, ...]}`; + `None` when nothing was filtered. Not written anywhere by this activity — + the caller decides whether and how to persist it. Raises: Exception: If response validation fails or configuration is invalid @@ -369,10 +399,11 @@ class Gates(SientiaMonitoring): gate_type = input_data['type'] path_priority = input_data['path_priority'] - filter_output = [] - comments = [] + # (filter_name, policy, filter_config) per filter that triggered, in configured order + triggered_filters: list[tuple[str, str, dict[str, Any]]] = [] + status = payload.status or {} for fil, config in filters.items(): @@ -381,7 +412,7 @@ class Gates(SientiaMonitoring): policy, filter_config = self._read_filter_entry(config) try: if mlflow_response_filter_functions[fil](status, filter_config): - filter_output.append(policy) + triggered_filters.append((fil, policy, filter_config)) comments.append(status.get('message', 'Unknown MLFlow API error')) self.send_notification( metadata=metadata, @@ -402,16 +433,42 @@ class Gates(SientiaMonitoring): attachment_content=trace, ) + triggered_policies = [policy for _, policy, _ in triggered_filters] + for path_flag in path_priority: - if path_flag in filter_output: + if path_flag in triggered_policies: self.info(f'Mlflow response gate result: {path_flag}', metadata) - return path_flag, mlflow_response_path_confidence[path_flag], ', '.join(comments) + + matched = [ + (fil, filter_config) + for fil, policy, filter_config in triggered_filters + if policy == path_flag + ] + document = { + 'filters': [ + { + 'name': fil, + 'policy': path_flag, + 'counter': mlflow_response_filter_counters.get(fil, lambda *_: 0)( + status, filter_config + ), + } + for fil, filter_config in matched + ], + } + + return ( + path_flag, + mlflow_response_path_confidence[path_flag], + ', '.join(comments), + document, + ) self.info('Nothing was filtered by the mlflow response gate', metadata) del data - return None, 0, '' + return None, 0, '', None @activity.defn(name='mlflow_content_gate') def mlflow_content_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]: diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 9c59240..cc90bc7 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -156,7 +156,7 @@ class PredictionProcess: ) # Validate MLFlow transform response - path_flag, confidence, comment = await workflow.execute_local_activity_method( + path_flag, confidence, comment, document = await workflow.execute_local_activity_method( Activities.mlflow_response_gate, { **metadata, @@ -171,7 +171,7 @@ class PredictionProcess: # Handle path decision based on transform validation if await self.path_flag_handler( - data, path_flag, input_data, confidence, last_timestamp, comment + data, path_flag, input_data, confidence, last_timestamp, comment, document ): return @@ -206,7 +206,7 @@ class PredictionProcess: ) # Validate MLFlow prediction response - path_flag, confidence, comment = await workflow.execute_local_activity_method( + path_flag, confidence, comment, document = await workflow.execute_local_activity_method( Activities.mlflow_response_gate, { **metadata, @@ -221,7 +221,7 @@ class PredictionProcess: # Handle path decision based on prediction validation if await self.path_flag_handler( - data, path_flag, input_data, confidence, last_timestamp, comment + data, path_flag, input_data, confidence, last_timestamp, comment, document ): return @@ -271,9 +271,9 @@ class PredictionProcess: """ if not document: # Skip the remote activity call entirely rather than dispatching it just to have - # `write_prediction_metadata` decide there's nothing to write — `mlflow_response_gate` - # and `mlflow_content_gate` don't build a `document` yet, so this is the common case - # today, not a rare edge case (review finding, 2026-10-01). + # `write_prediction_metadata` decide there's nothing to write — `mlflow_content_gate` + # doesn't build a `document` yet, so this is the common case today, not a rare edge + # case (review finding, 2026-10-01). return None try: diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index 8f0eb80..d6771af 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -6,7 +6,13 @@ from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.metrics_controller import MetricsController from sientia_do.observability.sientia_monitoring import SientiaMonitoring -from laborious.activities.gates import Gates, input_filter_counters, input_filter_functions +from laborious.activities.gates import ( + Gates, + input_filter_counters, + input_filter_functions, + mlflow_response_filter_counters, + mlflow_response_filter_functions, +) @fixture(autouse=True) @@ -314,7 +320,7 @@ def test_mlflow_response_gate_invalid_filter(gates_activity): result = gates_activity.mlflow_response_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) @patch('laborious.activities.gates.mlflow_response_filter_functions') @@ -341,7 +347,7 @@ def test_mlflow_response_gate_filter_exception( result = gates_activity.mlflow_response_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.send_notification.assert_called_once_with( metadata=metadata['metadata'], notification_id='MLFLOW_GATE_RESPONSE_FILTER__INVALID_FILTER', @@ -369,7 +375,7 @@ def test_mlflow_response_gate_no_filters(gates_activity): result = gates_activity.mlflow_response_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.debug.assert_called() @@ -387,10 +393,11 @@ def test_mlflow_response_gate_with_filter(gates_activity): } # Act - result = gates_activity.mlflow_response_gate(input_data) + path_flag, confidence, comment, document = gates_activity.mlflow_response_gate(input_data) # Assert - assert result == ('STOP', -1, 'API error occurred') + assert (path_flag, confidence, comment) == ('STOP', -1, 'API error occurred') + assert document == {'filters': [{'name': 'API_ERROR', 'policy': 'STOP', 'counter': 1}]} gates_activity.debug.assert_called() gates_activity.send_notification.assert_called() @@ -412,7 +419,12 @@ def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity): result = gates_activity.mlflow_response_gate(input_data) # Assert - assert result == ('STOP', -1, 'API error occurred') + assert result == ( + 'STOP', + -1, + 'API error occurred', + {'filters': [{'name': 'API_ERROR', 'policy': 'STOP', 'counter': 1}]}, + ) def test_mlflow_response_gate_with_filter_not_caught(gates_activity): @@ -432,10 +444,18 @@ def test_mlflow_response_gate_with_filter_not_caught(gates_activity): result = gates_activity.mlflow_response_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.debug.assert_called() +def test_mlflow_response_filter_counters_covers_every_mlflow_response_filter_function(): + # Same parity guard as `test_input_filter_counters_covers_every_input_filter_function` + # (review finding, 2026-09-30): a filter added to `mlflow_response_filter_functions` without + # a matching counter here degrades to counter=0 instead of crashing, but this test still + # catches the gap before it ships. + assert set(mlflow_response_filter_counters) == set(mlflow_response_filter_functions) + + def test_mlflow_content_gate_invalid_filter(gates_activity): # Arrange input_data = { diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 82abe8c..040cac8 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -59,11 +59,11 @@ async def test_run(workflow_mock, prediction_process): workflow_mock.execute_local_activity_method.side_effect = [ ('continue', 0.95, 'Input data with bad quality', None), # input_gate # mlflow_response_gate (transform) - ('continue', 0.95, 'Error'), + ('continue', 0.95, 'Error', None), # mlflow_content_gate (transform) ('continue', 0.95, 'Transformed data not passed the content filter'), # mlflow_response_gate (predict) - ('continue', 0.95, 'Error'), + ('continue', 0.95, 'Error', None), ] # Act @@ -276,7 +276,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ ] workflow_mock.execute_local_activity_method.side_effect = [ ('repeat', 0.95, 'Input data with bad quality', None), # input_gate - ('continue', 0.95, 'Error'), # mlflow_response_gate (transform) + ('continue', 0.95, 'Error', None), # mlflow_response_gate (transform) ] # Act @@ -367,7 +367,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process workflow_mock.execute_local_activity_method.side_effect = [ ('continue', 0.95, 'Input data with bad quality', None), # input_gate # mlflow_response_gate (transform) - ('continue', 0.95, 'Error'), + ('continue', 0.95, 'Error', None), # mlflow_content_gate (transform) ('continue', 0.95, 'Transformed data not passed the content filter'), ] @@ -478,10 +478,10 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p workflow_mock.execute_local_activity_method.side_effect = [ ('continue', 0.95, 'Input data with bad quality', None), # input_gate # mlflow_response_gate (transform) - ('continue', 0.95, 'Error'), + ('continue', 0.95, 'Error', None), # mlflow_content_gate (transform) ('continue', 0.95, 'Transformed data not passed the content filter'), - ('continue', 0.95, 'Error'), # mlflow_response_gate (predict) + ('continue', 0.95, 'Error', None), # mlflow_response_gate (predict) ] # Act @@ -696,9 +696,9 @@ async def test_path_flag_handler_repeat_no_document_skips_write_activity_call( workflow_mock, prediction_process ): # Arrange: no document (e.g. REPEAT triggered by a gate that doesn't build one yet — today, - # `mlflow_response_gate`/`mlflow_content_gate`) must not even dispatch the remote - # write_prediction_metadata activity — that would be a wasted Temporal round-trip on every - # REPEAT caused by those two gates, not a rare edge case (review finding, 2026-10-01). + # `mlflow_content_gate`) must not even dispatch the remote write_prediction_metadata + # activity — that would be a wasted Temporal round-trip on every REPEAT caused by that + # gate, not a rare edge case (review finding, 2026-10-01). data = {'test': 'data'} # Act @@ -1075,9 +1075,9 @@ async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process): ] workflow_mock.execute_local_activity_method.side_effect = [ ('continue', 0.95, 'ok', None), # input_gate - ('continue', 0.95, ''), - ('continue', 0.95, ''), - ('continue', 0.95, ''), + ('continue', 0.95, '', None), # mlflow_response_gate (transform) + ('continue', 0.95, ''), # mlflow_content_gate (transform) + ('continue', 0.95, '', None), # mlflow_response_gate (predict) ] await prediction_process.run(input_data) From 2649476571715e4f31814fb3ab46874d59ac5fba Mon Sep 17 00:00:00 2001 From: Carla Pinto Date: Thu, 8 Oct 2026 13:58:55 +0200 Subject: [PATCH 2/9] rc/SIENTIAPDE-2165- refactor: inline counter=1 for API_ERROR, drop unnecessary dispatch (review finding) --- laborious/activities/gates.py | 38 ++++++------------------ tests/laborious/activities/test_gates.py | 16 +--------- 2 files changed, 10 insertions(+), 44 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 0c981fa..49e7c87 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -28,7 +28,6 @@ InputFilterFunc = Callable[[DataFrame, dict[str, Any]], bool] ResponseFilterFunc = Callable[[dict[str, Any], dict[str, Any]], bool] ContentFilterFunc = Callable[[DataFrame, dict[str, Any]], bool] InputCounterFunc = Callable[[DataFrame, dict[str, Any]], int] -ResponseCounterFunc = Callable[[dict[str, Any], dict[str, Any]], int] # Input filter function mappings input_filter_functions: dict[str, InputFilterFunc] = { @@ -80,30 +79,11 @@ input_path_confidence: Mapping[str, int] = { } -def _count_api_error(_response: dict[str, Any], _config: dict[str, Any]) -> int: - """ - Count for the API_ERROR filter. - - There's no per-row/per-item unit to count here — `api_error_filter` evaluates a single - API response as a whole, not tabular data — so this mirrors `_count_empty_data`'s - approach to a structural (not countable) trigger: a fixed value, here 1 (one API error - event), instead of 0 (zero rows present) like `_count_empty_data`'s own case. - - Return: - int: Always 1. - """ - return 1 - - # MLFlow response filter function mappings mlflow_response_filter_functions: dict[str, ResponseFilterFunc] = { 'API_ERROR': api_error_filter, } -mlflow_response_filter_counters: dict[str, ResponseCounterFunc] = { - 'API_ERROR': _count_api_error, -} - mlflow_response_path_confidence: Mapping[str, int] = { 'STOP': -1, 'CONTINUE': 10, @@ -439,21 +419,21 @@ class Gates(SientiaMonitoring): if path_flag in triggered_policies: self.info(f'Mlflow response gate result: {path_flag}', metadata) - matched = [ - (fil, filter_config) - for fil, policy, filter_config in triggered_filters - if policy == path_flag - ] + matched = [fil for fil, policy, _ in triggered_filters if policy == path_flag] document = { 'filters': [ { 'name': fil, 'policy': path_flag, - 'counter': mlflow_response_filter_counters.get(fil, lambda *_: 0)( - status, filter_config - ), + # No per-row/per-item unit to count here — `api_error_filter` + # evaluates a single API response as a whole, not tabular data — + # so, unlike `input_gate`'s per-filter counters, this is a fixed + # value: 1 (one API error event). Inlined rather than a counter + # function/dict, since there's only one response filter today and + # this isn't an override of anything (review finding). + 'counter': 1, } - for fil, filter_config in matched + for fil in matched ], } diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index d6771af..38002c2 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -6,13 +6,7 @@ from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.metrics_controller import MetricsController from sientia_do.observability.sientia_monitoring import SientiaMonitoring -from laborious.activities.gates import ( - Gates, - input_filter_counters, - input_filter_functions, - mlflow_response_filter_counters, - mlflow_response_filter_functions, -) +from laborious.activities.gates import Gates, input_filter_counters, input_filter_functions @fixture(autouse=True) @@ -448,14 +442,6 @@ def test_mlflow_response_gate_with_filter_not_caught(gates_activity): gates_activity.debug.assert_called() -def test_mlflow_response_filter_counters_covers_every_mlflow_response_filter_function(): - # Same parity guard as `test_input_filter_counters_covers_every_input_filter_function` - # (review finding, 2026-09-30): a filter added to `mlflow_response_filter_functions` without - # a matching counter here degrades to counter=0 instead of crashing, but this test still - # catches the gap before it ships. - assert set(mlflow_response_filter_counters) == set(mlflow_response_filter_functions) - - def test_mlflow_content_gate_invalid_filter(gates_activity): # Arrange input_data = { From 75b56251ef5ed4c2b21375e2117c0c66fa089ba9 Mon Sep 17 00:00:00 2001 From: Carla Pinto Date: Thu, 8 Oct 2026 14:10:40 +0200 Subject: [PATCH 3/9] rc/SIENTIAPDE-2165 - fix: read MLflow error message/traceback from status['content'], not the top level --- laborious/activities/gates.py | 12 +++++++++--- tests/laborious/activities/test_gates.py | 19 ++++++++++++++++--- 2 files changed, 25 insertions(+), 6 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 49e7c87..2f136e6 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -393,14 +393,20 @@ class Gates(SientiaMonitoring): try: if mlflow_response_filter_functions[fil](status, filter_config): triggered_filters.append((fil, policy, filter_config)) - comments.append(status.get('message', 'Unknown MLFlow API error')) + # The real error lives one level in, under `content` — `request_transform`/ + # `request_predict` nest `message`/`traceback` there on failure (mlflow.py), + # reusing the same `content` key the success path uses for the DataFrame + # (review finding, 2026-10-08: reading `status['message']` directly always + # fell back to the generic default, in every real production occurrence). + error_content = status.get('content') or {} + comments.append(error_content.get('message', 'Unknown MLFlow API error')) self.send_notification( metadata=metadata, notification_id=f'{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}', - message=status.get('message', 'Unknown MLFlow API error'), + message=error_content.get('message', 'Unknown MLFlow API error'), block='mlflow_gate', level=NotificationLevel.ERROR, - attachment_content=status.get('traceback'), + attachment_content=error_content.get('traceback'), ) except Exception as e: trace = traceback.format_exc() diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index 38002c2..d7ef33f 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -380,7 +380,10 @@ def test_mlflow_response_gate_with_filter(gates_activity): 'filters': {'API_ERROR': {'POLICY': 'STOP'}}, 'data': _minio_payload( {'content': {'message': 'API error occurred', 'traceback': 'error trace'}}, - status={'success': False, 'message': 'API error occurred'}, + status={ + 'success': False, + 'content': {'message': 'API error occurred', 'traceback': 'error trace'}, + }, ), 'type': 'test', 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], @@ -393,7 +396,14 @@ def test_mlflow_response_gate_with_filter(gates_activity): assert (path_flag, confidence, comment) == ('STOP', -1, 'API error occurred') assert document == {'filters': [{'name': 'API_ERROR', 'policy': 'STOP', 'counter': 1}]} gates_activity.debug.assert_called() - gates_activity.send_notification.assert_called() + gates_activity.send_notification.assert_called_once_with( + metadata=metadata['metadata'], + notification_id='TEST_GATE_RESPONSE_FILTER__API_ERROR', + message='API error occurred', + block='mlflow_gate', + level=NotificationLevel.ERROR, + attachment_content='error trace', + ) def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity): @@ -403,7 +413,10 @@ def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity): 'filters': {'API_ERROR': {'Policy': 'STOP'}}, 'data': _minio_payload( {'content': {'message': 'API error occurred', 'traceback': 'error trace'}}, - status={'success': False, 'message': 'API error occurred'}, + status={ + 'success': False, + 'content': {'message': 'API error occurred', 'traceback': 'error trace'}, + }, ), 'type': 'test', 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], From 57b79799994059596bc21fb284893d8fbb8527e8 Mon Sep 17 00:00:00 2001 From: Carla Pinto Date: Thu, 8 Oct 2026 15:36:06 +0200 Subject: [PATCH 4/9] rc/SIENTIAPDE-2166- feat: rich metadata (filter/policy/counter) for mlflow_content_gate --- laborious/activities/gates.py | 70 +++++++++++++++++-- .../sub_workflows/prediction_process.py | 15 ++-- tests/laborious/activities/test_gates.py | 37 +++++++--- .../subworkflows/test_prediction_process.py | 16 ++--- 4 files changed, 108 insertions(+), 30 deletions(-) diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 2f136e6..c1f91f3 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -90,12 +90,34 @@ mlflow_response_path_confidence: Mapping[str, int] = { 'REPEAT': -1, } + +def _count_nan_values(data: DataFrame, _config: dict[str, Any]) -> int: + """ + Count rows for the NAN_VALUES filter. + + `NAN_VALUES` only triggers when every value in the DataFrame is NaN, so every row in it + matched the condition — same convention as `_count_specific_variables_null_values`: + counting rows that satisfied the filter's own trigger condition. + + Return: + int: Number of rows in the evaluated DataFrame. + """ + return len(data) + + # MLFlow content filter function mappings mlflow_content_filter_functions: dict[str, ContentFilterFunc] = { 'NAN_VALUES': nan_values_filter, 'EMPTY_DATA': filter_empty_data, } +# `EMPTY_DATA` reuses `_count_empty_data` — same function (`filter_empty_data`) and same +# trigger condition (absence of rows) as `input_gate`'s own EMPTY_DATA. +mlflow_content_filter_counters: dict[str, InputCounterFunc] = { + 'NAN_VALUES': _count_nan_values, + 'EMPTY_DATA': _count_empty_data, +} + mlflow_content_path_confidence: Mapping[str, int] = { 'STOP': -1, 'CONTINUE': 18, @@ -457,7 +479,9 @@ class Gates(SientiaMonitoring): return None, 0, '', None @activity.defn(name='mlflow_content_gate') - def mlflow_content_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]: + def mlflow_content_gate( + self, input_data: dict[str, Any] + ) -> tuple[str | None, int, str, dict[str, Any] | None]: """ Validate MLFlow prediction content quality and integrity. @@ -471,6 +495,9 @@ class Gates(SientiaMonitoring): 3. Determines path decisions based on content validation results 4. Provides confidence scores and detailed validation comments 5. Handles content validation failures and quality issues + 6. Builds a metadata document describing which filter(s) triggered, with their policy + and a counter, whenever one does — this activity only computes it, it never writes it + (see `PredictionMetadata.write_prediction_metadata` for the actual write) Args: input_data: Configuration and data for content validation @@ -482,10 +509,14 @@ class Gates(SientiaMonitoring): - path_priority (list[str]): Priority order for path decisions Returns: - tuple: (path_flag, confidence, comment) + tuple: (path_flag, confidence, comment, document) - path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None) - confidence (int): Confidence score for the decision - comment (str): Detailed explanation of the decision + - document (dict | None): Metadata describing the filter(s) that produced + `path_flag`, shaped `{'filters': [{'name', 'policy', 'counter'}, ...]}`; + `None` when nothing was filtered. Not written anywhere by this activity — + the caller decides whether and how to persist it. Raises: Exception: If content validation fails or configuration is invalid @@ -501,7 +532,8 @@ class Gates(SientiaMonitoring): gate_type = input_data['type'] path_priority = input_data['path_priority'] - filter_output = [] + # (filter_name, policy, filter_config) per filter that triggered, in configured order + triggered_filters: list[tuple[str, str, dict[str, Any]]] = [] self._debug_dataframe('Input data:', data, metadata) self.debug(f'Filters: \n {filters}', metadata) @@ -512,7 +544,7 @@ class Gates(SientiaMonitoring): policy, filter_config = self._read_filter_entry(config) try: if mlflow_content_filter_functions[fil](data, filter_config): - filter_output.append(policy) + triggered_filters.append((fil, policy, filter_config)) self.send_notification( metadata=metadata, notification_id=f'{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}', @@ -532,20 +564,46 @@ class Gates(SientiaMonitoring): attachment_content=trace, ) + triggered_policies = [policy for _, policy, _ in triggered_filters] + for path_flag in path_priority: - if path_flag in filter_output: + if path_flag in triggered_policies: self.info(f'Mlflow content gate result: {path_flag}', metadata) + + matched = [ + (fil, filter_config) + for fil, policy, filter_config in triggered_filters + if policy == path_flag + ] + document = { + 'filters': [ + { + 'name': fil, + 'policy': path_flag, + # `.get(..., lambda *_: 0)`, not `[fil]`: same defensive fallback as + # `input_gate` (review finding, 2026-09-30) — a filter added to + # `mlflow_content_filter_functions` without a matching counter here + # degrades to counter=0 instead of a raw KeyError. + 'counter': mlflow_content_filter_counters.get(fil, lambda *_: 0)( + data, filter_config + ), + } + for fil, filter_config in matched + ], + } + return ( path_flag, mlflow_content_path_confidence[path_flag], 'Transformed data not passed the content filter', + document, ) self.info('Nothing was filtered by the mlflow content gate', metadata) del data - return None, 0, '' + return None, 0, '', None def get_prediction_store_policy( self, prediction_store_policy: str, metadata: dict[str, Any] diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index cc90bc7..34999f4 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -175,7 +175,7 @@ class PredictionProcess: ): return - path_flag, confidence, comment = await workflow.execute_local_activity_method( + path_flag, confidence, comment, document = await workflow.execute_local_activity_method( Activities.mlflow_content_gate, { **metadata, @@ -189,7 +189,7 @@ class PredictionProcess: ) if await self.path_flag_handler( - data, path_flag, input_data, confidence, last_timestamp, comment + data, path_flag, input_data, confidence, last_timestamp, comment, document ): return @@ -262,8 +262,7 @@ class PredictionProcess: Args: metadata: Workflow metadata for contextual logging. document: Metadata produced by the gate that resolved this `path_flag`; `None` when - nothing was filtered, or when the gate that produced this `path_flag` doesn't - build one yet. + nothing was filtered. Return: The generated id, or `None` when there's nothing to write, including when the @@ -271,9 +270,11 @@ class PredictionProcess: """ if not document: # Skip the remote activity call entirely rather than dispatching it just to have - # `write_prediction_metadata` decide there's nothing to write — `mlflow_content_gate` - # doesn't build a `document` yet, so this is the common case today, not a rare edge - # case (review finding, 2026-10-01). + # `write_prediction_metadata` decide there's nothing to write (review finding, + # 2026-10-01). All 3 gates now build a `document` whenever they return a non-`None` + # `path_flag`, so this branch is a defensive fallback today, not the common case it + # was when this was first written — kept in case a future caller passes a `path_flag` + # without one. return None try: diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index d7ef33f..b3aebe1 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -6,7 +6,13 @@ from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.metrics_controller import MetricsController from sientia_do.observability.sientia_monitoring import SientiaMonitoring -from laborious.activities.gates import Gates, input_filter_counters, input_filter_functions +from laborious.activities.gates import ( + Gates, + input_filter_counters, + input_filter_functions, + mlflow_content_filter_counters, + mlflow_content_filter_functions, +) @fixture(autouse=True) @@ -469,7 +475,7 @@ def test_mlflow_content_gate_invalid_filter(gates_activity): result = gates_activity.mlflow_content_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) @patch('laborious.activities.gates.mlflow_content_filter_functions') @@ -491,7 +497,7 @@ def test_mlflow_content_gate_filter_exception(mock_mlflow_content_filter_functio result = gates_activity.mlflow_content_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.debug.assert_called() gates_activity.send_notification.assert_called_once_with( metadata=metadata['metadata'], @@ -517,12 +523,12 @@ def test_mlflow_content_gate_no_filters(gates_activity): result = gates_activity.mlflow_content_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.debug.assert_called() def test_mlflow_content_gate_with_filter(gates_activity): - # Arrange + # Arrange: 3 rows, all-NaN after the filter's own normalization — triggers NAN_VALUES. input_data = { **metadata, 'filters': {'NAN_VALUES': {'POLICY': 'STOP', 'CONFIG': {}}}, @@ -532,10 +538,15 @@ def test_mlflow_content_gate_with_filter(gates_activity): } # Act - result = gates_activity.mlflow_content_gate(input_data) + path_flag, confidence, comment, document = gates_activity.mlflow_content_gate(input_data) # Assert - assert result == ('STOP', -1, 'Transformed data not passed the content filter') + assert (path_flag, confidence, comment) == ( + 'STOP', + -1, + 'Transformed data not passed the content filter', + ) + assert document == {'filters': [{'name': 'NAN_VALUES', 'policy': 'STOP', 'counter': 3}]} gates_activity.debug.assert_called() gates_activity.send_notification.assert_called() @@ -554,7 +565,7 @@ def test_mlflow_content_gate_with_filter_not_caught(gates_activity): result = gates_activity.mlflow_content_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.debug.assert_called() @@ -569,10 +580,18 @@ def test_mlflow_content_gate_filter_returns_false(gates_activity): result = gates_activity.mlflow_content_gate(input_data) - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.debug.assert_called() +def test_mlflow_content_filter_counters_covers_every_mlflow_content_filter_function(): + # Same parity guard as `test_input_filter_counters_covers_every_input_filter_function` + # (review finding, 2026-09-30): a filter added to `mlflow_content_filter_functions` without + # a matching counter here degrades to counter=0 instead of crashing, but this test still + # catches the gap before it ships. + assert set(mlflow_content_filter_counters) == set(mlflow_content_filter_functions) + + def test_get_prediction_store_policy_invalid_policy(gates_activity): # Arrange prediction_store_policy = 'INVALID_POLICY' diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 040cac8..9b01022 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -61,7 +61,7 @@ async def test_run(workflow_mock, prediction_process): # mlflow_response_gate (transform) ('continue', 0.95, 'Error', None), # mlflow_content_gate (transform) - ('continue', 0.95, 'Transformed data not passed the content filter'), + ('continue', 0.95, 'Transformed data not passed the content filter', None), # mlflow_response_gate (predict) ('continue', 0.95, 'Error', None), ] @@ -369,7 +369,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process # mlflow_response_gate (transform) ('continue', 0.95, 'Error', None), # mlflow_content_gate (transform) - ('continue', 0.95, 'Transformed data not passed the content filter'), + ('continue', 0.95, 'Transformed data not passed the content filter', None), ] # Act @@ -480,7 +480,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p # mlflow_response_gate (transform) ('continue', 0.95, 'Error', None), # mlflow_content_gate (transform) - ('continue', 0.95, 'Transformed data not passed the content filter'), + ('continue', 0.95, 'Transformed data not passed the content filter', None), ('continue', 0.95, 'Error', None), # mlflow_response_gate (predict) ] @@ -695,10 +695,10 @@ async def test_path_flag_handler_repeat(workflow_mock, prediction_process): async def test_path_flag_handler_repeat_no_document_skips_write_activity_call( workflow_mock, prediction_process ): - # Arrange: no document (e.g. REPEAT triggered by a gate that doesn't build one yet — today, - # `mlflow_content_gate`) must not even dispatch the remote write_prediction_metadata - # activity — that would be a wasted Temporal round-trip on every REPEAT caused by that - # gate, not a rare edge case (review finding, 2026-10-01). + # Arrange: no document must not even dispatch the remote write_prediction_metadata + # activity — that would be a wasted Temporal round-trip (review finding, 2026-10-01). + # Defensive case: all 3 gates now always build a document alongside a non-`None` + # `path_flag`, so this guards a caller that doesn't, not today's common case. data = {'test': 'data'} # Act @@ -1076,7 +1076,7 @@ async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process): workflow_mock.execute_local_activity_method.side_effect = [ ('continue', 0.95, 'ok', None), # input_gate ('continue', 0.95, '', None), # mlflow_response_gate (transform) - ('continue', 0.95, ''), # mlflow_content_gate (transform) + ('continue', 0.95, '', None), # mlflow_content_gate (transform) ('continue', 0.95, '', None), # mlflow_response_gate (predict) ] From 99392034e5c67abb78c5c29180dc4630687da35b Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 8 Oct 2026 13:51:09 +0000 Subject: [PATCH 5/9] chore: bump version to 0.9.0-rc3 --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index b808703..02dfded 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "laborious" -version = "0.9.0-rc2" +version = "0.9.0-rc3" description = "Sientia DataOps Laborious - ML Model Orchestration System" readme = "README.md" requires-python = ">=3.11" From 9f218af6b03e80f3a81fbbb462a9218c331c1ff2 Mon Sep 17 00:00:00 2001 From: Carla Pinto Date: Thu, 8 Oct 2026 16:43:18 +0200 Subject: [PATCH 6/9] rc/SIENTIAPDE-2167- test: add e2e coverage for metadata_id reaching predictions.metadata_id --- e2e/conftest.py | 3 +- ...st_predictions_batch_prediction_process.py | 71 +++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/e2e/conftest.py b/e2e/conftest.py index 807a5ce..0295326 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -368,6 +368,7 @@ def _worker_activity_list(test_activities: Activities): test_activities.mlflow_content_gate, test_activities.request_predict, test_activities.repeat_last_prediction, + test_activities.write_prediction_metadata, test_activities.format_prediction, test_activities.format_transformed_data, test_activities.format_default_prediction, @@ -624,5 +625,3 @@ async def temporal_worker_real_opc(temporal_test_env, test_activities_real_opc): activity_executor=activity_executor, ) as worker: yield worker - - diff --git a/e2e/test_predictions_batch_prediction_process.py b/e2e/test_predictions_batch_prediction_process.py index f15621a..c3f4767 100644 --- a/e2e/test_predictions_batch_prediction_process.py +++ b/e2e/test_predictions_batch_prediction_process.py @@ -8,6 +8,7 @@ from unittest.mock import MagicMock, patch import numpy as np import pandas as pd import pytest +from pymongo import MongoClient from sqlalchemy import text from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker @@ -86,6 +87,76 @@ async def test_scenario_2_1_1_input_gate_triggers_continue( mlflow_repository_stub.stub_wrapper.predict.assert_not_called() +@pytest.mark.asyncio +@pytest.mark.integration +async def test_e2e_metadata_id_reaches_postgres_when_gate_triggers( + temporal_test_env: WorkflowEnvironment, + temporal_worker: Worker, + test_activities: Activities, + postgres_engine, + mongo_container, + mlflow_repository_stub, +): + """ + input_gate CONTINUE: the gate's document is written to Mongo, and the resulting + metadata_id lands on the predictions row — independently verified on both sides. + """ + client = temporal_test_env.client + model_id = 261 + # sensor_1 NULL, sensor_2 present — matches `input_filters.SPECIFIC_VARIABLES_NULL_VALUES` + # (`variables: ["sensor_1"]`, POLICY CONTINUE) in prediction_process_base.json. + insert_sample_data(postgres_engine, model_id, ['NULL', 78.2]) + input_data = get_base_input_data(model_id) + await start_and_await_workflow( + client, PredictionsBatch.run, input_data, make_workflow_id('test-metadata-id-continue') + ) + assert_continue(postgres_engine, model_id) + + with postgres_engine.connect() as conn: + metadata_id = conn.execute( + text(f'SELECT metadata_id FROM sientia_data.predictions WHERE model_id = {model_id}') + ).scalar() + assert metadata_id is not None, 'Expected a metadata_id on the predictions row' + + # Independent readback: a fresh client against the same testcontainer, not the + # `mongo_repository`/`PredictionMetadata` instances the workflow itself wrote through. + mongo_port = mongo_container.get_exposed_port(27017) + verify_client = MongoClient(f'mongodb://localhost:{mongo_port}') + try: + document = verify_client['test_db']['prediction_metadata'].find_one({'_id': metadata_id}) + finally: + verify_client.close() + assert document is not None, f'Expected a prediction_metadata document for id {metadata_id}' + assert document['filters'] == [ + {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 1} + ] + + +@pytest.mark.asyncio +@pytest.mark.integration +async def test_e2e_metadata_id_is_null_when_nothing_triggers( + temporal_test_env: WorkflowEnvironment, + temporal_worker: Worker, + test_activities: Activities, + postgres_engine, +): + """Clean prediction, no gate triggers: predictions.metadata_id stays NULL.""" + client = temporal_test_env.client + model_id = 262 + insert_sample_data(postgres_engine, model_id, [60.0, 78.2]) + input_data = get_base_input_data(model_id) + await start_and_await_workflow( + client, PredictionsBatch.run, input_data, make_workflow_id('test-metadata-id-none') + ) + assert_prediction(postgres_engine, model_id, prediction=0.5, prediction_confidence=0) + + with postgres_engine.connect() as conn: + metadata_id = conn.execute( + text(f'SELECT metadata_id FROM sientia_data.predictions WHERE model_id = {model_id}') + ).scalar() + assert metadata_id is None, f'Expected metadata_id to be NULL, got {metadata_id!r}' + + @pytest.mark.asyncio @pytest.mark.integration async def test_scenario_2_1_2_input_gate_triggers_stop( From 1174bb3f542484438c06abccc2365d6d6e36424b Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 8 Oct 2026 17:14:22 +0000 Subject: [PATCH 7/9] chore: bump version to 0.9.0-rc4 --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 02dfded..ecee217 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "laborious" -version = "0.9.0-rc3" +version = "0.9.0-rc4" description = "Sientia DataOps Laborious - ML Model Orchestration System" readme = "README.md" requires-python = ">=3.11" From 56baae78b69e11aaec54fd7cb2df3628ede537ee Mon Sep 17 00:00:00 2001 From: Carla Pinto Date: Fri, 9 Oct 2026 14:39:46 +0200 Subject: [PATCH 8/9] =?UTF-8?q?rc/SIENTIAPDE-2167-=20fix:=20corrige=20imag?= =?UTF-8?q?em=20MinIO=20indispon=C3=ADvel=20e=20cobre=20metadata=5Fid=20no?= =?UTF-8?q?s=20testes=20e2e?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- e2e/conftest.py | 18 ++++++++++++++++-- ...est_predictions_batch_prediction_process.py | 6 +++--- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/e2e/conftest.py b/e2e/conftest.py index 0295326..b9c9591 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -48,8 +48,22 @@ def postgres_container(): @pytest_asyncio.fixture(scope='session') def minio_container(): - """MinIO testcontainer used by E2E offload and payload retrieval paths.""" - minio = MinioContainer() + """ + MinIO testcontainer used by E2E offload and payload retrieval paths. + + `MinioContainer`'s default image (`minio/minio:RELEASE.2022-12-02T19-19-22Z`) no longer + pulls — Docker Hub removed the `minio/minio` repository entirely after MinIO's licensing + change. Production (`sientia-helm-chart`) already moved to Bitnami's post-Broadcom-migration + "legacy" namespace; mirroring that image here. Two more adjustments on top of just the image: + the image runs as non-root by default and can't write to the container's anonymous volume + without `user='root'`; and Bitnami deprecated `MINIO_ACCESS_KEY`/`MINIO_SECRET_KEY` (what + `MinioContainer` sets) in favor of `MINIO_ROOT_USER`/`MINIO_ROOT_PASSWORD` — reusing the same + values so `get_client()` still works unmodified. + """ + minio = MinioContainer(image='docker.io/bitnamilegacy/minio:2025.7.23-debian-12-r3') + minio.with_env('MINIO_ROOT_USER', minio.access_key) + minio.with_env('MINIO_ROOT_PASSWORD', minio.secret_key) + minio.with_kwargs(user='root') minio.start() yield minio minio.stop() diff --git a/e2e/test_predictions_batch_prediction_process.py b/e2e/test_predictions_batch_prediction_process.py index c3f4767..9925e18 100644 --- a/e2e/test_predictions_batch_prediction_process.py +++ b/e2e/test_predictions_batch_prediction_process.py @@ -275,7 +275,7 @@ async def test_scenario_2_2_1_transform_gate_triggers_continue( postgres_engine=postgres_engine, model_id=model_id, prediction_confidence=Decimal(10), - comments='Unknown MLFlow API error', + comments='Bad data model', ) @@ -403,7 +403,7 @@ async def test_scenario_2_3_1_predict_gate_triggers_continue( postgres_engine=postgres_engine, model_id=model_id, prediction_confidence=Decimal(10), - comments='Unknown MLFlow API error', + comments='Bad predict model', ) @@ -521,7 +521,7 @@ async def test_scenario_2_4_1_priority_conflict_resolution( postgres_engine=postgres_engine, model_id=model_id, prediction_confidence=Decimal(10), - comments='Unknown MLFlow API error', + comments='Bad data model', ) From 693efbb72ba3c156d417fef24197e26a7af53aae Mon Sep 17 00:00:00 2001 From: Carla Pinto Date: Fri, 9 Oct 2026 14:55:42 +0200 Subject: [PATCH 9/9] =?UTF-8?q?rc/SIENTIAPDE-2167-=20fix:=20corrige=20arte?= =?UTF-8?q?fato=20de=20refer=C3=AAncia=20desatualizado=20no=20teste=20e2e?= =?UTF-8?q?=20de=20drift?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- e2e/test_drift.py | 5 +++-- laborious/activities/mlflow.py | 8 ++++---- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/e2e/test_drift.py b/e2e/test_drift.py index 7941bf5..f52b910 100644 --- a/e2e/test_drift.py +++ b/e2e/test_drift.py @@ -142,7 +142,7 @@ def _recent_minute_timestamps(count: int, offset_minutes: int = 6) -> list[str]: def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFrame) -> None: """ Wire ``mlflow_repository_stub`` so ``get_reference_data`` returns - ``reference_rows`` by writing them to ``dst_path/retrain_input.csv``. + ``reference_rows`` by writing them to ``dst_path/evaluation_data.csv``. Args: - mlflow_repository_stub: External MLflow repository fixture. @@ -158,7 +158,8 @@ def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFram run_id='fake-reference-run' ) file_info = MagicMock() - file_info.path = 'retrain_input.csv' + # Matches `MLFlow._REFERENCE_ARTIFACT_CANDIDATES` (SIENTIAPDE-1646). + file_info.path = 'evaluation_data.csv' mlflow_repository_stub._client.list_artifacts.return_value = [file_info] mlflow_repository_stub.download_artifacts.side_effect = _download diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 80d007d..b985cfd 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -664,7 +664,7 @@ class MLFlow(SientiaMonitoring): """ Pick the first available reference CSV artifact path from the MLflow run. - Candidates are checked in priority order: ``retrain_input.csv``, then ``train_data.csv``. + Candidates are checked in priority order: ``evaluation_data.csv``, then ``test_data.csv``. A path matches when it equals the candidate or ends with ``/`` for nested layouts. Args: @@ -703,9 +703,9 @@ class MLFlow(SientiaMonitoring): """ Download reference training CSV from the MLflow run linked to the production alias. - Resolves ``retrain_input.csv`` or ``train_data.csv`` via artifact listing before download. - ``retrain_input.csv`` is preferred when both exist (most recent retrain snapshot). Used by - drift workflows to compare live data against the reference distribution logged with the model. + Resolves ``evaluation_data.csv`` or ``test_data.csv`` via artifact listing before download. + ``evaluation_data.csv`` is preferred when both exist. Used by drift workflows to compare + live data against the reference distribution logged with the model. Timestamps are normalized to ``DATETIME_FORMAT`` string columns before returning records. Args: