Merge branch 'main' into gl/feat/tool-call-stream-events

* fix: gracefully terminate the future when executing a task async

* core: add unit test

---------

Co-authored-by: Greyson LaLonde <greyson.r.lalonde@gmail.com>

lib/crewai/src/crewai/cli/tools/main.py

@@ -1,4 +1,5 @@
 import base64
+from json import JSONDecodeError
 import os
 from pathlib import Path
 import subprocess
@@ -162,9 +163,19 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
 
         if login_response.status_code != 200:
             console.print(
-                "Authentication failed. Verify if the currently active organization access to the tool repository, and run 'crewai login' again. ",
+                "Authentication failed. Verify if the currently active organization can access the tool repository, and run 'crewai login' again.",
                 style="bold red",
             )
+            try:
+                console.print(
+                    f"[{login_response.status_code} error - {login_response.json().get('message', 'Unknown error')}]",
+                    style="bold red italic",
+                )
+            except JSONDecodeError:
+                console.print(
+                    f"[{login_response.status_code} error - Unknown error - Invalid JSON response]",
+                    style="bold red italic",
+                )
             raise SystemExit
 
         login_response_json = login_response.json()

lib/crewai/src/crewai/task.py

@@ -494,8 +494,11 @@ class Task(BaseModel):
         future: Future[TaskOutput],
     ) -> None:
         """Execute the task asynchronously with context handling."""
-        result = self._execute_core(agent, context, tools)
-        future.set_result(result)
+        try:
+          result = self._execute_core(agent, context, tools)
+          future.set_result(result)
+        except Exception as e:
+          future.set_exception(e)
 
     async def aexecute_sync(
         self,

lib/crewai/tests/test_task.py

@@ -1727,3 +1727,24 @@ def test_task_output_includes_messages():
     assert hasattr(task2_output, "messages")
     assert isinstance(task2_output.messages, list)
     assert len(task2_output.messages) > 0
+
+
+def test_async_execution_fails():
+    researcher = Agent(
+      role="Researcher",
+      goal="Make the best research and analysis on content about AI and AI agents",
+      backstory="You're an expert researcher, specialized in technology, software engineering, AI and startups. You work as a freelancer and is now working on doing research and analysis for a new customer.",
+      allow_delegation=False,
+    )
+
+    task = Task(
+      description="Give me a list of 5 interesting ideas to explore for na article, what makes them unique and interesting.",
+      expected_output="Bullet point list of 5 interesting ideas.",
+      async_execution=True,
+      agent=researcher,
+    )
+
+    with patch.object(Task, "_execute_core", side_effect=RuntimeError("boom!")):
+      with pytest.raises(RuntimeError, match="boom!"):
+        execution = task.execute_async(agent=researcher)
+        execution.result()