From a5d7ed76df1a96ddbe3ff31e5cbfa29603a965fd Mon Sep 17 00:00:00 2001 From: Animesh Kumar Date: Sun, 13 Sep 2026 01:31:17 +0530 Subject: [PATCH] fix(crew): keep the call id when a parallel tool fails _execute_single already catches every Exception and returns a ToolResult carrying the call's id, so the fallback in _execute_parallel only ever sees a BaseException. It built that result with an empty tool_call_id and name, which reaches the model as {"role": "tool", "tool_call_id": ""}: a tool message has to name the call it answers, so the request is malformed rather than just uninformative. gather preserves order, so pair each result with its own call. CancelledError is the BaseException that actually turns up here. Re-raise it instead of turning it into a result, so a cancelled turn stops rather than reporting a fabricated answer for the call. --- src/smallestai/atoms/crew/tools/registry.py | 18 ++- .../test_tool_registry_parallel_errors.py | 137 ++++++++++++++++++ 2 files changed, 148 insertions(+), 7 deletions(-) create mode 100644 tests/custom/test_tool_registry_parallel_errors.py diff --git a/src/smallestai/atoms/crew/tools/registry.py b/src/smallestai/atoms/crew/tools/registry.py index 71218700..d7983983 100644 --- a/src/smallestai/atoms/crew/tools/registry.py +++ b/src/smallestai/atoms/crew/tools/registry.py @@ -163,17 +163,21 @@ async def _execute_parallel(self, tool_calls: List[ToolCall], context: Optional[ tasks = [self._execute_single(call, context) for call in tool_calls] results = await asyncio.gather(*tasks, return_exceptions=True) - # Handle exceptions + # gather preserves input order, so each result pairs with its own call and the + # error path can carry the id the LLM needs to match the result to the call. final_results: List[ToolResult] = [] - for result in results: + for call, result in zip(tool_calls, results): + if isinstance(result, asyncio.CancelledError): + # Cancellation is not a tool failure: swallowing it here would let a + # cancelled turn carry on and report a made-up result for the call. + raise result if isinstance(result, BaseException): - # Create error result - logger.exception(f"Tool execution failed: {result}") + logger.exception(f"Tool execution failed: {call.name}: {result}") final_results.append( ToolResult( - tool_call_id="", - name="", - content=str(result), + tool_call_id=call.id, + name=call.name, + content=str(result) or type(result).__name__, is_error=True, ) ) diff --git a/tests/custom/test_tool_registry_parallel_errors.py b/tests/custom/test_tool_registry_parallel_errors.py new file mode 100644 index 00000000..c0aaea2d --- /dev/null +++ b/tests/custom/test_tool_registry_parallel_errors.py @@ -0,0 +1,137 @@ +"""Regression test: a parallel tool failure keeps the id the LLM needs. + +`_execute_single` swallows every `Exception` and returns a `ToolResult` carrying the +call's id, so the fallback in `_execute_parallel` only ever sees a `BaseException`. +That fallback built its result with `tool_call_id=""` and `name=""`, which reaches the +model as `{"role": "tool", "tool_call_id": "", ...}`. A tool message has to name the +call it answers, so the whole request is malformed rather than merely uninformative. + +`asyncio.CancelledError` is the `BaseException` that actually turns up here, and +turning it into a result let a cancelled turn carry on. +""" + +import asyncio +import json +import unittest + +from smallestai.atoms.crew.clients.types import ToolCall +from smallestai.atoms.crew.tools.decorator import function_tool +from smallestai.atoms.crew.tools.registry import ToolRegistry + + +class _Fatal(BaseException): + """A BaseException that is not cancellation, so it still becomes a tool result.""" + + +@function_tool +async def works(x: str) -> str: + """A tool that succeeds. + + Args: + x: anything. + """ + return "ok" + + +@function_tool +async def fatal(x: str) -> str: + """A tool that raises a BaseException. + + Args: + x: anything. + """ + raise _Fatal("boom") + + +@function_tool +async def silent(x: str) -> str: + """A tool whose BaseException carries no message, so str() on it is empty. + + Args: + x: anything. + """ + raise _Fatal() + + +@function_tool +async def cancels(x: str) -> str: + """A tool that is cancelled. + + Args: + x: anything. + """ + raise asyncio.CancelledError() + + +@function_tool +async def explodes(x: str) -> str: + """A tool that raises an ordinary Exception. + + Args: + x: anything. + """ + raise ValueError("ordinary failure") + + +def _call(call_id: str, name: str) -> ToolCall: + return ToolCall(id=call_id, name=name, arguments=json.dumps({"x": "1"})) + + +class ParallelToolErrorTest(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.registry = ToolRegistry() + for tool in (works, fatal, silent, cancels, explodes): + self.registry.register(tool) + + async def test_a_base_exception_result_keeps_its_call_id_and_name(self): + results = await self.registry.execute([_call("call_bbb", "fatal")], parallel=True) + + self.assertEqual(len(results), 1) + self.assertEqual(results[0].tool_call_id, "call_bbb") + self.assertEqual(results[0].name, "fatal") + self.assertTrue(results[0].is_error) + + async def test_the_error_message_is_never_blank(self): + """str() on a message-less exception is empty, which left the model a tool + message with no content at all. Fall back to the exception's type name.""" + results = await self.registry.execute([_call("call_ccc", "silent")], parallel=True) + + self.assertTrue(results[0].content) + self.assertIn("_Fatal", results[0].content) + + async def test_results_stay_aligned_with_their_calls(self): + calls = [ + _call("call_1", "works"), + _call("call_2", "fatal"), + _call("call_3", "works"), + ] + + results = await self.registry.execute(calls, parallel=True) + + self.assertEqual([r.tool_call_id for r in results], ["call_1", "call_2", "call_3"]) + self.assertEqual([r.is_error for r in results], [False, True, False]) + + async def test_cancellation_propagates_instead_of_becoming_a_result(self): + calls = [_call("call_1", "works"), _call("call_2", "cancels")] + + with self.assertRaises(asyncio.CancelledError): + await self.registry.execute(calls, parallel=True) + + async def test_an_ordinary_exception_is_still_reported_as_a_tool_result(self): + """That path never reaches the fallback; it must keep working unchanged.""" + results = await self.registry.execute([_call("call_ddd", "explodes")], parallel=True) + + self.assertEqual(results[0].tool_call_id, "call_ddd") + self.assertEqual(results[0].name, "explodes") + self.assertTrue(results[0].is_error) + self.assertIn("ordinary failure", results[0].content) + + async def test_sequential_execution_is_unchanged(self): + results = await self.registry.execute([_call("call_eee", "explodes")], parallel=False) + + self.assertEqual(results[0].tool_call_id, "call_eee") + self.assertTrue(results[0].is_error) + + +if __name__ == "__main__": + unittest.main()