Skip to content

fix(agents): preserve framework failure reasons #505

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mikemikimike wants to merge 1 commit into
base: main
Choose a base branch
from
+119 −14
Open

fix(agents): preserve framework failure reasons #505

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 61 additions & 14 deletions src/conductor/ai/agents/runtime/runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
Expand All @@ -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),
)
Expand Down Expand Up @@ -3216,15 +3219,25 @@ 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,
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=failure_reason,
token_usage=token_usage,
events=events,
sub_results=self._extract_sub_results(output),
Expand Down Expand Up @@ -4414,20 +4427,23 @@ 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,
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=failure_reason,
token_usage=token_usage,
events=captured_events,
sub_results=self._extract_sub_results(output),
Expand All @@ -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
)
Expand All @@ -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),
)
Expand Down Expand Up @@ -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."""
Expand Down
58 changes: 58 additions & 0 deletions tests/unit/ai/test_runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import logging
import threading
import uuid
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
Expand Down Expand Up @@ -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."""

Expand Down