diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 71d19cc..28e63d5 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -29,6 +29,7 @@ with workflow.unsafe.imports_passed_through(): 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] # Input filter function mappings input_filter_functions: dict[str, InputFilterFunc] = { @@ -36,6 +37,42 @@ input_filter_functions: dict[str, InputFilterFunc] = { 'EMPTY_DATA': filter_empty_data, } + +def _count_specific_variables_null_values(data: DataFrame, config: dict[str, Any]) -> int: + """ + Count rows with a null value in one of the configured variables. + + Mirrors `filter_specific_variables_null_values`'s own trigger condition, so the count + reflects exactly what made that filter fire. + + Return: + int: Number of rows with a null value in `config['variables']`. + """ + if data.empty: + return 0 + return int((data['variable'].isin(config['variables']) & data['value'].isna()).sum()) + + +def _count_empty_data(_data: DataFrame, _config: dict[str, Any]) -> int: + """ + Count rows for the EMPTY_DATA filter. + + `EMPTY_DATA` only triggers when there are no rows at all, so the count is always 0. + + Return: + int: Always 0. + """ + return 0 + + +# Per-filter row counts for the `counter` field in prediction metadata (gate-local: mirrors each +# filter's own trigger condition instead of changing `conditional_filters.py`'s bool contract, +# which `mlflow_content_gate` also depends on for EMPTY_DATA). +input_filter_counters: dict[str, InputCounterFunc] = { + 'SPECIFIC_VARIABLES_NULL_VALUES': _count_specific_variables_null_values, + 'EMPTY_DATA': _count_empty_data, +} + # Confidence mappings kept separate from function maps to avoid Union types input_path_confidence: Mapping[str, int] = { 'STOP': -1, @@ -183,6 +220,14 @@ class Gates(SientiaMonitoring): codebase's convention of string `_id`s (e.g. the `models` collection) and the 24-hex-char size of the `predictions.metadata_id` Postgres column that will reference it. + Caller contract (not enforced here — no caller exists yet in this codebase): only call + this when the gate's resolved `path_flag` is `CONTINUE`. `STOP` terminates the workflow + before any `predictions` row is written (`prediction_process.py`'s `path_flag_handler`); + `REPEAT` reuses the last existing row instead of writing a new one. In both cases, a + document written here would be orphaned — no `predictions.metadata_id` will ever + reference it. Only `CONTINUE` reaches `format_and_export_prediction`, the one path that + writes a new `predictions` row. + Args: document: Metadata payload (filter, policy, counter, etc.); `None`/empty skips the write entirely. No `_id` key expected. @@ -226,7 +271,7 @@ class Gates(SientiaMonitoring): return policy, filter_config @activity.defn(name='input_gate') - def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]: + def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str, str | None]: """ Apply input data quality filters and validation. @@ -241,6 +286,8 @@ class Gates(SientiaMonitoring): 3. Determines appropriate path decisions (STOP, CONTINUE, REPEAT) 4. Provides confidence scores and detailed comments 5. Handles errors gracefully with notification integration + 6. When the resolved path is CONTINUE, best-effort writes which filter(s) triggered, + the policy and a counter to `prediction_metadata` (see `_write_prediction_metadata`) Args: input_data: Configuration and data for input validation @@ -251,10 +298,14 @@ class Gates(SientiaMonitoring): - path_priority (list[str]): Priority order for path decisions Returns: - tuple: (path_flag, confidence, comment) + tuple: (path_flag, confidence, comment, metadata_id) - 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 + - metadata_id (str | None): Id of the `prediction_metadata` document written for + this gate's decision; `None` unless `path_flag` is `CONTINUE` (STOP/REPEAT + never write a new `predictions` row, so writing metadata for them would + orphan it — see `_write_prediction_metadata`'s caller contract) Raises: Exception: If filter execution fails or configuration is invalid @@ -268,7 +319,8 @@ class Gates(SientiaMonitoring): data = payload.retrieve(self.minio_repository, metadata) 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: {filters}', metadata) @@ -282,7 +334,7 @@ class Gates(SientiaMonitoring): try: if input_filter_functions[fil](data, filter_config): self.debug(f'Data not passed the input filter {fil}:{config}', metadata) - filter_output.append(policy) + triggered_filters.append((fil, policy, filter_config)) except Exception as e: trace = traceback.format_exc() self.send_notification( @@ -294,16 +346,45 @@ 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'Input gate result: {path_flag}', metadata) - return path_flag, input_path_confidence[path_flag], 'Input data with bad quality' + + matched = [ + (fil, filter_config) + for fil, policy, filter_config in triggered_filters + if policy == path_flag + ] + document = { + 'filters': [ + { + 'name': fil, + 'policy': path_flag, + 'counter': input_filter_counters[fil](data, filter_config), + } + for fil, filter_config in matched + ], + } + metadata_id = ( + self._write_prediction_metadata(document, metadata) + if path_flag == 'CONTINUE' + else None + ) + + return ( + path_flag, + input_path_confidence[path_flag], + 'Input data with bad quality', + metadata_id, + ) self.info('Nothing was filtered by the input gate', metadata) del data - return None, 0, '' + 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]: diff --git a/laborious/workflows/sub_workflows/prediction_process.py b/laborious/workflows/sub_workflows/prediction_process.py index 0f86fa1..391be1c 100644 --- a/laborious/workflows/sub_workflows/prediction_process.py +++ b/laborious/workflows/sub_workflows/prediction_process.py @@ -132,7 +132,9 @@ class PredictionProcess: 'path_priority': input_data['path_priority'], } - path_flag, confidence, comment = await workflow.execute_local_activity_method( + # `_metadata_id` (prediction_metadata document id, only non-None when path_flag ends up + # CONTINUE) isn't threaded through yet — SIENTIAPDE-2167 wires it to Postgres. + path_flag, confidence, comment, _metadata_id = await workflow.execute_local_activity_method( Activities.input_gate, gate_input, retry_policy=retry_policy, diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index 80fc28f..7ba7111 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -136,7 +136,7 @@ def test_write_prediction_metadata_insert_failure_swallowed(gates_activity, mong gates_activity.error.assert_called_once() -def test_input_gate_invalid_filter(gates_activity): +def test_input_gate_invalid_filter(gates_activity, mongo: MagicMock): # Arrange input_data = { **metadata, @@ -149,10 +149,11 @@ def test_input_gate_invalid_filter(gates_activity): result = gates_activity.input_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.error.assert_called_once_with( 'Filter INVALID_FILTER not found', metadata['metadata'] ) + mongo.insert.assert_not_called() @patch('laborious.activities.gates.input_filter_functions') @@ -173,7 +174,7 @@ def test_input_gate_filter_exception(mock_input_filter_functions, gates_activity result = gates_activity.input_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='INTPUT_GATE_ERROR__EMPTY_DATA', @@ -184,7 +185,7 @@ def test_input_gate_filter_exception(mock_input_filter_functions, gates_activity ) -def test_input_gate_no_filters(gates_activity): +def test_input_gate_no_filters(gates_activity, mongo: MagicMock): # Arrange input_data = { **metadata, @@ -197,12 +198,15 @@ def test_input_gate_no_filters(gates_activity): result = gates_activity.input_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.debug.assert_called() + mongo.insert.assert_not_called() -def test_input_gate_with_filter(gates_activity): - # Arrange +def test_input_gate_with_filter(gates_activity, mongo: MagicMock): + # Arrange: STOP never writes to Mongo — see business rule refinement (2026-09-29): STOP + # terminates the workflow before any `predictions` row is written, so a metadata document + # written here would be orphaned. input_data = { **metadata, 'filters': {'EMPTY_DATA': {'POLICY': 'STOP', 'CONFIG': {}}}, @@ -214,8 +218,9 @@ def test_input_gate_with_filter(gates_activity): result = gates_activity.input_gate(input_data) # Assert - assert result == ('STOP', -1, 'Input data with bad quality') + assert result == ('STOP', -1, 'Input data with bad quality', None) gates_activity.debug.assert_called() + mongo.insert.assert_not_called() def test_input_gate_with_filter_lowercase_keys(gates_activity): @@ -231,7 +236,7 @@ def test_input_gate_with_filter_lowercase_keys(gates_activity): result = gates_activity.input_gate(input_data) # Assert - assert result == ('STOP', -1, 'Input data with bad quality') + assert result == ('STOP', -1, 'Input data with bad quality', None) def test_input_gate_with_filter_capitalized_keys(gates_activity): @@ -247,10 +252,10 @@ def test_input_gate_with_filter_capitalized_keys(gates_activity): result = gates_activity.input_gate(input_data) # Assert - assert result == ('STOP', -1, 'Input data with bad quality') + assert result == ('STOP', -1, 'Input data with bad quality', None) -def test_input_gate_with_filter_not_caught(gates_activity): +def test_input_gate_with_filter_not_caught(gates_activity, mongo: MagicMock): # Arrange input_data = { **metadata, @@ -263,8 +268,99 @@ def test_input_gate_with_filter_not_caught(gates_activity): result = gates_activity.input_gate(input_data) # Assert - assert result == (None, 0, '') + assert result == (None, 0, '', None) gates_activity.debug.assert_called() + mongo.insert.assert_not_called() + + +def test_input_gate_repeat_does_not_write_prediction_metadata(gates_activity, mongo: MagicMock): + # Arrange: REPEAT reuses the last existing row instead of writing a new one, so it must not + # write metadata either — same orphaning risk as STOP. + input_data = { + **metadata, + 'filters': {'EMPTY_DATA': {'POLICY': 'REPEAT', 'CONFIG': {}}}, + 'data': _minio_payload(DataFrame({'value': []})), + 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], + } + + # Act + result = gates_activity.input_gate(input_data) + + # Assert + assert result == ('REPEAT', -1, 'Input data with bad quality', None) + mongo.insert.assert_not_called() + + +def test_input_gate_continue_writes_prediction_metadata(gates_activity, mongo: MagicMock): + # Arrange: two rows with a null in one of the configured variables ('x'); a third row is + # clean, to confirm the counter reflects only the rows that actually matched the condition. + input_data = { + **metadata, + 'filters': { + 'SPECIFIC_VARIABLES_NULL_VALUES': { + 'POLICY': 'CONTINUE', + 'CONFIG': {'variables': ['x']}, + } + }, + 'data': _minio_payload( + DataFrame({'variable': ['x', 'x', 'y'], 'value': [None, None, None]}) + ), + 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], + } + + # Act + path_flag, confidence, comment, metadata_id = gates_activity.input_gate(input_data) + + # Assert + assert (path_flag, confidence, comment) == ('CONTINUE', 2, 'Input data with bad quality') + assert metadata_id is not None + collection, inserted_document, inserted_metadata = mongo.insert.call_args.args + assert collection == Gates._PREDICTION_METADATA_COLLECTION + assert inserted_document == { + 'filters': [ + {'name': 'SPECIFIC_VARIABLES_NULL_VALUES', 'policy': 'CONTINUE', 'counter': 2} + ], + '_id': metadata_id, + } + assert inserted_metadata == metadata['metadata'] + + +def test_input_gate_continue_multiple_filters_same_policy_lists_each( + gates_activity, mongo: MagicMock +): + # Arrange: force both input filters to trigger (not reachable with today's real filter + # conditions — EMPTY_DATA and SPECIFIC_VARIABLES_NULL_VALUES are mutually exclusive) to lock + # in the list-building behavior approved by Vitor (2026-09-30): one entry per filter sharing + # the winning policy, instead of a single joined/aggregated value. + with patch('laborious.activities.gates.input_filter_functions') as mock_filters: + mock_filters.__contains__.return_value = True + mock_filters.__getitem__.return_value = MagicMock(return_value=True) + + input_data = { + **metadata, + 'filters': { + 'SPECIFIC_VARIABLES_NULL_VALUES': { + 'POLICY': 'CONTINUE', + 'CONFIG': {'variables': ['x']}, + }, + 'EMPTY_DATA': {'POLICY': 'CONTINUE', 'CONFIG': {}}, + }, + 'data': _minio_payload(DataFrame({'variable': ['x'], 'value': [None]})), + 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], + } + + # Act + path_flag, _, _, metadata_id = gates_activity.input_gate(input_data) + + # Assert + assert path_flag == 'CONTINUE' + assert metadata_id is not None + _, inserted_document, _ = mongo.insert.call_args.args + assert {f['name'] for f in inserted_document['filters']} == { + 'SPECIFIC_VARIABLES_NULL_VALUES', + 'EMPTY_DATA', + } + assert all(f['policy'] == 'CONTINUE' for f in inserted_document['filters']) def test_mlflow_response_gate_invalid_filter(gates_activity): diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 6f431e8..cc84661 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -57,7 +57,7 @@ async def test_run(workflow_mock, prediction_process): MagicMock(), ] workflow_mock.execute_local_activity_method.side_effect = [ - ('continue', 0.95, 'Input data with bad quality'), # input_gate + ('continue', 0.95, 'Input data with bad quality', None), # input_gate # mlflow_response_gate (transform) ('continue', 0.95, 'Error'), # mlflow_content_gate (transform) @@ -217,7 +217,7 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process): # Mock the activity responses workflow_mock.execute_local_activity_method.side_effect = [ - ('stop', 0.95, 'Input data with bad quality'), # input_gate + ('stop', 0.95, 'Input data with bad quality', None), # input_gate ] # Act @@ -275,7 +275,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ MagicMock(), ] workflow_mock.execute_local_activity_method.side_effect = [ - ('repeat', 0.95, 'Input data with bad quality'), # input_gate + ('repeat', 0.95, 'Input data with bad quality', None), # input_gate ('continue', 0.95, 'Error'), # mlflow_response_gate (transform) ] @@ -365,7 +365,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process MagicMock(), ] workflow_mock.execute_local_activity_method.side_effect = [ - ('continue', 0.95, 'Input data with bad quality'), # input_gate + ('continue', 0.95, 'Input data with bad quality', None), # input_gate # mlflow_response_gate (transform) ('continue', 0.95, 'Error'), # mlflow_content_gate (transform) @@ -476,7 +476,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p MagicMock(), ] workflow_mock.execute_local_activity_method.side_effect = [ - ('continue', 0.95, 'Input data with bad quality'), # input_gate + ('continue', 0.95, 'Input data with bad quality', None), # input_gate # mlflow_response_gate (transform) ('continue', 0.95, 'Error'), # mlflow_content_gate (transform) @@ -850,7 +850,7 @@ async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process): MagicMock(), ] workflow_mock.execute_local_activity_method.side_effect = [ - ('continue', 0.95, 'ok'), + ('continue', 0.95, 'ok', None), # input_gate ('continue', 0.95, ''), ('continue', 0.95, ''), ('continue', 0.95, ''),