[AAASM-4848] 🐛 (python): Surface positional tool args to content policy + assert pydantic_ai execution hook

agent_assembly/adapters/_shared/positional_args.py

@@ -0,0 +1,46 @@
+"""Fold positional tool-call arguments into the governance-visible arg mapping.
+
+Framework adapters build the ``tool_args`` mapping the policy inspects from a
+tool call's *keyword* arguments. A tool invoked positionally
+(``tool.invoke("secret")``) would otherwise present an empty or partial mapping,
+so argument-CONTENT policy rules go blind to the positional values — the
+allow/deny gate still fires on the tool name, but content rules never see the
+data. Folding the positionals in (mirroring the smolagents/mcp handling) keeps
+content policy able to inspect them.
+
+This module deliberately imports only the stdlib so every leaf adapter can reuse
+it without risking an import cycle through ``_shared.tool_governance`` (which
+itself imports the CrewAI leaf helpers).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+
+def merge_positional_tool_args(tool_args: dict[str, Any], args: tuple[Any, ...]) -> dict[str, Any]:
+    """Fold positional call arguments into ``tool_args`` in place and return it.
+
+    A single positional ``Mapping`` (the ``tool(payload)`` convenience shape) is
+    flattened by key; any other positionals are recorded under ``arg{index}`` so
+    their values remain visible to content policy even when the tool's real
+    parameter names can't be recovered from the wrapper's ``*args`` signature.
+    Existing keys win via ``setdefault`` so a real keyword argument is never
+    overwritten by a positional fallback.
+
+    Args:
+        tool_args: Mapping already built from the call's keyword arguments.
+        args: The positional arguments the tool was invoked with (excluding
+            ``self``).
+
+    Returns:
+        The same ``tool_args`` dict, with positional values folded in.
+    """
+    if len(args) == 1 and isinstance(args[0], Mapping):
+        for key, value in args[0].items():
+            tool_args.setdefault(str(key), value)
+    elif args:
+        for index, value in enumerate(args):
+            tool_args.setdefault(f"arg{index}", value)
+    return tool_args

agent_assembly/adapters/crewai/patch.py

@@ -9,6 +9,7 @@
 from threading import local
 from typing import Any, Literal, cast
 
+from agent_assembly.adapters._shared.positional_args import merge_positional_tool_args
 from agent_assembly.core.spawn import _SPAWN_CTX, SpawnContext, spawn_context_scope
 
 _TOOLS_PATCHED_FLAG = "_agent_assembly_crewai_tools_patched"
@@ -183,9 +184,7 @@ def _current_spawn_depth() -> int:
 
 def _format_blocked_message(reason: str | None) -> str:
     reason_text = reason or "No reason provided."
-    return (
-        f"[BLOCKED by governance policy] {reason_text}. " "Please choose a different approach to accomplish this task."
-    )
+    return f"[BLOCKED by governance policy] {reason_text}. Please choose a different approach to accomplish this task."
 
 
 def _format_approval_rejected_message(reason: str | None) -> str:
@@ -369,7 +368,9 @@ def _apply_basetool_run_patch(base_tool_cls: type[Any], callback_handler: Any) -
     @wraps(original_run)
     def patched_run(self: Any, *args: Any, **kwargs: Any) -> Any:
         tool_name = getattr(self, "name", self.__class__.__name__)
-        tool_args = dict(kwargs)
+        # Fold positional args in so content policy inspects their values; a
+        # positionally-invoked tool would otherwise present an empty mapping.
+        tool_args = merge_positional_tool_args(dict(kwargs), args)
         agent_id = _get_thread_local_agent_id()
         decision = _invoke_sync_tool_check(
             callback_handler,

agent_assembly/adapters/haystack/patch.py

@@ -25,6 +25,8 @@
 from functools import wraps
 from typing import Any, Literal, cast
 
+from agent_assembly.adapters._shared.positional_args import merge_positional_tool_args
+
 _TOOL_PATCHED_FLAG = "_agent_assembly_haystack_tool_patched"
 _ORIGINAL_TOOL_INVOKE = "_agent_assembly_original_haystack_tool_invoke"
 _DEFAULT_PENDING_APPROVAL_TIMEOUT_SECONDS = 300
@@ -236,9 +238,9 @@ def patched_invoke(self: Any, *args: Any, **kwargs: Any) -> Any:
         tool_name = str(getattr(self, "name", self.__class__.__name__))
         # Haystack's ToolInvoker calls ``invoke(**final_args)`` (keyword-only), but
         # a direct ``Tool.invoke(x)`` positional call must be governed and forwarded
-        # too rather than raising TypeError; positional args are opaque to the
-        # governance check, so only keyword args populate the inspected tool_args.
-        tool_args = dict(kwargs)
+        # too. Fold any positional args into tool_args so content policy inspects
+        # their values rather than seeing an empty mapping.
+        tool_args = merge_positional_tool_args(dict(kwargs), args)
         decision = _invoke_tool_check(
             callback_handler,
             tool_name=tool_name,

agent_assembly/adapters/llamaindex/patch.py

@@ -27,6 +27,7 @@
 from threading import local
 from typing import Any
 
+from agent_assembly.adapters._shared.positional_args import merge_positional_tool_args
 from agent_assembly.adapters.crewai.patch import (
     _get_pending_tool_approval_timeout_seconds as _resolve_pending_timeout_seconds,
 )
@@ -273,7 +274,9 @@ def _apply_tool_call_patch(tool_cls: type[Any], callback_handler: Any) -> bool:
     @wraps(original_call)
     def patched_call(self: Any, *args: Any, **kwargs: Any) -> Any:
         tool_name = _tool_name(self)
-        tool_args = dict(kwargs)
+        # Fold positional args in so content policy inspects their values; a
+        # positionally-invoked tool would otherwise present an empty mapping.
+        tool_args = merge_positional_tool_args(dict(kwargs), args)
         agent_id = _get_process_agent_id()
         status, reason, is_pending_flow = _resolve_governance_decision(
             callback_handler,
@@ -313,7 +316,9 @@ def _apply_tool_acall_patch(tool_cls: type[Any], callback_handler: Any) -> bool:
     @wraps(original_acall)
     async def patched_acall(self: Any, *args: Any, **kwargs: Any) -> Any:
         tool_name = _tool_name(self)
-        tool_args = dict(kwargs)
+        # Fold positional args in so content policy inspects their values; a
+        # positionally-invoked tool would otherwise present an empty mapping.
+        tool_args = merge_positional_tool_args(dict(kwargs), args)
         agent_id = _get_process_agent_id()
         status, reason, is_pending_flow = _resolve_governance_decision(
             callback_handler,

agent_assembly/adapters/microsoft_agent_framework/patch.py

@@ -268,6 +268,30 @@ async def _record_async_tool_result(
     return None
 
 
+def _recover_invoke_call_args(args: tuple[Any, ...], kwargs: dict[str, Any]) -> tuple[Any, Any, dict[str, Any]]:
+    """Recover ``(arguments, context, direct_kwargs)`` of a Microsoft Agent
+    Framework ``FunctionTool.invoke`` call.
+
+    ``invoke`` may be called positionally (arguments, then context); fall back to
+    the positional slots so content policy still inspects the argument values
+    instead of an empty mapping (AAASM-4848). ``direct_kwargs`` are the extras the
+    framework forwards straight to the wrapped function (i.e. not the reserved
+    invoke parameters).
+    """
+    arguments = kwargs.get("arguments")
+    if arguments is None and args:
+        arguments = args[0]
+    context = kwargs.get("context")
+    if context is None and len(args) >= 2:
+        context = args[1]
+    direct_kwargs = {
+        key: value
+        for key, value in kwargs.items()
+        if key not in ("arguments", "context", "tool_call_id", "skip_parsing")
+    }
+    return arguments, context, direct_kwargs
+
+
 def _apply_function_tool_invoke_patch(function_tool_cls: type[Any], callback_handler: Any) -> bool:
     if vars(function_tool_cls).get(_TOOLS_PATCHED_FLAG, False):
         return True
@@ -281,15 +305,7 @@ def _apply_function_tool_invoke_patch(function_tool_cls: type[Any], callback_han
     @wraps(original_invoke)
     async def patched_invoke(self: Any, *args: Any, **kwargs: Any) -> Any:
         tool_name = str(getattr(self, "name", self.__class__.__name__))
-        arguments = kwargs.get("arguments")
-        context = kwargs.get("context")
-        # Direct argument kwargs are any extras the framework forwards straight
-        # to the wrapped function (i.e. not the reserved invoke parameters).
-        direct_kwargs = {
-            key: value
-            for key, value in kwargs.items()
-            if key not in ("arguments", "context", "tool_call_id", "skip_parsing")
-        }
+        arguments, context, direct_kwargs = _recover_invoke_call_args(args, kwargs)
         tool_args = _serialize_tool_args(arguments, context, direct_kwargs)
         agent_id = _resolve_agent_id(context)
 

agent_assembly/adapters/pydantic_ai/patch.py

@@ -57,9 +57,15 @@ def apply(self) -> bool:
         """Apply patch wiring and return whether a tool hook was installed.
 
         Detects the tool-execution hook across Pydantic AI versions: the
