fix: aworkflow loop expansion, concurrent turn-tool race, MCP skill-gate discovery

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

@@ -20,7 +20,7 @@
 from .tool_execution import ToolExecutionMixin, BackoffPolicy
 from .chat_handler import ChatHandlerMixin
 from .session_manager import SessionManagerMixin
-from .async_safety import AsyncSafeState
+from .async_safety import AsyncSafeState, DualLock
 # NOTE: UnifiedExecutionMixin is deprecated and unused by any production path
 # (Issue #2644). It is kept in the MRO for backward compatibility during the
 # deprecation cycle and will be removed afterwards.
@@ -2224,7 +2224,10 @@ def __init__(
         # Per-turn tool-name buffer feeding the self-improve review policy.
         # Populated in _execute_tool_with_context, reset each chat turn, and
         # read by _trigger_after_agent_hook when tools_used is not passed.
+        # Guarded by a DualLock (like chat_history) so concurrent chat()/achat()
+        # turns on the same Agent instance don't corrupt each other's buffer.
         self._turn_tools_used = []
+        self._turn_tools_lock = DualLock()
 
         # Database persistence (lazy - no imports until used)
         self._db = db

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

@@ -2447,8 +2447,7 @@ def _chat_impl(self, prompt, temperature, tools, output_json, output_pydantic, r
         """Internal chat implementation (extracted for trace wrapping)."""
         # Reset the per-turn tool buffer so the self-improve review policy only
         # sees tools used in this turn (not during a nested skill-review turn).
-        if not getattr(self, "_in_skill_review", False):
-            self._turn_tools_used = []
+        self._reset_turn_tools()
         # Apply rate limiter if configured (before any LLM call)
         if self._rate_limiter is not None:
             self._rate_limiter.acquire()
@@ -3049,8 +3048,7 @@ async def _achat_impl(self, prompt, temperature, tools, output_json, output_pyda
         """Internal async chat implementation (extracted for trace wrapping)."""
         # Reset the per-turn tool buffer so the self-improve review policy only
         # sees tools used in this turn (not during a nested skill-review turn).
-        if not getattr(self, "_in_skill_review", False):
-            self._turn_tools_used = []
+        self._reset_turn_tools()
         # C2 - cooperative cancellation: abort early if a pre-set token is given
         _cancel = cancel_token if cancel_token is not None else getattr(self, "interrupt_controller", None)
         if _cancel is not None and getattr(_cancel, "is_set", lambda: False)():

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

@@ -1146,13 +1146,9 @@ async def execute_tool_async(self, function_name: str, arguments: Dict[str, Any]
         # Record the tool name for this turn so the self-improve review policy
         # sees async tool usage too (issue #3037). Mirrors the sync path in
         # _execute_tool_with_context; skipped during a guarded review turn so
-        # the review's own calls are not tracked and cannot recurse.
-        if not getattr(self, "_in_skill_review", False):
-            turn_tools = getattr(self, "_turn_tools_used", None)
-            if turn_tools is None:
-                self._turn_tools_used = []
-                turn_tools = self._turn_tools_used
-            turn_tools.append(function_name)
+        # the review's own calls are not tracked and cannot recurse. Locked so
+        # concurrent turns on the same Agent don't corrupt the buffer (#3307).
+        self._record_turn_tool(function_name)
 
         # Enforce BEFORE_TOOL/AFTER_TOOL security hooks for every async caller,
         # mirroring the sync execute_tool path in tool_execution.py. Without

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

@@ -471,6 +471,46 @@ def _get_tool_middleware_manager(self):
             self._middleware_manager = manager
         return manager if manager.has_tool_hooks else None
 
+    def _get_turn_tools_lock(self):
+        """Return the DualLock guarding the per-turn tool buffer.
+
+        Falls back to a lazily-created lock so subclasses or objects that
+        predate the lock attribute stay safe. The lock protects
+        ``_turn_tools_used`` from corruption when concurrent chat()/achat()
+        turns run on the same Agent instance (issue #3307).
+        """
+        lock = getattr(self, "_turn_tools_lock", None)
+        if lock is None:
+            from .async_safety import DualLock
+            lock = DualLock()
+            self._turn_tools_lock = lock
+        return lock
+
+    def _reset_turn_tools(self):
+        """Reset the per-turn tool buffer under lock (start of a chat turn)."""
+        if getattr(self, "_in_skill_review", False):
+            return
+        with self._get_turn_tools_lock().sync():
+            self._turn_tools_used = []
+
+    def _record_turn_tool(self, function_name):
+        """Append a tool name to the per-turn buffer under lock."""
+        if getattr(self, "_in_skill_review", False):
+            return
+        with self._get_turn_tools_lock().sync():
+            turn_tools = getattr(self, "_turn_tools_used", None)
+            if turn_tools is None:
+                turn_tools = []
+                self._turn_tools_used = turn_tools
+            turn_tools.append(function_name)
+
+    def _drain_turn_tools(self):
+        """Return a copy of the per-turn buffer and clear it, under lock."""
+        with self._get_turn_tools_lock().sync():
+            tools_used = list(getattr(self, "_turn_tools_used", []) or [])
+            self._turn_tools_used = []
+        return tools_used
+
     def _execute_tool_with_context(self, function_name, arguments, state, tool_call_id=None):
         """Execute tool within injection context, with optional output truncation.
 
@@ -487,13 +527,9 @@ def _execute_tool_with_context(self, function_name, arguments, state, tool_call_
 
         # Record the tool name for this turn so the self-improve review policy
         # can see what ran (issue #3037). Skipped during a guarded review turn
-        # to avoid tracking the review's own tool calls or recursing.
-        if not getattr(self, "_in_skill_review", False):
-            turn_tools = getattr(self, "_turn_tools_used", None)
-            if turn_tools is None:
-                self._turn_tools_used = []
-                turn_tools = self._turn_tools_used
-            turn_tools.append(function_name)
+        # to avoid tracking the review's own tool calls or recursing. Locked so
+        # concurrent turns on the same Agent don't corrupt the buffer (#3307).
+        self._record_turn_tool(function_name)
 
         # Emit tool call start event (zero overhead when not set)
         _trace_emitter = get_context_emitter()
@@ -1132,8 +1168,9 @@ def _trigger_after_agent_hook(self, prompt, response, start_time, tools_used=Non
         # tools_used, so without this the review policy always sees an empty
         # list and never runs. Consume the buffer so the next turn starts clean.
         if tools_used is None:
-            tools_used = list(getattr(self, "_turn_tools_used", []) or [])
-        self._turn_tools_used = []
+            tools_used = self._drain_turn_tools()
+        else:
+            self._drain_turn_tools()
         # Trigger AFTER_AGENT hook (only build the input if a hook is actually registered)
         from ..hooks import HookEvent
         if self._hook_runner.registry.has_hooks(HookEvent.AFTER_AGENT):
@@ -1189,8 +1226,9 @@ async def _atrigger_after_agent_hook(self, prompt, response, start_time, tools_u
         # Default tools_used from the per-turn buffer when the caller did not
         # pass it explicitly (issue #3037); mirrors the sync path.
         if tools_used is None:
-            tools_used = list(getattr(self, "_turn_tools_used", []) or [])
-        self._turn_tools_used = []
+            tools_used = self._drain_turn_tools()
+        else:
+            self._drain_turn_tools()
         # Trigger AFTER_AGENT hook (only build the input if a hook is actually registered)
         from ..hooks import HookEvent
         if self._hook_runner.registry.has_hooks(HookEvent.AFTER_AGENT):

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

@@ -283,6 +283,18 @@ class MCP:
         ```
     """
 
+    # Process-level registry of active MCP server names (sanitized prefixes),
+    # mirroring how tools/registry.py tracks tool names. Populated when a
+    # server is namespaced via with_tool_prefix(), so skills' capability
+    # validator can discover which MCP servers are actually connected
+    # (issue #3307) instead of always seeing an empty set.
+    _active_server_names: set = set()
+
+    @classmethod
+    def list_active_server_names(cls) -> set:
+        """Return the set of active MCP server names (sanitized prefixes)."""
+        return set(cls._active_server_names)
+
     def __init__(self, command_or_string=None, args=None, *, command=None, timeout=60, debug=False, 
                  allowed_tools: Optional[List[str]] = None, disabled_tools: Optional[List[str]] = None, **kwargs):
         """
@@ -834,6 +846,14 @@ def with_tool_prefix(self, prefix: str) -> "MCP":
 
         self._tool_prefix = sanitized
 
+        # Track this server name so skills' capability validator can see that
+        # it is connected (issue #3307). Register both the caller-supplied
+        # name and its sanitized prefix, since skill requirements may use
+        # either spelling.
+        MCP._active_server_names.add(sanitized)
+        if prefix:
+            MCP._active_server_names.add(prefix)
+
         # Rename already-generated callable tools. Dispatch inside each
         # wrapper closes over the original tool name, so only the public
         # __name__/__qualname__ needs updating for schema construction.

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

@@ -616,11 +616,33 @@ async def aworkflow(self) -> AsyncGenerator[str, None]:
             start_task = list(self.tasks.values())[0]
             logging.debug(f"No start task marked, using first task: {start_task.name}")
 
+        # If loop type and no input_file, default to tasks.csv
+        if start_task and start_task.task_type == "loop" and not start_task.input_file:
+            start_task.input_file = "tasks.csv"
+
+        # --- If loop + input_file, read file & create tasks using consolidated helper
+        # Mirrors the sync workflow() pre-expansion so async workflows with a
+        # CSV-driven loop start task run once per row instead of a single time.
+        if start_task and start_task.task_type == "loop" and getattr(start_task, "input_file", None):
+            try:
+                parent_loop_task = start_task
+                parent_input_file = parent_loop_task.input_file
+                self._create_loop_subtasks(parent_loop_task, decision_mode=True)
+                subtasks = [
+                    t for t in self.tasks.values()
+                    if t.name.startswith(parent_loop_task.name + "_")
+                ]
+                if subtasks:
+                    parent_loop_task.status = "completed"
+                    parent_loop_task._subtasks_created = True
+                    start_task = next((t for t in subtasks if t.is_start), subtasks[0])
+                    logging.info(f"Created {len(subtasks)} tasks from: {parent_input_file}")
+            except Exception as e:
+                logging.error(f"Failed to read file tasks: {e}")
+
         current_task = start_task
         visited_tasks = set()
 
-        # TODO: start task with loop feature is not available in aworkflow method
-
         while current_task:
             current_iter += 1
             if current_iter > self.max_iter:

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

@@ -87,7 +87,6 @@ def __init__(self, enforcement_level: EnforcementLevel = EnforcementLevel.WARN):
         """
         self.enforcement_level = enforcement_level
         self._tool_cache: Optional[Set[str]] = None
-        self._server_cache: Optional[Set[str]] = None
 
     def validate_skill(
         self,
@@ -204,12 +203,25 @@ def _get_available_tools(self) -> Set[str]:
         return self._tool_cache
 
     def _get_available_servers(self) -> Set[str]:
-        """Get set of available MCP server names."""
-        if self._server_cache is None:
-            # TODO: Implement MCP server discovery
-            # For now, return empty set - this can be extended later
-            self._server_cache = set()
-        return self._server_cache
+        """Get set of available MCP server names.
+
+        Derives names from the MCP client registry of servers that have been
+        namespaced (via ``with_tool_prefix``) in this process. Without this an
+        MCP-server-gated skill could never pass STRICT validation because the
+        set was always empty (issue #3307).
+
+        Queried live (not cached) because MCP servers register lazily: a skill
+        may be validated before its required server connects. Caching the first
+        empty snapshot would leave STRICT validation permanently reporting the
+        skill unavailable even after the server registers. The registry read is
+        just a cheap ``set`` copy, so there is no hot-path cost.
+        """
+        try:
+            from ..mcp.mcp import MCP
+            return MCP.list_active_server_names()
+        except ImportError:
+            logger.debug("MCP not available")
+            return set()
 
     def _log_validation_result(self, result: ValidationResult) -> None:
         """Log validation result based on enforcement level."""
@@ -233,4 +245,3 @@ def _log_validation_result(self, result: ValidationResult) -> None:
     def clear_cache(self) -> None:
         """Clear cached capability information."""
         self._tool_cache = None
-        self._server_cache = None