fix: guard malformed tool-call JSON, tool-pair-safe window trimming, and same-batch async dependency ordering (fixes #4075)

1. Guard the tool-call argument JSON parse in the non-streaming native
   loops (openai_client.chat_completion_with_tools / achat_completion_with_tools)
   so a malformed/truncated arguments string is reported back to the model
   instead of aborting the whole agent turn, matching the streaming path.
   Also read the real tool name from tool_call["function"]["name"] in
   llm._parse_tool_call_arguments' error branch instead of the never-present
   top-level "name" key.

2. Route DefaultSessionStore._enforce_window's window slice through the
   existing _trim_preserving_tool_exchanges guard, and skip leading orphaned
   role="tool" messages in the context TruncateOptimizer/SlidingWindowOptimizer,
   so a persisted/optimized window never starts mid tool-exchange.

3. Split an async task batch on an intra-batch context dependency in
   arun_all_tasks (sequential + workflow) so a dependent async task's context
   is built after its same-batch dependency completes, not before.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Mervin Praison <[email protected]>

src/praisonai-agents/praisonaiagents/agents/agents.py

@@ -1466,6 +1466,22 @@ class AgentTeam(SpawnAnnounceProtocol):
                 raise result
         return results
 
+    @staticmethod
+    def _depends_on_pending(task, pending):
+        """True if ``task`` has a context dependency still pending in the batch.
+
+        ``pending`` is a list of ``(task_id, coroutine)`` pairs already queued
+        for parallel execution. If one of the task's context dependencies is in
+        that set, its result isn't available yet, so the batch must be flushed
+        before queuing this task (otherwise process_task_context reads empty
+        context for the still-in-progress dependency).
+        """
+        deps = getattr(task, 'context', None) or []
+        if not deps:
+            return False
+        pending_ids = {tid for tid, _ in pending}
+        return any(getattr(dep, 'id', None) in pending_ids for dep in deps)
+
     async def arun_all_tasks(self):
         """Async version of run_all_tasks method"""
         process = Process(
@@ -1477,14 +1493,21 @@ class AgentTeam(SpawnAnnounceProtocol):
         )
         
         if self.process == "workflow":
-            tasks_to_run = []
+            tasks_to_run = []  # list of (task_id, coroutine)
             async for task_id in process.aworkflow():
-                if self.tasks[task_id].async_execution:
-                    tasks_to_run.append(self.arun_task(task_id))
+                task = self.tasks[task_id]
+                if task.async_execution:
+                    # If this async task depends on another async task still
+                    # pending in the current batch, flush first so its context
+                    # is available before this one reads it.
+                    if self._depends_on_pending(task, tasks_to_run):
+                        await self._gather_with_isolation([c for _, c in tasks_to_run])
+                        tasks_to_run = []
+                    tasks_to_run.append((task_id, self.arun_task(task_id)))
                 else:
                     # If we encounter a sync task, we must wait for the previous async tasks to finish.
                     if tasks_to_run:
-                        await self._gather_with_isolation(tasks_to_run)
+                        await self._gather_with_isolation([c for _, c in tasks_to_run])
                         tasks_to_run = []
                     
                     # Run sync task in an executor to avoid blocking the event loop
@@ -1494,16 +1517,16 @@ class AgentTeam(SpawnAnnounceProtocol):
                     await loop.run_in_executor(None, copy_context_to_callable(lambda tid=task_id: self.run_task(tid)))
 
             if tasks_to_run:
-                await self._gather_with_isolation(tasks_to_run)
+                await self._gather_with_isolation([c for _, c in tasks_to_run])
                 
         elif self.process == "sequential":
-            async_tasks_to_run = []
+            async_tasks_to_run = []  # list of (task_id, coroutine)
             
             async def flush_async_tasks():
                 """Execute all pending async tasks"""
                 nonlocal async_tasks_to_run
                 if async_tasks_to_run:
-                    await self._gather_with_isolation(async_tasks_to_run)
+                    await self._gather_with_isolation([c for _, c in async_tasks_to_run])
                     async_tasks_to_run = []
 
             def _deps_failed(task_id):
@@ -1525,9 +1548,16 @@ class AgentTeam(SpawnAnnounceProtocol):
                 return False
 
             async for task_id in process.asequential():
-                if self.tasks[task_id].async_execution:
+                task = self.tasks[task_id]
+                if task.async_execution:
+                    # If this async task depends on another async task still
+                    # pending in the current batch, flush first so its context
+                    # is available before this one reads it (avoids a same-batch
+                    # race that silently substitutes empty dependency context).
+                    if self._depends_on_pending(task, async_tasks_to_run):
+                        await flush_async_tasks()
                     # Collect async tasks to run in parallel
-                    async_tasks_to_run.append(self.arun_task(task_id))
+                    async_tasks_to_run.append((task_id, self.arun_task(task_id)))
                 else:
                     # Before running a sync task, execute all pending async tasks
                     await flush_async_tasks()

src/praisonai-agents/praisonaiagents/context/optimizer.py

@@ -14,6 +14,21 @@ from .compressor import ContextCompressor
 logger = logging.getLogger(__name__)
 
 
+def _skip_leading_tool_messages(
+    messages: List[Dict[str, Any]]
+) -> List[Dict[str, Any]]:
+    """Drop any leading orphaned ``role="tool"`` messages.
+
+    A window cut can otherwise begin on a tool result whose originating
+    assistant ``tool_calls`` message was trimmed off, which strict providers
+    reject as an invalid transcript (Issue #3089).
+    """
+    start = 0
+    while start < len(messages) and messages[start].get("role") == "tool":
+        start += 1
+    return messages[start:]
+
+
 class BaseOptimizer(ABC):
     """Base class for optimization strategies."""
     
@@ -71,8 +86,9 @@ class TruncateOptimizer(BaseOptimizer):
             else:
                 other_msgs.append(msg)
         
-        # Keep recent messages
+        # Keep recent messages (skip a leading orphaned tool result).
         recent = other_msgs[-self.preserve_recent:] if other_msgs else []
+        recent = _skip_leading_tool_messages(recent)
         
         result = system_msgs + recent
         
@@ -133,9 +149,11 @@ class SlidingWindowOptimizer(BaseOptimizer):
             else:
                 break
         
-        # Ensure order
+        # Ensure order (skip a leading orphaned tool result in the tail).
         system_msgs = [m for m in result if m.get("role") == "system"]
-        other_msgs = [m for m in result if m.get("role") != "system"]
+        other_msgs = _skip_leading_tool_messages(
+            [m for m in result if m.get("role") != "system"]
+        )
         result = system_msgs + other_msgs
         
         optimized_tokens = estimate_messages_tokens(result)

src/praisonai-agents/praisonaiagents/llm/llm.py

@@ -1766,7 +1766,11 @@ Respond with ONLY a valid JSON tool call in this format:
                 
         except (KeyError, json.JSONDecodeError, TypeError) as e:
             logging.error(f"Error parsing tool call arguments: {e}")
-            function_name = tool_call.get("name", "unknown_function")
+            function_name = (
+                tool_call.get("function", {}).get("name")
+                if isinstance(tool_call.get("function"), dict)
+                else None
+            ) or tool_call.get("name", "unknown_function")
             arguments = {}
             tool_call_id = tool_call.get("id", f"tool_{id(tool_call)}")
             

src/praisonai-agents/praisonaiagents/llm/openai_client.py

@@ -1906,13 +1906,26 @@ class OpenAIClient:
                 # replies for this turn (keeps tool replies consecutive).
                 _deferred_media_followups = []
                 for tool_call in tool_calls:
-                    # Handle both ToolCall dataclass and OpenAI object
-                    if isinstance(tool_call, ToolCall):
-                        function_name = tool_call.function["name"]
-                        arguments = json.loads(tool_call.function["arguments"])
-                    else:
-                        function_name = tool_call.function.name
-                        arguments = json.loads(tool_call.function.arguments)
+                    # Handle both ToolCall dataclass and OpenAI object.
+                    # Guard the parse itself so malformed/truncated argument
+                    # JSON is reported back to the model instead of aborting
+                    # the whole run (matches chat_completion_with_tools_stream).
+                    _tool_call_id = tool_call.id if hasattr(tool_call, 'id') else tool_call.get('id')
+                    try:
+                        if isinstance(tool_call, ToolCall):
+                            function_name = tool_call.function["name"]
+                            arguments = json.loads(tool_call.function["arguments"])
+                        else:
+                            function_name = tool_call.function.name
+                            arguments = json.loads(tool_call.function.arguments)
+                    except (json.JSONDecodeError, KeyError, AttributeError, TypeError) as e:
+                        logging.warning(f"Failed to parse tool call arguments: {e}")
+                        messages.append({
+                            "role": "tool",
+                            "tool_call_id": _tool_call_id,
+                            "content": json.dumps({"error": f"Invalid arguments JSON: {e}"}),
+                        })
+                        continue
                     
                     # Always trigger callback for tool call tracking (even when verbose=False)
                     display_tool_call_fn = _get_display_tool_call()
@@ -1921,7 +1934,6 @@ class OpenAIClient:
                     # Capture failures and report them back to the model instead of
                     # aborting the whole run (safe-by-default, matching the streaming
                     # path chat_completion_with_tools_stream).
-                    _tool_call_id = tool_call.id if hasattr(tool_call, 'id') else tool_call.get('id')
                     try:
                         tool_result = execute_tool_fn(
                             function_name,
@@ -2209,13 +2221,26 @@ class OpenAIClient:
                 # replies for this turn (keeps tool replies consecutive).
                 _deferred_media_followups = []
                 for tool_call in tool_calls:
-                    # Handle both ToolCall dataclass and OpenAI object
-                    if isinstance(tool_call, ToolCall):
-                        function_name = tool_call.function["name"]
-                        arguments = json.loads(tool_call.function["arguments"])
-                    else:
-                        function_name = tool_call.function.name
-                        arguments = json.loads(tool_call.function.arguments)
+                    # Handle both ToolCall dataclass and OpenAI object.
+                    # Guard the parse itself so malformed/truncated argument
+                    # JSON is reported back to the model instead of aborting
+                    # the whole run (matches chat_completion_with_tools_stream).
+                    _tool_call_id = tool_call.id if hasattr(tool_call, 'id') else tool_call.get('id')
+                    try:
+                        if isinstance(tool_call, ToolCall):
+                            function_name = tool_call.function["name"]
+                            arguments = json.loads(tool_call.function["arguments"])
+                        else:
+                            function_name = tool_call.function.name
+                            arguments = json.loads(tool_call.function.arguments)
+                    except (json.JSONDecodeError, KeyError, AttributeError, TypeError) as e:
+                        logging.warning(f"Failed to parse tool call arguments: {e}")
+                        messages.append({
+                            "role": "tool",
+                            "tool_call_id": _tool_call_id,
+                            "content": json.dumps({"error": f"Invalid arguments JSON: {e}"}),
+                        })
+                        continue
                     
                     # Always trigger callback for tool call tracking (even when verbose=False)
                     display_tool_call_fn = _get_display_tool_call()
@@ -2228,7 +2253,6 @@ class OpenAIClient:
                     # Capture failures and report them back to the model instead of
                     # aborting the whole run (safe-by-default, matching the streaming
                     # path chat_completion_with_tools_stream).
-                    _tool_call_id = tool_call.id if hasattr(tool_call, 'id') else tool_call.get('id')
                     try:
                         if asyncio.iscoroutinefunction(execute_tool_fn):
                             tool_result = await execute_tool_fn(

src/praisonai-agents/praisonaiagents/session/store.py

@@ -694,8 +694,12 @@ class DefaultSessionStore:
         if self.retention == RETENTION_KEEP_ALL:
             return
 
-        overflow = session.messages[:-window]
-        recent = session.messages[-window:]
+        # Nudge the window boundary forward past any leading orphaned
+        # ``role="tool"`` results so the retained tail never starts mid
+        # tool-exchange (would be rejected by strict providers). Reuses the
+        # same guard as get_chat_history (Issue #3089).
+        recent = SessionData._trim_preserving_tool_exchanges(session.messages, window)
+        overflow = session.messages[: len(session.messages) - len(recent)]
 
         if self.retention == RETENTION_TRUNCATE:
             session.messages = recent