async def test_nested_async_error_handling(exporter):
    """Test error handling in an async workflow that calls an async task that raises an error"""
    error_message = "Intentional nested async error"

    @task(name="failing_nested_async_task")
    async def failing_nested_async_task():
        await asyncio.sleep(0.1)
        raise ValueError(error_message)

    @workflow(name="parent_async_workflow")
    async def parent_async_workflow():
        return await failing_nested_async_task()

    with pytest.raises(ValueError, match=error_message):
        await parent_async_workflow()

    spans = exporter.get_finished_spans()
    assert len(spans) == 2

    # Check task span
    task_span = next(span for span in spans if span.name == "failing_nested_async_task.task")
    assert task_span.status.status_code == StatusCode.ERROR
    assert error_message in task_span.status.description

    # Check workflow span
    workflow_span = next(span for span in spans if span.name == "parent_async_workflow.workflow")
    assert workflow_span.status.status_code == StatusCode.ERROR
    assert error_message in workflow_span.status.description
Test error handling in an async workflow that calls an async task that raises an error
packages/traceloop-sdk/tests/test_workflows.py