6 Commits

5 changed files with 103 additions and 41 deletions

View File

@@ -78,6 +78,7 @@ input_path_confidence: Mapping[str, int] = {
'REPEAT': -1, 'REPEAT': -1,
} }
# MLFlow response filter function mappings # MLFlow response filter function mappings
mlflow_response_filter_functions: dict[str, ResponseFilterFunc] = { mlflow_response_filter_functions: dict[str, ResponseFilterFunc] = {
'API_ERROR': api_error_filter, 'API_ERROR': api_error_filter,
@@ -320,7 +321,9 @@ class Gates(SientiaMonitoring):
return None, 0, '', None return None, 0, '', None
@activity.defn(name='mlflow_response_gate') @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. Validate MLFlow API response quality and integrity.
@@ -334,6 +337,9 @@ class Gates(SientiaMonitoring):
3. Determines path decisions based on response validation results 3. Determines path decisions based on response validation results
4. Provides confidence scores and detailed validation comments 4. Provides confidence scores and detailed validation comments
5. Handles API errors and response validation failures 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: Args:
input_data: Configuration and data for response validation input_data: Configuration and data for response validation
@@ -345,10 +351,14 @@ class Gates(SientiaMonitoring):
- path_priority (list[str]): Priority order for path decisions - path_priority (list[str]): Priority order for path decisions
Returns: Returns:
tuple: (path_flag, confidence, comment) tuple: (path_flag, confidence, comment, document)
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None) - path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
- confidence (int): Confidence score for the decision - confidence (int): Confidence score for the decision
- comment (str): Detailed explanation of 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: Raises:
Exception: If response validation fails or configuration is invalid Exception: If response validation fails or configuration is invalid
@@ -369,10 +379,11 @@ class Gates(SientiaMonitoring):
gate_type = input_data['type'] gate_type = input_data['type']
path_priority = input_data['path_priority'] path_priority = input_data['path_priority']
filter_output = []
comments = [] 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 {} status = payload.status or {}
for fil, config in filters.items(): for fil, config in filters.items():
@@ -381,15 +392,21 @@ class Gates(SientiaMonitoring):
policy, filter_config = self._read_filter_entry(config) policy, filter_config = self._read_filter_entry(config)
try: try:
if mlflow_response_filter_functions[fil](status, filter_config): 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')) # The real error lives one level in, under `content` — `request_transform`/
# `request_predict` nest `message`/`traceback` there on failure (mlflow.py),
# reusing the same `content` key the success path uses for the DataFrame
# (review finding, 2026-10-08: reading `status['message']` directly always
# fell back to the generic default, in every real production occurrence).
error_content = status.get('content') or {}
comments.append(error_content.get('message', 'Unknown MLFlow API error'))
self.send_notification( self.send_notification(
metadata=metadata, metadata=metadata,
notification_id=f'{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}', notification_id=f'{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}',
message=status.get('message', 'Unknown MLFlow API error'), message=error_content.get('message', 'Unknown MLFlow API error'),
block='mlflow_gate', block='mlflow_gate',
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
attachment_content=status.get('traceback'), attachment_content=error_content.get('traceback'),
) )
except Exception as e: except Exception as e:
trace = traceback.format_exc() trace = traceback.format_exc()
@@ -402,16 +419,42 @@ class Gates(SientiaMonitoring):
attachment_content=trace, attachment_content=trace,
) )
triggered_policies = [policy for _, policy, _ in triggered_filters]
for path_flag in path_priority: 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) self.info(f'Mlflow response gate result: {path_flag}', metadata)
return path_flag, mlflow_response_path_confidence[path_flag], ', '.join(comments)
matched = [fil for fil, policy, _ in triggered_filters if policy == path_flag]
document = {
'filters': [
{
'name': fil,
'policy': path_flag,
# No per-row/per-item unit to count here — `api_error_filter`
# evaluates a single API response as a whole, not tabular data —
# so, unlike `input_gate`'s per-filter counters, this is a fixed
# value: 1 (one API error event). Inlined rather than a counter
# function/dict, since there's only one response filter today and
# this isn't an override of anything (review finding).
'counter': 1,
}
for fil 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) self.info('Nothing was filtered by the mlflow response gate', metadata)
del data del data
return None, 0, '' return None, 0, '', None
@activity.defn(name='mlflow_content_gate') @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]:

View File

