diff --git a/backend-fastapi/ai_platform/services/workflow_service.py b/backend-fastapi/ai_platform/services/workflow_service.py index 17f7db5..1e4b1a1 100644 --- a/backend-fastapi/ai_platform/services/workflow_service.py +++ b/backend-fastapi/ai_platform/services/workflow_service.py @@ -286,6 +286,8 @@ def _make_stream_event_log_entry(event: Dict[str, Any]) -> Optional[dict]: timestamp = event_data.get('timestamp') or datetime.now().isoformat() event_data.setdefault('timestamp', timestamp) communication = dict(event_data.get('communication') or {}) + collaboration = dict(event_data.get('collaboration') or {}) + actor = dict(communication.get('actor') or {}) communication.setdefault('event', event_type) communication.setdefault('timestamp', timestamp) @@ -294,6 +296,8 @@ def _make_stream_event_log_entry(event: Dict[str, Any]) -> Optional[dict]: node_label = ( event_data.get('node_label') or event_data.get('workflow_name') + or collaboration.get('node_label') + or actor.get('node_label') or communication.get('message') or event_type ) @@ -317,7 +321,7 @@ def _make_stream_event_log_entry(event: Dict[str, Any]) -> Optional[dict]: 'events': [event_data], 'metadata': { 'stream_event': True, - 'collaboration': event_data.get('collaboration') or {}, + 'collaboration': collaboration, 'communication': communication, }, } @@ -334,7 +338,19 @@ def _merge_execution_logs(node_logs: List[dict], event_logs: List[dict]) -> List log for log in (node_logs or []) if not (log.get('metadata') or {}).get('stream_event') ] - merged = base_logs + list(event_logs or []) + completed_base_nodes = { + log.get('node_id') + for log in base_logs + if log.get('node_id') and log.get('status') in {'completed', 'failed', 'waiting'} + } + filtered_event_logs = [ + log for log in (event_logs or []) + if not ( + log.get('type') == 'node_complete' + and log.get('node_id') in completed_base_nodes + ) + ] + merged = base_logs + filtered_event_logs return sorted( merged, key=lambda item: (_log_timestamp(item), 0 if item.get('metadata', {}).get('stream_event') else 1),