rc/SIENTIAPDE-2166- feat: rich metadata (filter/policy/counter) for mlflow_content_gate

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]

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:

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'

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