@@ -156,7 +156,7 @@ class PredictionProcess:
) )
# Validate MLFlow transform response # 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, Activities.mlflow_response_gate,
{ {
**metadata, **metadata,
@@ -171,7 +171,7 @@ class PredictionProcess:
# Handle path decision based on transform validation # Handle path decision based on transform validation
if await self.path_flag_handler( 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 return
@@ -206,7 +206,7 @@ class PredictionProcess:
) )
# Validate MLFlow prediction response # 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, Activities.mlflow_response_gate,
{ {
**metadata, **metadata,
@@ -221,7 +221,7 @@ class PredictionProcess:
# Handle path decision based on prediction validation # Handle path decision based on prediction validation
if await self.path_flag_handler( 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 return
@@ -271,9 +271,9 @@ class PredictionProcess:
""" """
if not document: if not document:
# Skip the remote activity call entirely rather than dispatching it just to have # 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` # `write_prediction_metadata` decide there's nothing to write — `mlflow_content_gate`
# and `mlflow_content_gate` don't build a `document` yet, so this is the common case # doesn't build a `document` yet, so this is the common case today, not a rare edge
# today, not a rare edge case (review finding, 2026-10-01). # case (review finding, 2026-10-01).
return None return None
try: try:

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "laborious" name = "laborious"
version = "0.9.0-rc2" version = "0.9.0-rc3"
description = "Sientia DataOps Laborious - ML Model Orchestration System" description = "Sientia DataOps Laborious - ML Model Orchestration System"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"

View File

@@ -314,7 +314,7 @@ def test_mlflow_response_gate_invalid_filter(gates_activity):
result = gates_activity.mlflow_response_gate(input_data) result = gates_activity.mlflow_response_gate(input_data)
# Assert # Assert
assert result == (None, 0, '') assert result == (None, 0, '', None)
@patch('laborious.activities.gates.mlflow_response_filter_functions') @patch('laborious.activities.gates.mlflow_response_filter_functions')
@@ -341,7 +341,7 @@ def test_mlflow_response_gate_filter_exception(
result = gates_activity.mlflow_response_gate(input_data) result = gates_activity.mlflow_response_gate(input_data)
# Assert # Assert
assert result == (None, 0, '') assert result == (None, 0, '', None)
gates_activity.send_notification.assert_called_once_with( gates_activity.send_notification.assert_called_once_with(
metadata=metadata['metadata'], metadata=metadata['metadata'],
notification_id='MLFLOW_GATE_RESPONSE_FILTER__INVALID_FILTER', notification_id='MLFLOW_GATE_RESPONSE_FILTER__INVALID_FILTER',
@@ -369,7 +369,7 @@ def test_mlflow_response_gate_no_filters(gates_activity):
result = gates_activity.mlflow_response_gate(input_data) result = gates_activity.mlflow_response_gate(input_data)
# Assert # Assert
assert result == (None, 0, '') assert result == (None, 0, '', None)
gates_activity.debug.assert_called() gates_activity.debug.assert_called()
@@ -380,19 +380,30 @@ def test_mlflow_response_gate_with_filter(gates_activity):
'filters': {'API_ERROR': {'POLICY': 'STOP'}}, 'filters': {'API_ERROR': {'POLICY': 'STOP'}},
'data': _minio_payload( 'data': _minio_payload(
{'content': {'message': 'API error occurred', 'traceback': 'error trace'}}, {'content': {'message': 'API error occurred', 'traceback': 'error trace'}},
status={'success': False, 'message': 'API error occurred'}, status={
'success': False,
'content': {'message': 'API error occurred', 'traceback': 'error trace'},
},
), ),
'type': 'test', 'type': 'test',
'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
} }
# Act # Act
result = gates_activity.mlflow_response_gate(input_data) path_flag, confidence, comment, document = gates_activity.mlflow_response_gate(input_data)
# Assert # 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.debug.assert_called()
gates_activity.send_notification.assert_called() gates_activity.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='TEST_GATE_RESPONSE_FILTER__API_ERROR',
message='API error occurred',
block='mlflow_gate',
level=NotificationLevel.ERROR,
attachment_content='error trace',
)
def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity): def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity):
@@ -402,7 +413,10 @@ def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity):
'filters': {'API_ERROR': {'Policy': 'STOP'}}, 'filters': {'API_ERROR': {'Policy': 'STOP'}},
'data': _minio_payload( 'data': _minio_payload(
{'content': {'message': 'API error occurred', 'traceback': 'error trace'}}, {'content': {'message': 'API error occurred', 'traceback': 'error trace'}},
status={'success': False, 'message': 'API error occurred'}, status={
'success': False,
'content': {'message': 'API error occurred', 'traceback': 'error trace'},
},
), ),
'type': 'test', 'type': 'test',
'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'],
@@ -412,7 +426,12 @@ def test_mlflow_response_gate_with_filter_capitalized_keys(gates_activity):
result = gates_activity.mlflow_response_gate(input_data) result = gates_activity.mlflow_response_gate(input_data)
# Assert # 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): def test_mlflow_response_gate_with_filter_not_caught(gates_activity):
@@ -432,7 +451,7 @@ def test_mlflow_response_gate_with_filter_not_caught(gates_activity):
result = gates_activity.mlflow_response_gate(input_data) result = gates_activity.mlflow_response_gate(input_data)
# Assert # Assert
assert result == (None, 0, '') assert result == (None, 0, '', None)
gates_activity.debug.assert_called() gates_activity.debug.assert_called()

