Merge 446d670e24 into 97d7554438

backend/app/utils/telemetry/workforce_metrics.py

@@ -74,6 +74,7 @@ ATTR_TASK_QUALITY_SCORE = "eigent.task.quality_score"
 ATTR_TASK_TIMESTAMP = "eigent.task.timestamp"
 ATTR_TASK_DEPENDENCIES = "eigent.task.dependencies"
 ATTR_TASK_SUBTASK_IDS = "eigent.task.subtask_ids"
+ATTR_TASK_FAILURE_COUNT = "eigent.task.failure_count"
 
 # Attribute keys for eigent.worker namespace
 ATTR_WORKER_ID = "eigent.worker.id"
@@ -553,6 +554,10 @@ class WorkforceMetricsCallback(WorkforceMetrics):
                 span.set_attribute(ATTR_TASK_PARENT_ID, event.parent_task_id)
             if event.worker_id:
                 span.set_attribute(ATTR_WORKER_ID, event.worker_id)
+            if event.metadata and isinstance(event.metadata.get("failure_count"),
+                                             int):
+                span.set_attribute(ATTR_TASK_FAILURE_COUNT,
+                                   event.metadata["failure_count"])
 
             span.set_status(Status(StatusCode.ERROR, event.error_message))
             span.end()

backend/app/utils/workforce.py

@@ -14,7 +14,7 @@
 
 import asyncio
 import logging
-from collections.abc import Generator
+from collections.abc import Generator, Mapping
 
 from camel.agents import ChatAgent
 from camel.societies.workforce.base import BaseNode
@@ -22,7 +22,6 @@ from camel.societies.workforce.events import (
     TaskAssignedEvent,
     TaskCompletedEvent,
     TaskCreatedEvent,
-    TaskFailedEvent,
     WorkerCreatedEvent,
 )
 from camel.societies.workforce.prompts import TASK_DECOMPOSE_PROMPT
@@ -800,15 +799,28 @@ class Workforce(BaseWorkforce):
         metrics_callbacks = [
             cb for cb in self._callbacks if isinstance(cb, WorkforceMetrics)
         ]
-        if metrics_callbacks and hasattr(metrics_callbacks[0], "log_entries"):
-            for entry in reversed(metrics_callbacks[0].log_entries):
+        for callback in metrics_callbacks:
+            log_entries = getattr(callback, "log_entries", None)
+            if not isinstance(log_entries, list):
+                continue
+
+            for entry in reversed(log_entries):
+                if not isinstance(entry, Mapping):
+                    continue
                 if (
                     entry.get("event_type") == "task_failed"
                     and entry.get("task_id") == task.id
                 ):
-                    error_message = entry.get("error_message")
+                    error_message = str(entry.get("error_message") or "")
                     break
 
+            if error_message:
+                break
+
+        resolved_error_message = error_message or str(
+            task.result or "Unknown error"
+        )
+
         task_lock = get_task_lock(self.api_task_id)
         await task_lock.put_queue(
             ActionTaskStateData(
@@ -817,26 +829,11 @@ class Workforce(BaseWorkforce):
                     "content": task.content,
                     "state": task.state,
                     "failure_count": task.failure_count,
-                    "result": str(error_message),
+                    "result": resolved_error_message,
                 }
             )
         )
 
-        if metrics_callbacks:
-            error_msg = error_message or str(task.result or "Unknown error")
-            # Pass all values during construction since TaskFailedEvent is frozen
-            worker_id = (
-                task.assigned_worker_id
-                if hasattr(task, "assigned_worker_id")
-                else None
-            )
-            event = TaskFailedEvent(
-                task_id=task.id,
-                error_message=error_msg,
-                worker_id=worker_id,
-            )
-            metrics_callbacks[0].log_task_failed(event)
-
         return result
 
     async def _get_returned_task(self) -> Task | None:

backend/tests/app/utils/telemetry/test_workforce_metrics.py

@@ -17,6 +17,11 @@ from datetime import datetime
 from unittest.mock import MagicMock, Mock, patch
 
 import pytest
