diff --git a/src/conductor/ai/agents/runtime/runtime.py b/src/conductor/ai/agents/runtime/runtime.py index e4a1709c..af0824e7 100644 --- a/src/conductor/ai/agents/runtime/runtime.py +++ b/src/conductor/ai/agents/runtime/runtime.py @@ -2864,16 +2864,19 @@ def _run_framework( output = status.output raw_status = status.status + failure_reason = self._extract_framework_failure_reason( + execution_id, raw_status, status.reason + ) if raw_status in ("FAILED", "TERMINATED"): logger.warning("Framework agent '%s' execution %s", agent_name, raw_status) has_output = output and not ( isinstance(output, dict) and all(v is None for v in output.values()) ) - if not has_output and status.reason: - output = status.reason + if not has_output and failure_reason: + output = failure_reason - output = self._normalize_output(output, raw_status, status.reason) + output = self._normalize_output(output, raw_status, failure_reason) logger.info( "Framework agent '%s' completed (execution_id=%s)", agent_name, execution_id ) @@ -2884,7 +2887,7 @@ def _run_framework( correlation_id=correlation_id, status=raw_status, finish_reason=self._derive_finish_reason(raw_status, status.output), - error=status.reason if raw_status in ("FAILED", "TERMINATED") else None, + error=failure_reason, token_usage=token_usage, sub_results=self._extract_sub_results(output), ) @@ -3216,7 +3219,17 @@ def _run_framework_with_events( on_event(event) status = self._poll_status_until_complete(execution_id, timeout=timeout) - output = self._normalize_output(status.output, status.status, status.reason) + failure_reason = self._extract_framework_failure_reason( + execution_id, status.status, status.reason + ) + output = status.output + if ( + status.status in ("FAILED", "TERMINATED") + and not output + and failure_reason + ): + output = failure_reason + output = self._normalize_output(output, status.status, failure_reason) token_usage = self._extract_token_usage(execution_id) return AgentResult( output=output, @@ -3224,7 +3237,7 @@ def _run_framework_with_events( correlation_id=correlation_id, status=status.status, finish_reason=self._derive_finish_reason(status.status, status.output), - error=status.reason if status.status in ("FAILED", "TERMINATED") else None, + error=failure_reason, token_usage=token_usage, events=events, sub_results=self._extract_sub_results(output), @@ -4414,12 +4427,15 @@ async def _run_framework_async( status = await self._poll_status_until_complete_async(execution_id, timeout=timeout) output = status.output + failure_reason = self._extract_framework_failure_reason( + execution_id, status.status, status.reason + ) has_output = output and not ( isinstance(output, dict) and all(v is None for v in output.values()) ) - if not has_output and status.reason and status.status in ("FAILED", "TERMINATED"): - output = status.reason - output = self._normalize_output(output, status.status, status.reason) + if not has_output and failure_reason: + output = failure_reason + output = self._normalize_output(output, status.status, failure_reason) token_usage = self._extract_token_usage(execution_id) return AgentResult( output=output, @@ -4427,7 +4443,7 @@ async def _run_framework_async( correlation_id=correlation_id, status=status.status, finish_reason=self._derive_finish_reason(status.status, status.output), - error=status.reason if status.status in ("FAILED", "TERMINATED") else None, + error=failure_reason, token_usage=token_usage, events=captured_events, sub_results=self._extract_sub_results(output), @@ -4437,16 +4453,19 @@ async def _run_framework_async( output = status.output raw_status = status.status + failure_reason = self._extract_framework_failure_reason( + execution_id, raw_status, status.reason + ) if raw_status in ("FAILED", "TERMINATED"): logger.warning("Framework agent '%s' execution %s", agent_name, raw_status) has_output = output and not ( isinstance(output, dict) and all(v is None for v in output.values()) ) - if not has_output and status.reason: - output = status.reason + if not has_output and failure_reason: + output = failure_reason - output = self._normalize_output(output, raw_status, status.reason) + output = self._normalize_output(output, raw_status, failure_reason) logger.info( "Framework agent '%s' completed (execution_id=%s)", agent_name, execution_id ) @@ -4457,7 +4476,7 @@ async def _run_framework_async( correlation_id=correlation_id, status=raw_status, finish_reason=self._derive_finish_reason(raw_status, status.output), - error=status.reason if raw_status in ("FAILED", "TERMINATED") else None, + error=failure_reason, token_usage=token_usage, sub_results=self._extract_sub_results(output), ) @@ -4957,6 +4976,34 @@ def _extract_failed_task_reason(wf: Any) -> Optional[str]: return f"Task '{ref}' failed" return None + def _extract_framework_failure_reason( + self, + execution_id: str, + raw_status: str, + status_reason: Optional[str], + ) -> Optional[str]: + """Return the most useful failure reason for a framework execution. + + Framework status responses can omit the reason even though the full + workflow contains a failed task with a diagnostic reason. Keep the + status endpoint as the fallback, but prefer the task and workflow + reasons when they are available. + """ + if raw_status not in ("FAILED", "TERMINATED"): + return None + try: + wf = self._workflow_client.get_workflow(execution_id, include_tasks=True) + return ( + self._extract_failed_task_reason(wf) + or getattr(wf, "reason_for_incompletion", None) + or status_reason + ) + except Exception as exc: + logger.debug( + "Could not fetch framework failure details for %s: %s", execution_id, exc + ) + return status_reason + @staticmethod def _extract_sub_results(output: Dict[str, Any]) -> Dict[str, Any]: """Extract subResults from server-normalized output, if present.""" diff --git a/tests/unit/ai/test_runtime.py b/tests/unit/ai/test_runtime.py index c43532da..1429b8f3 100644 --- a/tests/unit/ai/test_runtime.py +++ b/tests/unit/ai/test_runtime.py @@ -7,6 +7,7 @@ import logging import threading import uuid +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -1974,6 +1975,63 @@ def _status_with_registry_check(execution_id, timeout=None): assert "wf-framework-1" not in _workflow_credentials +class TestFrameworkFailureReason: + """Test that framework failures retain server-side diagnostic reasons.""" + + @pytest.fixture() + def runtime(self): + with patch("conductor.client.orkes_clients.OrkesClients"): + with patch("conductor.ai.agents.runtime.worker_manager.TaskHandler", create=True): + from conductor.ai.agents.runtime.config import AgentConfig + from conductor.ai.agents.runtime.runtime import AgentRuntime + + config = AgentConfig() + return AgentRuntime(settings=config) + + def test_framework_failure_uses_failed_task_reason_when_status_reason_is_empty(self, runtime): + fake_framework_agent = object() + failed_task = SimpleNamespace( + status="FAILED", + reference_task_name="Assistant_llm", + reason_for_incompletion="Responses API failed with status 401", + ) + workflow = SimpleNamespace( + tasks=[failed_task], + reason_for_incompletion="workflow failed", + ) + status = AgentStatus( + execution_id="wf-framework-failed", + is_complete=True, + status="FAILED", + output=None, + reason=None, + ) + + with patch( + "conductor.ai.agents.frameworks.serializer.detect_framework", return_value="openai" + ): + with patch( + "conductor.ai.agents.frameworks.serializer.serialize_agent", + return_value=({"name": "fw_agent"}, []), + ): + with patch.object( + runtime, "_start_framework_via_server", return_value="wf-framework-failed" + ): + with patch.object(runtime, "_poll_status_until_complete", return_value=status): + with patch.object(runtime, "_extract_token_usage", return_value=None): + runtime._workflow_client.get_workflow = MagicMock( + return_value=workflow + ) + result = runtime.run(fake_framework_agent, "hello") + + expected = "Task 'Assistant_llm' failed: Responses API failed with status 401" + assert result.error == expected + assert result.output == {"error": expected, "status": "FAILED"} + runtime._workflow_client.get_workflow.assert_called_once_with( + "wf-framework-failed", include_tasks=True + ) + + class TestPollStatusUntilComplete: """Test _poll_status_until_complete() polling logic."""