fix: async loop expansion, tool-tracking race, MCP skill-gate (fixes #3307)

Gap 1: port loop-task pre-expansion from workflow() into aworkflow() so
async workflows with a CSV-driven loop start task expand per-row instead
of running once.

Gap 2: guard the per-turn _turn_tools_used buffer with the same
AsyncSafeState lock that protects chat_history, routing all sites through
_reset/_record/_drain helpers so concurrent chat()/achat() turns on one
Agent no longer corrupt hook/self-improve tool data. Backward-compatible
_turn_tools_used property retained.

Gap 3: track namespaced MCP server names in a process-level registry
(MCP.list_active_server_names(), populated in with_tool_prefix) and read
it from CapabilityValidator._get_available_servers so MCP-server-gated
skills can pass STRICT validation instead of always failing closed.

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

Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>

Author: praisonai-triage-agent[bot]

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

@@ -282,18 +282,19 @@ class MCP:
         agent.start("What is the stock price of Tesla?")
         ```
     """
-    
-    # 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.
+
+    # Process-level registry of sanitized MCP server names that have been
+    # namespaced via with_tool_prefix(), mirroring how tools/registry.py tracks
+    # tool names. Lets skills' CapabilityValidator discover connected servers
+    # instead of always failing closed (issue #3307).
     _active_server_names: set = set()
+    _active_server_names_lock = threading.Lock()
 
     @classmethod
     def list_active_server_names(cls) -> set:
-        """Return the set of active MCP server names (sanitized prefixes)."""
-        return set(cls._active_server_names)
+        """Return the set of sanitized names of MCP servers namespaced this run."""
+        with cls._active_server_names_lock:
+            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):
@@ -846,13 +847,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)
+        # Record this server in the process-level registry so skills'
+        # CapabilityValidator can discover it (issue #3307). Store both the
+        # original name and its sanitized form so a skill requirement matches
+        # regardless of which spelling it declares.
+        with type(self)._active_server_names_lock:
+            if prefix:
+                type(self)._active_server_names.add(prefix)
+            type(self)._active_server_names.add(sanitized)
 
         # Rename already-generated callable tools. Dispatch inside each
         # wrapper closes over the original tool name, so only the public

src/praisonai-agents/tests/test_aworkflow_loop_expansion.py

@@ -0,0 +1,89 @@
+"""Regression test for issue #3307 Gap 1.
+
+`Process.aworkflow()` must pre-expand a loop-type start task into one subtask
+per input-file row, mirroring the sync `Process.workflow()` behaviour. Before
+the fix, the async engine skipped this step (only a TODO comment) and ran the
+loop task once as an ordinary task.
+"""
+
+import os
+import sys
+import asyncio
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from praisonaiagents import Task
+from praisonaiagents.process.process import Process
+
+
+def _make_loop_process(tmp_path):
+    csv_path = os.path.join(tmp_path, "rows.csv")
+    with open(csv_path, "w") as fh:
+        fh.write("alpha\nbeta\ngamma\n")
+
+    loop_task = Task(
+        name="loop_start",
+        description="process each row",
+        task_type="loop",
+        input_file=csv_path,
+        is_start=True,
+    )
+    tasks = {"loop_start": loop_task}
+    return Process(tasks=tasks, agents=[]), loop_task
+
+
+def test_aworkflow_expands_loop_start_task(tmp_path):
+    process, loop_task = _make_loop_process(str(tmp_path))
+
+    async def drive():
+        gen = process.aworkflow()
+        # Pull the first yielded task id; the pre-expansion runs before the
+        # first yield, so we only need one step to observe the effect.
+        try:
+            await gen.__anext__()
+        except StopAsyncIteration:
+            pass
+        await gen.aclose()
+
+    asyncio.run(drive())
+
+    subtasks = [t for t in process.tasks.values()
+                if t.name.startswith("loop_start_")]
+    # One subtask per CSV row (3 rows) must have been created.
+    assert len(subtasks) == 3
+    # Parent loop task is marked completed once expanded.
+    assert loop_task.status == "completed"
+
+
+def test_sync_and_async_loop_expansion_match(tmp_path):
+    async_process, _ = _make_loop_process(str(tmp_path))
+
+    async def drive():
+        gen = async_process.aworkflow()
+        try:
+            await gen.__anext__()
+        except StopAsyncIteration:
+            pass
+        await gen.aclose()
+
+    asyncio.run(drive())
+    async_subtasks = {t.name for t in async_process.tasks.values()
+                      if t.name.startswith("loop_start_")}
+
+    sync_process, _ = _make_loop_process(str(tmp_path))
+    gen = sync_process.workflow()
+    try:
+        next(gen)
+    except StopIteration:
+        pass
+    gen.close()
+    sync_subtasks = {t.name for t in sync_process.tasks.values()
+                     if t.name.startswith("loop_start_")}
+
+    assert async_subtasks == sync_subtasks
+    assert len(async_subtasks) == 3
+
+
+if __name__ == "__main__":
+    import pytest
+    pytest.main([__file__, "-v"])

src/praisonai-agents/tests/unit/skills/test_capability_validator.py

@@ -254,6 +254,38 @@ def test_validate_skill_missing_env_vars_strict(self):
         assert len(result.warnings) == 0
         assert len(result.errors) == 1
 
+    def test_available_servers_read_from_mcp_registry(self):
+        """Issue #3307 Gap 3: _get_available_servers must reflect active MCP servers.
+
+        Previously it was a stub returning an empty set, so any skill with an
+        MCP-server requirement failed closed under STRICT enforcement no matter
+        what was connected.
+        """
+        validator = CapabilityValidator(EnforcementLevel.STRICT)
+        with patch(
+            "praisonaiagents.mcp.mcp.MCP.list_active_server_names",
+            return_value={"filesystem"},
+        ):
+            servers = validator._get_available_servers()
+        assert "filesystem" in servers
+
+    def test_mcp_gated_skill_passes_strict_when_server_active(self):
+        """Issue #3307 Gap 3: an MCP-server-gated skill can now pass STRICT."""
+        requirements = SkillRequirements(servers=["filesystem"])
+        skill = SkillProperties(
+            name="fs-skill",
+            description="needs filesystem MCP server",
+            requirements=requirements,
+        )
+        validator = CapabilityValidator(EnforcementLevel.STRICT)
+        result = validator.validate_skill(
+            skill,
+            available_tools=set(),
+            available_servers={"filesystem"},
+        )
+        assert result.state != SkillState.UNAVAILABLE
+        assert result.satisfied_servers == ["filesystem"]
+
     def test_validation_result_to_dict(self):
         """Test ValidationResult serialization."""
         result = ValidationResult(

src/praisonai-agents/tests/unit/skills/test_self_improve.py

@@ -416,3 +416,33 @@ def start_job(self, func, job_id=None):
     # Both scoped to the session but distinct so neither replaces the other.
     assert all(jid.startswith("self-improve:sess:") for jid in job_ids)
     assert job_ids[0] != job_ids[1]
+
+
+def test_turn_tools_helpers_are_thread_safe():
+    """Issue #3307 Gap 2: concurrent record/drain must not lose tool names."""
+    import threading
+
+    agent = Agent(instructions="x", self_improve=True)
+    agent._reset_turn_tools()
+
+    errors = []
+
+    def worker():
+        try:
+            for _ in range(200):
+                agent._record_turn_tool("t")
+        except Exception as e:  # pragma: no cover - defensive
+            errors.append(e)
+
+    threads = [threading.Thread(target=worker) for _ in range(8)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert not errors
+    drained = agent._drain_turn_tools()
+    # 8 threads * 200 appends, none lost to unlocked list mutation.
+    assert len(drained) == 8 * 200
+    # Buffer is empty after draining.
+    assert agent._drain_turn_tools() == []