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)