From 986351dae09628a981a9aed9e71f02f2488cba47 Mon Sep 17 00:00:00 2001 From: nandanadileep <116407139+nandanadileep@users.noreply.github.com> Date: Sat, 24 Oct 2026 19:49:50 +0000 Subject: [PATCH] fix(agents): report failed task reason for --- src/conductor/ai/agents/runtime/runtime.py | 48 ++++++- tests/unit/ai/test_runtime.py | 154 +++++++++++++++++++++ 2 files changed, 198 insertions(+), 4 deletions(-) diff --git a/src/conductor/ai/agents/runtime/runtime.py b/src/conductor/ai/agents/runtime/runtime.py index e53b921a..289199de 100644 --- a/src/conductor/ai/agents/runtime/runtime.py +++ b/src/conductor/ai/agents/runtime/runtime.py @@ -3177,13 +3177,22 @@ def _run_framework( ) token_usage = self._extract_token_usage(execution_id) tool_calls, events = self._extract_tool_calls_and_events(execution_id) + + # Build the richest error message available: prefer task-level reason + # (includes which task failed and why) over the workflow-level reason, + # mirroring the native-agent run() path. status.reason is often empty + # for framework failures, so fall back to the failed task's reason. + error_reason: Optional[str] = None + if raw_status in ("FAILED", "TERMINATED"): + error_reason = status.reason or self._framework_failure_reason(execution_id) + return AgentResult( output=output, execution_id=execution_id, 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=error_reason, token_usage=token_usage, tool_calls=tool_calls, events=events, @@ -3520,13 +3529,16 @@ def _run_framework_with_events( output = self._normalize_output(status.output, status.status, status.reason) token_usage = self._extract_token_usage(execution_id) tool_calls, _ = self._extract_tool_calls_and_events(execution_id) + error_reason: Optional[str] = None + if status.status in ("FAILED", "TERMINATED"): + error_reason = status.reason or self._framework_failure_reason(execution_id) return AgentResult( output=output, execution_id=execution_id, 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=error_reason, token_usage=token_usage, tool_calls=tool_calls, events=events, @@ -4512,13 +4524,18 @@ async def _run_framework_async( tool_calls, _ = await loop.run_in_executor( None, lambda: self._extract_tool_calls_and_events(execution_id) ) + error_reason: Optional[str] = None + if status.status in ("FAILED", "TERMINATED"): + error_reason = status.reason or await loop.run_in_executor( + None, lambda: self._framework_failure_reason(execution_id) + ) return AgentResult( output=output, execution_id=execution_id, 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=error_reason, token_usage=token_usage, tool_calls=tool_calls, events=captured_events, @@ -4547,13 +4564,18 @@ async def _run_framework_async( tool_calls, events = await loop.run_in_executor( None, lambda: self._extract_tool_calls_and_events(execution_id) ) + error_reason = None + if raw_status in ("FAILED", "TERMINATED"): + error_reason = status.reason or await loop.run_in_executor( + None, lambda: self._framework_failure_reason(execution_id) + ) return AgentResult( output=output, execution_id=execution_id, 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=error_reason, token_usage=token_usage, tool_calls=tool_calls, events=events, @@ -5055,6 +5077,24 @@ def _extract_failed_task_reason(wf: Any) -> Optional[str]: return f"Task '{ref}' failed" return None + def _framework_failure_reason(self, execution_id: str) -> Optional[str]: + """Resolve a failure reason for a framework execution. + + ``status.reason`` from the status endpoint is frequently empty for + framework-agent failures even though the server recorded a reason. + Fetch the full execution and prefer the first FAILED task's + ``reasonForIncompletion`` (via :meth:`_extract_failed_task_reason`), + falling back to the workflow's own ``reasonForIncompletion``. + """ + try: + wf = self._workflow_client.get_workflow(execution_id, include_tasks=True) + except Exception as exc: + logger.debug("Could not fetch execution details for %s: %s", execution_id, exc) + return None + return self._extract_failed_task_reason(wf) or getattr( + wf, "reason_for_incompletion", None + ) + @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 2b5ace45..d2182fd4 100644 --- a/tests/unit/ai/test_runtime.py +++ b/tests/unit/ai/test_runtime.py @@ -1974,6 +1974,160 @@ def _status_with_registry_check(execution_id, timeout=None): assert "wf-framework-1" not in _workflow_credentials +class TestFrameworkFailureReason: + """Issue #483: framework failures must surface the failed task's reason. + + status.reason from the status endpoint is empty for framework failures, so + the framework run paths must fall back to the failed task's + reasonForIncompletion (then the workflow's own reasonForIncompletion). + """ + + @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 _failed_task(self, status, ref="Assistant_llm", reason=None): + t = MagicMock() + t.status = status + t.reference_task_name = ref + t.task_type = ref + t.reason_for_incompletion = reason + return t + + def _failed_wf(self, tasks, wf_reason="Task execution failed"): + wf = MagicMock() + wf.tasks = tasks + wf.reason_for_incompletion = wf_reason + return wf + + def _patch_serializer(self): + return patch.multiple( + "conductor.ai.agents.frameworks.serializer", + detect_framework=MagicMock(return_value="openai"), + serialize_agent=MagicMock(return_value=({"name": "fw_agent"}, [])), + ) + + def _run_framework(self, runtime, wf, status_reason=None, on_event=None, status="FAILED"): + fake_agent = object() + with self._patch_serializer(): + with patch.object( + runtime, "_start_framework_via_server", return_value="wf-framework-1" + ): + with patch.object(runtime, "_stream_workflow", return_value=iter([])): + with patch.object( + runtime, + "_poll_status_until_complete", + return_value=AgentStatus( + execution_id="wf-framework-1", + is_complete=True, + status=status, + output=None if status == "FAILED" else {"result": "ok"}, + reason=status_reason, + ), + ): + with patch.object(runtime, "_extract_token_usage", return_value=None): + with patch.object( + runtime, "_extract_tool_calls_and_events", return_value=([], []) + ): + runtime._workflow_client.get_workflow = MagicMock(return_value=wf) + return runtime.run(fake_agent, "hello", on_event=on_event) + + def test_failed_framework_reports_failed_task_reason(self, runtime): + """When status.reason is empty, the failed task's reason is surfaced.""" + wf = self._failed_wf( + [ + self._failed_task("COMPLETED", ref="start"), + self._failed_task( + "FAILED", ref="Assistant_llm", reason="Responses API failed with status 401" + ), + ] + ) + result = self._run_framework(runtime, wf, status_reason=None) + assert result.status == "FAILED" + assert result.error is not None + assert "Assistant_llm" in result.error + assert "Responses API failed with status 401" in result.error + + def test_failed_framework_falls_back_to_workflow_reason(self, runtime): + """With no FAILED task, the workflow's reasonForIncompletion is used.""" + wf = self._failed_wf( + [self._failed_task("COMPLETED", ref="start")], + wf_reason="Workflow-level failure reason", + ) + result = self._run_framework(runtime, wf, status_reason=None) + assert result.error == "Workflow-level failure reason" + + def test_failed_framework_prefers_status_reason_when_present(self, runtime): + """An explicit status.reason takes precedence over the fallback.""" + wf = self._failed_wf( + [self._failed_task("FAILED", ref="Assistant_llm", reason="task-level reason")], + wf_reason="workflow-level reason", + ) + result = self._run_framework(runtime, wf, status_reason="status-level reason") + assert result.error == "status-level reason" + + def test_failed_framework_events_reports_failed_task_reason(self, runtime): + """The on_event streaming path surfaces the failed task's reason.""" + wf = self._failed_wf( + [self._failed_task("FAILED", ref="Assistant_llm", reason="API failed 401")] + ) + result = self._run_framework(runtime, wf, status_reason=None, on_event=lambda e: None) + assert result.error is not None + assert "Assistant_llm" in result.error + assert "API failed 401" in result.error + + def test_completed_framework_has_no_error(self, runtime): + """A successful framework run leaves error unset.""" + wf = self._failed_wf([], wf_reason=None) + result = self._run_framework(runtime, wf, status="COMPLETED") + assert result.status == "COMPLETED" + assert result.error is None + + @pytest.mark.asyncio + async def test_failed_framework_async_reports_failed_task_reason(self, runtime): + """Async framework failure surfaces the failed task's reason.""" + fake_agent = object() + wf = self._failed_wf( + [self._failed_task("FAILED", ref="Assistant_llm", reason="API failed 401")] + ) + with self._patch_serializer(): + with patch.object( + runtime, + "_start_framework_via_server_async", + AsyncMock(return_value="wf-framework-1"), + ): + with patch.object( + runtime, + "_poll_status_until_complete_async", + AsyncMock( + return_value=AgentStatus( + execution_id="wf-framework-1", + is_complete=True, + status="FAILED", + output=None, + reason=None, + ) + ), + ): + with patch.object(runtime, "_extract_token_usage", return_value=None): + with patch.object( + runtime, "_extract_tool_calls_and_events", return_value=([], []) + ): + runtime._workflow_client.get_workflow = MagicMock(return_value=wf) + result = await runtime.run_async(fake_agent, "hello") + + assert result.status == "FAILED" + assert result.error is not None + assert "Assistant_llm" in result.error + assert "API failed 401" in result.error + + class TestPollStatusUntilComplete: """Test _poll_status_until_complete() polling logic."""