gitea_admin/sientia-dataops-laborious-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rc/SIENTIAPDE-2165- feat: rich metadata (filter/policy/counter) for mlflow_response_gate

Browse Source
This commit is contained in:
Carla Pinto
2026-10-05 22:14:18 +02:00
parent 14b27e4c30
commit 0742078fad
4 changed files with 112 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
laborious/activities/gates.py
View File

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

14
laborious/workflows/sub_workflows/prediction_process.py
View File

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

36
tests/laborious/activities/test_gates.py
View File

@@ -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 = {

24
tests/laborious/workflows/subworkflows/test_prediction_process.py
View File

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