diff --git a/src/praisonai-agents/praisonaiagents/agent/agent.py b/src/praisonai-agents/praisonaiagents/agent/agent.py index 54e95881f..5fbc78065 100644 --- a/src/praisonai-agents/praisonaiagents/agent/agent.py +++ b/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 diff --git a/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py b/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py index ffa569d4d..4b83e634b 100644 --- a/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py +++ b/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)(): diff --git a/src/praisonai-agents/praisonaiagents/agent/execution_mixin.py b/src/praisonai-agents/praisonaiagents/agent/execution_mixin.py index c4ed4479f..bb76c870d 100644 --- a/src/praisonai-agents/praisonaiagents/agent/execution_mixin.py +++ b/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 diff --git a/src/praisonai-agents/praisonaiagents/agent/tool_execution.py b/src/praisonai-agents/praisonaiagents/agent/tool_execution.py index 8b62d62a9..ab46da850 100644 --- a/src/praisonai-agents/praisonaiagents/agent/tool_execution.py +++ b/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): diff --git a/src/praisonai-agents/praisonaiagents/mcp/mcp.py b/src/praisonai-agents/praisonaiagents/mcp/mcp.py index 29de27bf6..73f2181fb 100644 --- a/src/praisonai-agents/praisonaiagents/mcp/mcp.py +++ b/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. diff --git a/src/praisonai-agents/praisonaiagents/process/process.py b/src/praisonai-agents/praisonaiagents/process/process.py index 3f3d5a160..697f2eed0 100644 --- a/src/praisonai-agents/praisonaiagents/process/process.py +++ b/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: diff --git a/src/praisonai-agents/praisonaiagents/skills/capability_validator.py b/src/praisonai-agents/praisonaiagents/skills/capability_validator.py index 1fc5cbe44..e1fbd3208 100644 --- a/src/praisonai-agents/praisonaiagents/skills/capability_validator.py +++ b/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