fix: Framework-agent failures return None instead of the failed task's reason

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."""

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."""