diff --git a/src/praisonai-agents/praisonaiagents/agents/agents.py b/src/praisonai-agents/praisonaiagents/agents/agents.py index 666595861..7ea98c693 100644 --- a/src/praisonai-agents/praisonaiagents/agents/agents.py +++ b/src/praisonai-agents/praisonaiagents/agents/agents.py @@ -388,6 +388,18 @@ def _build_execution_context(agents_instance, task_id, skip_memory_init=False): # Build context first to include in task prompt context_text = "" + # Inter-task context / validation feedback assembled by the workflow process + # engine (Process._build_task_context) is stored on the task; fold it in so + # downstream/retried tasks actually see upstream output and rejection reasons. + extra_context = getattr(task, '_execution_context', None) + if extra_context: + context_text = extra_context + # NOTE: We intentionally do NOT clear _execution_context here. Task + # retries (guardrail/completion failures) re-enter this helper via the + # run_task/arun_task retry loops, and clearing would strip the upstream + # output + validation feedback the retry needs. The Process engine owns + # this field's lifecycle: it re-sets it before each task yield and resets + # every task's _execution_context to None before selecting the next task. if task.context: context_results = [] # Collect contexts then de-duplicate for context_item in task.context: @@ -404,7 +416,11 @@ def _build_execution_context(agents_instance, task_id, skip_memory_init=False): for i, ctx in enumerate(unique_contexts): logger.debug(f"Context {i+1}: {ctx[:100]}...") context_separator = '\n\n' - context_text = context_separator.join(unique_contexts) + joined_contexts = context_separator.join(unique_contexts) + if context_text and joined_contexts: + context_text = context_text + context_separator + joined_contexts + elif joined_contexts: + context_text = joined_contexts # Build task prompt using DRY helper task_prompt = _prepare_task_prompt(task, task_description, context_text) diff --git a/src/praisonai-agents/praisonaiagents/mcp/mcp.py b/src/praisonai-agents/praisonaiagents/mcp/mcp.py index f143538c1..6d9eb3a71 100644 --- a/src/praisonai-agents/praisonaiagents/mcp/mcp.py +++ b/src/praisonai-agents/praisonaiagents/mcp/mcp.py @@ -78,12 +78,14 @@ async def _run_async(self): response_queue, kind, name, arguments = item try: if kind == "resource": - result = await session.read_resource(name) + result = await asyncio.wait_for(session.read_resource(name), timeout=self.timeout) elif kind == "prompt": - result = await session.get_prompt(name, arguments or None) + result = await asyncio.wait_for(session.get_prompt(name, arguments or None), timeout=self.timeout) else: - result = await session.call_tool(name, arguments) + result = await asyncio.wait_for(session.call_tool(name, arguments), timeout=self.timeout) response_queue.put((True, result)) + except asyncio.TimeoutError: + response_queue.put((False, f"MCP {kind} call timed out after {self.timeout} seconds (server side)")) except Exception as e: response_queue.put((False, str(e))) except queue.Empty: diff --git a/src/praisonai-agents/praisonaiagents/workflows/workflows.py b/src/praisonai-agents/praisonaiagents/workflows/workflows.py index 7186b7136..d321fe5b7 100644 --- a/src/praisonai-agents/praisonaiagents/workflows/workflows.py +++ b/src/praisonai-agents/praisonaiagents/workflows/workflows.py @@ -2821,21 +2821,24 @@ def execute_item(idx_item_tuple, opt_prev=optimized_previous): emitter.clear_branch() with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - futures = [ - executor.submit(copy_context_to_callable(lambda pair=(idx, item): execute_item(pair))) - for idx, item in enumerate(items) - ] + # Map each future back to its original index so error attribution + # survives even if execute_item raises before returning its index. + future_to_idx = {} + for idx, item in enumerate(items): + fut = executor.submit(copy_context_to_callable(lambda pair=(idx, item): execute_item(pair))) + future_to_idx[fut] = idx # Collect results in order indexed_results = [] - for future in concurrent.futures.as_completed(futures): + for future in concurrent.futures.as_completed(future_to_idx): + idx = future_to_idx[future] try: - idx, step_result = future.result() + _, step_result = future.result() indexed_results.append((idx, step_result)) if verbose: print(f" ✓ Item {idx + 1}/{num_items} complete") except Exception as e: - logger.error(f"Parallel loop iteration failed: {e}") + logger.error(f"Parallel loop iteration {idx} failed: {e}") indexed_results.append((idx, {"step": f"loop_{idx}", "output": f"Error: {e}"})) # Sort by index to maintain order