+import app.utils.telemetry.workforce_metrics as wm_module
+from app.utils.telemetry.workforce_metrics import (
+    ATTR_TASK_FAILURE_COUNT,
+    WorkforceMetricsCallback,
+)
 from camel.societies.workforce.events import (
     LogEvent,
     TaskAssignedEvent,
@@ -29,9 +34,6 @@ from camel.societies.workforce.events import (
     WorkerCreatedEvent,
 )
 
-import app.utils.telemetry.workforce_metrics as wm_module
-from app.utils.telemetry.workforce_metrics import WorkforceMetricsCallback
-
 
 @pytest.fixture(autouse=True)
 def reset_global_tracer_provider():
@@ -238,12 +240,14 @@ def test_log_task_failed(metrics_callback):
         worker_id="worker_1",
         parent_task_id="parent_1",
         error_message="Test error",
+        metadata={"failure_count": 3},
     )
 
     metrics_callback.log_task_failed(event)
 
     # Verify span was ended with error status
     assert task_id not in metrics_callback.task_spans
+    mock_span.set_attribute.assert_any_call(ATTR_TASK_FAILURE_COUNT, 3)
     assert mock_span.set_attribute.called
     assert mock_span.set_status.called
     assert mock_span.end.called

backend/tests/app/utils/test_workforce.py

@@ -24,6 +24,7 @@ from camel.societies.workforce.workforce import (
     Workforce as BaseWorkforce,
     WorkforceState,
 )
+from camel.societies.workforce.workforce_metrics import WorkforceMetrics
 from camel.tasks import Task
 from camel.tasks.task import TaskState
 
@@ -422,6 +423,77 @@ async def test_handle_failed_task(mock_task_lock):
         mock_super_handle.assert_called_once_with(task)
 
 
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_handle_failed_task_avoids_duplicate_callback_logs(
+    mock_task_lock,
+):
+    """Test _handle_failed_task does not double-log failed events."""
+    api_task_id = "test_api_task_123"
+    workforce = Workforce(
+        api_task_id=api_task_id, description="Test workforce"
+    )
+
+    task = Task(content="Failed task", id="failed_123")
+    task.state = TaskState.FAILED
+    task.failure_count = workforce.failure_handling_config.max_retries
+    task.result = "Test failure"
+
+    class DummyMetrics(WorkforceMetrics):
+        def __init__(self):
+            self.log_entries = []
+            self.log_task_failed = MagicMock()
+
+        def reset_task_data(self) -> None:
+            return None
+
+        def dump_to_json(self, file_path: str) -> None:
+            return None
+
+        def get_ascii_tree_representation(self) -> str:
+            return ""
+
+        def get_kpis(self):
+            return {}
+
+    callback_one = DummyMetrics()
+    callback_one.log_entries = [object()]
+    callback_two = DummyMetrics()
+    callback_two.log_entries = [
+        {
+            "event_type": "task_failed",
+            "task_id": "failed_123",
+            "error_message": "Failure from callback two",
+        }
+    ]
+    workforce._callbacks = [callback_one, callback_two]
+
+    async def fake_super_handle_failed(_task):
+        for callback in workforce._callbacks:
+            callback.log_task_failed(MagicMock())
+        return True
+
+    with (
+        patch(
+            "app.utils.workforce.get_task_lock",
+            return_value=mock_task_lock,
+        ),
+        patch.object(
+            workforce.__class__.__bases__[0],
+            "_handle_failed_task",
+            side_effect=fake_super_handle_failed,
+        ),
+    ):
+        await workforce._handle_failed_task(task)
+
+    mock_task_lock.put_queue.assert_called_once()
+    call_args = mock_task_lock.put_queue.call_args[0][0]
+    assert call_args.data["result"] == "Failure from callback two"
+
+    callback_one.log_task_failed.assert_called_once()
+    callback_two.log_task_failed.assert_called_once()
+
+
 @pytest.mark.unit
 @pytest.mark.asyncio
 async def test_stop_sends_end_notification(mock_task_lock):