feature-rc/SIENTIAPDE-2164 -Enrich input_gate metadata (filter/policy/counter) and write to Mongo on CONTINUE

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

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,