-        ``Tool._run`` hook on <0.3.0 and the ``AbstractToolset.call_tool``
-        hook on >=0.3.0. When neither hook point exists, this is a no-op that
-        returns ``False`` instead of raising ``AttributeError``.
+        ``AbstractToolset.call_tool`` hook on >=0.3.0 and the ``Tool._run`` hook
+        on <0.3.0. When neither hook point exists, this is a no-op that returns
+        ``False`` instead of raising ``AttributeError``.
+
+        The modern ``call_tool`` hook is selected FIRST, with ``Tool._run`` used
+        only as the legacy fallback when no toolset hook point exists. Selecting
+        by precedence rather than by whichever attribute merely exists means a
+        future *vestigial* ``Tool._run`` — present but off the execution path —
+        can't shadow the live ``call_tool`` hook and leave tool calls ungoverned.
 
         On >=0.3.0 the abstract base patch is shadowed by concrete toolsets
         (e.g. ``FunctionToolset``) that override ``call_tool`` without calling
@@ -69,15 +75,16 @@ def apply(self) -> bool:
         set_process_agent_id(self.process_agent_id)
 
         tool_hooked = False
-        tool_cls = _load_pydantic_ai_tool_class()
-        if tool_cls is not None:
-            tool_hooked = _apply_tool_run_patch(tool_cls, self.callback_handler)
+        toolset_cls = _load_pydantic_ai_toolset_class()
+        if toolset_cls is not None:
+            tool_hooked = _apply_toolset_call_tool_patch(toolset_cls, self.callback_handler)
+            for concrete_cls in _load_pydantic_ai_concrete_toolset_classes(toolset_cls):
+                if _apply_toolset_call_tool_patch(concrete_cls, self.callback_handler):
+                    tool_hooked = True
         if not tool_hooked:
-            toolset_cls = _load_pydantic_ai_toolset_class()
-            if toolset_cls is not None:
-                tool_hooked = _apply_toolset_call_tool_patch(toolset_cls, self.callback_handler)
-                for concrete_cls in _load_pydantic_ai_concrete_toolset_classes(toolset_cls):
-                    _apply_toolset_call_tool_patch(concrete_cls, self.callback_handler)
+            tool_cls = _load_pydantic_ai_tool_class()
+            if tool_cls is not None:
+                tool_hooked = _apply_tool_run_patch(tool_cls, self.callback_handler)
 
         if not tool_hooked:
             set_process_agent_id(None)

test/unit/adapters/pydantic_ai/test_apply_hook_selection.py

@@ -0,0 +1,70 @@
+"""Regression: apply() selects the real tool-execution hook, not a vestigial one.
+
+Pydantic AI executes tool calls through ``AbstractToolset.call_tool`` (>=0.3.0);
+``Tool._run`` is only the legacy execution hook (<0.3.0). ``apply()`` used to try
+``Tool._run`` FIRST and only fall back to ``call_tool``. That is correct today
+because ``Tool._run`` is absent on modern versions — but a future *vestigial*
+``Tool._run`` (present yet off the execution path) would satisfy the first branch,
+leave ``call_tool`` unpatched, and silently make governance a dead method
+(AAASM-4848). These tests exercise apply()'s hook SELECTION directly (not the
+``_apply_*`` helpers in isolation) so that regression class is caught.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from agent_assembly.adapters.pydantic_ai import patch as pydantic_patch
+
+
+class _Interceptor:
+    def check_tool_start(self, **_kwargs: Any) -> dict[str, str]:
+        return {"status": "allow"}
+
+
+def test_call_tool_hook_wins_over_vestigial_tool_run(monkeypatch: Any) -> None:
+    class VestigialTool:
+        """A Tool whose ``_run`` exists but is NOT the execution path."""
+
+        async def _run(self, _ctx: Any, _args: Any, **_kwargs: Any) -> str:
+            return "vestigial-run"
+
+    class ExecutionToolset:
+        """The real >=0.3.0 execution hook lives on ``call_tool``."""
+
+        async def call_tool(self, _name: Any, _args: Any, _ctx: Any, _tool: Any, **_kwargs: Any) -> str:
+            return "executed"
+
+    monkeypatch.setattr(pydantic_patch, "_load_pydantic_ai_toolset_class", lambda: ExecutionToolset)
+    monkeypatch.setattr(pydantic_patch, "_load_pydantic_ai_concrete_toolset_classes", lambda _base: [])
+    monkeypatch.setattr(pydantic_patch, "_load_pydantic_ai_tool_class", lambda: VestigialTool)
+    monkeypatch.setattr(pydantic_patch, "_load_pydantic_ai_agent_class", lambda: None)
+
+    patcher = pydantic_patch.PydanticAIPatch(_Interceptor())
+    try:
+        assert patcher.apply() is True
+
+        # The execution hook (call_tool) is patched...
+        assert vars(ExecutionToolset).get(pydantic_patch._TOOLS_PATCHED_FLAG) is True
+        # ...and the vestigial Tool._run is left untouched.
+        assert pydantic_patch._TOOLS_PATCHED_FLAG not in vars(VestigialTool)
+        assert not hasattr(VestigialTool, pydantic_patch._ORIGINAL_TOOL_RUN)
+    finally:
+        pydantic_patch.set_process_agent_id(None)
+
+
+def test_tool_run_is_used_as_legacy_fallback_when_no_toolset_hook(monkeypatch: Any) -> None:
+    class LegacyTool:
+        async def _run(self, _ctx: Any, _args: Any, **_kwargs: Any) -> str:
+            return "legacy-run"
+
+    monkeypatch.setattr(pydantic_patch, "_load_pydantic_ai_toolset_class", lambda: None)
+    monkeypatch.setattr(pydantic_patch, "_load_pydantic_ai_tool_class", lambda: LegacyTool)
+    monkeypatch.setattr(pydantic_patch, "_load_pydantic_ai_agent_class", lambda: None)
+
+    patcher = pydantic_patch.PydanticAIPatch(_Interceptor())
+    try:
+        assert patcher.apply() is True
+        assert vars(LegacyTool).get(pydantic_patch._TOOLS_PATCHED_FLAG) is True
+    finally:
+        pydantic_patch.set_process_agent_id(None)