From 36da0007e6d6961a4e62d9eaab79ad23615adb73 Mon Sep 17 00:00:00 2001 From: "praisonai-triage-agent[bot]" <272766704+praisonai-triage-agent[bot]@users.noreply.github.com> Date: Tue, 28 Jul 2026 07:18:39 +0000 Subject: [PATCH 1/2] fix: MCP per-call timeout, wire workflow context into prompt, fix parallel loop idx (fixes #3457) - mcp.py: wrap session.call_tool/read_resource/get_prompt in asyncio.wait_for so a hung server fails one request without wedging the runner loop forever - agents.py: fold task._execution_context (inter-task context + validation feedback built by Process._build_task_context) into the prompt builder, and clear it after consumption so it doesn't leak into later runs - workflows.py: track future->idx map in parallel loop so error attribution survives exceptions instead of using an unbound/stale idx Co-authored-by: MervinPraison --- .../praisonaiagents/agents/agents.py | 14 +++++++++++++- src/praisonai-agents/praisonaiagents/mcp/mcp.py | 8 +++++--- .../praisonaiagents/workflows/workflows.py | 17 ++++++++++------- 3 files changed, 28 insertions(+), 11 deletions(-) diff --git a/src/praisonai-agents/praisonaiagents/agents/agents.py b/src/praisonai-agents/praisonaiagents/agents/agents.py index 666595861..a017e742c 100644 --- a/src/praisonai-agents/praisonaiagents/agents/agents.py +++ b/src/praisonai-agents/praisonaiagents/agents/agents.py @@ -388,6 +388,14 @@ 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 + # Clear after consumption so it doesn't leak into a later run of this task + task._execution_context = "" if task.context: context_results = [] # Collect contexts then de-duplicate for context_item in task.context: @@ -404,7 +412,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 From de5a95d549ee259cd91f2259acb4e75df236797f Mon Sep 17 00:00:00 2001 From: "praisonai-triage-agent[bot]" <272766704+praisonai-triage-agent[bot]@users.noreply.github.com> Date: Tue, 28 Jul 2026 07:24:29 +0000 Subject: [PATCH 2/2] fix: preserve workflow _execution_context across task retries Do not clear task._execution_context inside _build_execution_context. The run_task/arun_task retry loops re-enter this helper on guardrail / completion failures; clearing stripped the upstream output and validation feedback the retry needs (Greptile P1). The Process engine already owns this field's lifecycle (re-set before each yield, reset to None before selecting the next task), so no cross-task leak. Co-authored-by: Mervin Praison --- src/praisonai-agents/praisonaiagents/agents/agents.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/praisonai-agents/praisonaiagents/agents/agents.py b/src/praisonai-agents/praisonaiagents/agents/agents.py index a017e742c..7ea98c693 100644 --- a/src/praisonai-agents/praisonaiagents/agents/agents.py +++ b/src/praisonai-agents/praisonaiagents/agents/agents.py @@ -394,8 +394,12 @@ def _build_execution_context(agents_instance, task_id, skip_memory_init=False): extra_context = getattr(task, '_execution_context', None) if extra_context: context_text = extra_context - # Clear after consumption so it doesn't leak into a later run of this task - task._execution_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: