fix(agent): surface knowledge retrieval failures (#38640)

api/clients/agent_backend/errors.py

@@ -54,8 +54,21 @@ class AgentBackendRunFailedError(AgentBackendError):
 
     run_id: str
     detail: Any
+    reason: str | None
+    source_event_id: str | None
 
-    def __init__(self, run_id: str, detail: Any) -> None:
+    def __init__(
+        self,
+        run_id: str,
+        detail: Any,
+        *,
+        message: str | None = None,
+        reason: str | None = None,
+        source_event_id: str | None = None,
+    ) -> None:
         self.run_id = run_id
         self.detail = detail
-        super().__init__(f"Agent backend run failed: {run_id}")
+        self.reason = reason
+        self.source_event_id = source_event_id
+        display_message = message or f"Agent backend run failed: {run_id}"
+        super().__init__(f"{display_message} (agent_run_id={run_id})")

api/core/app/apps/agent_app/app_runner.py

@@ -27,6 +27,7 @@ from clients.agent_backend import (
     AgentBackendInternalEventType,
     AgentBackendRunClient,
     AgentBackendRunEventAdapter,
+    AgentBackendRunFailedError,
     AgentBackendRunFailedInternalEvent,
     AgentBackendRunSucceededInternalEvent,
     AgentBackendStreamInternalEvent,
@@ -94,7 +95,18 @@ def _agent_backend_failure_to_exception(event: AgentBackendRunFailedInternalEven
     err_cls = _AGENT_BACKEND_INVOKE_ERROR_BY_REASON.get(event.reason or "")
     if err_cls is not None:
         return err_cls(event.error)
-    return AgentBackendError(event.error or "Agent backend run did not complete successfully.")
+    message = event.error or "Agent backend run did not complete successfully."
+    return AgentBackendRunFailedError(
+        event.run_id,
+        {
+            "error": event.error,
+            "reason": event.reason,
+            "source_event_id": event.source_event_id,
+        },
+        message=message,
+        reason=event.reason,
+        source_event_id=event.source_event_id,
+    )
 
 
 def _prompt_messages_from_query(user_query: str | None) -> list[PromptMessage]:

api/tests/unit_tests/core/app/apps/agent_app/test_app_runner.py

@@ -37,6 +37,7 @@ from pydantic_ai.messages import (
 from clients.agent_backend import (
     AgentBackendError,
     AgentBackendRunEventAdapter,
+    AgentBackendRunFailedError,
     AgentBackendRunFailedInternalEvent,
     AgentBackendStreamInternalEvent,
     FakeAgentBackendRunClient,
@@ -1207,7 +1208,7 @@ def test_failed_run_raises_agent_backend_error():
     store = _FakeSessionStore()
     qm = _FakeQueueManager()
 
-    with pytest.raises(AgentBackendError):
+    with pytest.raises(AgentBackendRunFailedError, match="fake failure .*agent_run_id=fake-run-1"):
         _run(_runner(client, store), qm)
     # No message-end on failure; no snapshot saved.
     assert not [e for e in qm.events if isinstance(e, QueueMessageEndEvent)]
@@ -1227,6 +1228,28 @@ def test_agent_backend_failure_to_exception_maps_rate_limit_reason():
     assert str(err) == "quota exceeded"
 
 
+def test_agent_backend_failure_to_exception_preserves_unknown_reason_context():
+    err = app_runner_module._agent_backend_failure_to_exception(
+        AgentBackendRunFailedInternalEvent(
+            run_id="run-1",
+            source_event_id="event-1",
+            error="Knowledge retrieval failed",
+            reason="knowledge_retrieve_failed",
+        )
+    )
+
+    assert isinstance(err, AgentBackendRunFailedError)
+    assert err.run_id == "run-1"
+    assert err.reason == "knowledge_retrieve_failed"
+    assert err.source_event_id == "event-1"
+    assert err.detail == {
+        "error": "Knowledge retrieval failed",
+        "reason": "knowledge_retrieve_failed",
+        "source_event_id": "event-1",
+    }
+    assert str(err) == "Knowledge retrieval failed (agent_run_id=run-1)"
+
+
 def test_stopped_task_cancels_agent_backend_run_and_skips_session_save():
     client = _RecordingFakeAgentBackendRunClient()
     store = _FakeSessionStore()

api/tests/unit_tests/core/app/task_pipeline/test_based_generate_task_pipeline.py

@@ -3,6 +3,7 @@ from unittest.mock import Mock
 
 import pytest
 
+from clients.agent_backend.errors import AgentBackendRunFailedError
 from core.app.apps.base_app_generate_response_converter import AppGenerateResponseConverter
 from core.app.entities.queue_entities import QueueErrorEvent
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
@@ -41,6 +42,22 @@ class TestBasedGenerateTaskPipeline:
         err = pipeline.handle_error(event=event)
         assert err is event.error
 
+    def test_handle_error_preserves_agent_backend_run_failed_error(self, pipeline):
+        event = QueueErrorEvent(
+            error=AgentBackendRunFailedError(
+                "run-1",
+                {"reason": "knowledge_retrieve_failed"},
+                message="Knowledge retrieval failed",
+                reason="knowledge_retrieve_failed",
+            )
+        )
+
+        err = pipeline.handle_error(event=event)
+
+        assert err is event.error
+        assert "Knowledge retrieval failed" in str(err)
+        assert "agent_run_id=run-1" in str(err)
+
     def test_handle_error_updates_message_when_found(self, pipeline):
         event = QueueErrorEvent(error=ValueError("oops"))
         message = SimpleNamespace(status=MessageStatus.NORMAL, error=None)
@@ -74,6 +91,22 @@ class TestBasedGenerateTaskPipeline:
 
         assert data == {"code": "rate_limit_error", "status": 429, "message": "quota exceeded"}
 
+    def test_stream_converter_maps_agent_backend_run_failed_error(self):
+        data = AppGenerateResponseConverter._error_to_stream_response(
+            AgentBackendRunFailedError(
+                "run-1",
+                {"reason": "knowledge_retrieve_failed"},
+                message="Knowledge retrieval failed",
+                reason="knowledge_retrieve_failed",
+            )
+        )
+
+        assert data == {
+            "code": "completion_request_error",
+            "status": 400,
+            "message": "Knowledge retrieval failed (agent_run_id=run-1)",
+        }
+
     def test_handle_output_moderation_when_flagged(self, pipeline):
         handler = Mock()
         handler.moderation_completion.return_value = ("filtered", True)