From b19e4da70b5cc8ce39cd1eed92029d9d6ae27947 Mon Sep 17 00:00:00 2001 From: "praisonai-triage-agent[bot]" <272766704+praisonai-triage-agent[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 10:36:46 +0000 Subject: [PATCH 1/2] fix: async loop expansion, tool-tracking race, MCP skill-gate (fixes #3307) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../praisonaiagents/mcp/mcp.py | 32 +++---- .../tests/test_aworkflow_loop_expansion.py | 89 +++++++++++++++++++ .../unit/skills/test_capability_validator.py | 32 +++++++ .../tests/unit/skills/test_self_improve.py | 30 +++++++ 4 files changed, 168 insertions(+), 15 deletions(-) create mode 100644 src/praisonai-agents/tests/test_aworkflow_loop_expansion.py diff --git a/src/praisonai-agents/praisonaiagents/mcp/mcp.py b/src/praisonai-agents/praisonaiagents/mcp/mcp.py index 73f2181fbe..f143538c1e 100644 --- a/src/praisonai-agents/praisonaiagents/mcp/mcp.py +++ b/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 diff --git a/src/praisonai-agents/tests/test_aworkflow_loop_expansion.py b/src/praisonai-agents/tests/test_aworkflow_loop_expansion.py new file mode 100644 index 0000000000..bcdad0416e --- /dev/null +++ b/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"]) diff --git a/src/praisonai-agents/tests/unit/skills/test_capability_validator.py b/src/praisonai-agents/tests/unit/skills/test_capability_validator.py index 573ce928b1..9659285add 100644 --- a/src/praisonai-agents/tests/unit/skills/test_capability_validator.py +++ b/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( diff --git a/src/praisonai-agents/tests/unit/skills/test_self_improve.py b/src/praisonai-agents/tests/unit/skills/test_self_improve.py index e7416bd0de..1574820839 100644 --- a/src/praisonai-agents/tests/unit/skills/test_self_improve.py +++ b/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() == [] From d15657ba2132c65f25ab21ccceae8ba3bb67a4ad Mon Sep 17 00:00:00 2001 From: "praisonai-triage-agent[bot]" <272766704+praisonai-triage-agent[bot]@users.noreply.github.com> Date: Thu, 23 Jul 2026 10:56:07 +0000 Subject: [PATCH 2/2] fix(skills): read MCP server registry live to avoid stale STRICT gate CapabilityValidator cached the MCP server snapshot once, so servers that connected after the first validation stayed invisible under STRICT enforcement. Read the process-level registry live each call (cheap set-copy under lock); tool cache is unchanged. (#3307) Co-authored-by: Mervin Praison --- .../skills/capability_validator.py | 13 ++++++------- .../unit/skills/test_capability_validator.py | 19 +++++++++++++++++++ 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/src/praisonai-agents/praisonaiagents/skills/capability_validator.py b/src/praisonai-agents/praisonaiagents/skills/capability_validator.py index e1fbd3208f..6b65b4ad3c 100644 --- a/src/praisonai-agents/praisonaiagents/skills/capability_validator.py +++ b/src/praisonai-agents/praisonaiagents/skills/capability_validator.py @@ -203,22 +203,21 @@ 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. + """Get set of available MCP server names from the active MCP registry. 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. + The MCP registry is populated dynamically as servers connect during a + run, so this is read live (not cached) to avoid a stale snapshot that + would keep rejecting servers registered after the first validation. + The read is a cheap set copy under a lock, so there is no hot-path cost. """ try: from ..mcp.mcp import MCP - return MCP.list_active_server_names() + return set(MCP.list_active_server_names()) except ImportError: logger.debug("MCP not available") return set() diff --git a/src/praisonai-agents/tests/unit/skills/test_capability_validator.py b/src/praisonai-agents/tests/unit/skills/test_capability_validator.py index 9659285add..4210db191d 100644 --- a/src/praisonai-agents/tests/unit/skills/test_capability_validator.py +++ b/src/praisonai-agents/tests/unit/skills/test_capability_validator.py @@ -269,6 +269,25 @@ def test_available_servers_read_from_mcp_registry(self): servers = validator._get_available_servers() assert "filesystem" in servers + def test_available_servers_read_live_not_cached(self): + """Issue #3307 Gap 3: server availability must not be cached stale. + + The MCP registry fills in as servers connect during a run, so a server + registered after the first validation must become visible without an + explicit clear_cache() call. + """ + validator = CapabilityValidator(EnforcementLevel.STRICT) + with patch( + "praisonaiagents.mcp.mcp.MCP.list_active_server_names", + return_value=set(), + ): + assert validator._get_available_servers() == set() + with patch( + "praisonaiagents.mcp.mcp.MCP.list_active_server_names", + return_value={"filesystem"}, + ): + assert "filesystem" in validator._get_available_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"])