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) ]