View File

@@ -59,11 +59,11 @@ async def test_run(workflow_mock, prediction_process):
workflow_mock.execute_local_activity_method.side_effect = [ workflow_mock.execute_local_activity_method.side_effect = [
('continue', 0.95, 'Input data with bad quality', None), # input_gate ('continue', 0.95, 'Input data with bad quality', None), # input_gate
# mlflow_response_gate (transform) # mlflow_response_gate (transform)
('continue', 0.95, 'Error'), ('continue', 0.95, 'Error', None),
# mlflow_content_gate (transform) # mlflow_content_gate (transform)
('continue', 0.95, 'Transformed data not passed the content filter'), ('continue', 0.95, 'Transformed data not passed the content filter'),
# mlflow_response_gate (predict) # mlflow_response_gate (predict)
('continue', 0.95, 'Error'), ('continue', 0.95, 'Error', None),
] ]
# Act # 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 = [ workflow_mock.execute_local_activity_method.side_effect = [
('repeat', 0.95, 'Input data with bad quality', None), # input_gate ('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 # 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 = [ workflow_mock.execute_local_activity_method.side_effect = [
('continue', 0.95, 'Input data with bad quality', None), # input_gate ('continue', 0.95, 'Input data with bad quality', None), # input_gate
# mlflow_response_gate (transform) # mlflow_response_gate (transform)
('continue', 0.95, 'Error'), ('continue', 0.95, 'Error', None),
# mlflow_content_gate (transform) # mlflow_content_gate (transform)
('continue', 0.95, 'Transformed data not passed the content filter'), ('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 = [ workflow_mock.execute_local_activity_method.side_effect = [
('continue', 0.95, 'Input data with bad quality', None), # input_gate ('continue', 0.95, 'Input data with bad quality', None), # input_gate
# mlflow_response_gate (transform) # mlflow_response_gate (transform)
('continue', 0.95, 'Error'), ('continue', 0.95, 'Error', None),
# mlflow_content_gate (transform) # mlflow_content_gate (transform)
('continue', 0.95, 'Transformed data not passed the content filter'), ('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 # Act
@@ -696,9 +696,9 @@ async def test_path_flag_handler_repeat_no_document_skips_write_activity_call(
workflow_mock, prediction_process workflow_mock, prediction_process
): ):
# Arrange: no document (e.g. REPEAT triggered by a gate that doesn't build one yet — today, # 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 # `mlflow_content_gate`) must not even dispatch the remote write_prediction_metadata
# write_prediction_metadata activity — that would be a wasted Temporal round-trip on every # activity — that would be a wasted Temporal round-trip on every REPEAT caused by that
# REPEAT caused by those two gates, not a rare edge case (review finding, 2026-10-01). # gate, not a rare edge case (review finding, 2026-10-01).
data = {'test': 'data'} data = {'test': 'data'}
# Act # Act
@@ -1075,9 +1075,9 @@ async def test_run_with_cleanup_prefixes(workflow_mock, prediction_process):
] ]
workflow_mock.execute_local_activity_method.side_effect = [ workflow_mock.execute_local_activity_method.side_effect = [
('continue', 0.95, 'ok', None), # input_gate ('continue', 0.95, 'ok', None), # input_gate
('continue', 0.95, ''), ('continue', 0.95, '', None), # mlflow_response_gate (transform)
('continue', 0.95, ''), ('continue', 0.95, ''), # mlflow_content_gate (transform)
('continue', 0.95, ''), ('continue', 0.95, '', None), # mlflow_response_gate (predict)
] ]
await prediction_process.run(input_data) await prediction_process.run(input_data)