diff --git a/README.md b/README.md index 31dadcf..ea5d5ed 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ Python SDK for **AI Agent Assembly** — a governance-native runtime for AI agen ## Why use it -- **Framework adapters** for LangChain, LangGraph, CrewAI, OpenAI Agents, Pydantic AI, Google ADK, Haystack, Smolagents, Agno, LlamaIndex, and MCP servers — drop in, no SDK rewrites required. +- **Framework adapters** for LangChain, LangGraph, CrewAI, OpenAI Agents, Pydantic AI, Google ADK, Haystack, Smolagents, Agno, LlamaIndex, Microsoft Agent Framework, and MCP servers — drop in, no SDK rewrites required. - **Pre-execution policy enforcement** via the `FrameworkAdapter` ABC — block disallowed tool calls before they hit the LLM. - **Audit trail** — every tool call, prompt, and policy decision is emitted to the gateway with full agent lineage (parent / root / team). - **Native PyO3 fast path** (optional) — drop into a Rust runtime client when you need sub-millisecond policy checks. @@ -43,6 +43,7 @@ are expected to work but are not continuously tested. | OpenAI Agents | `agents` | `>=0.1.0` | `0.17.x` | | Smolagents | `smolagents` | `>=1.0.0,<2.0.0` | `1.26.x` | | Agno | `agno` | `>=2.0.0` | `2.6.x` | +| Microsoft Agent Framework | `agent_framework` | `>=1.0.0,<2.0` | `1.9.x` | Python framework compatibility is documented **authoritatively in the SDK docs** — [Framework compatibility](./docs/compatibility/frameworks.md) — because the Python diff --git a/agent_assembly/adapters/microsoft_agent_framework/__init__.py b/agent_assembly/adapters/microsoft_agent_framework/__init__.py new file mode 100644 index 0000000..1e8ef7a --- /dev/null +++ b/agent_assembly/adapters/microsoft_agent_framework/__init__.py @@ -0,0 +1,21 @@ +"""Microsoft Agent Framework adapter package. + +Governs tool/function execution for Microsoft's unified Agent Framework +(PyPI ``agent-framework``, importable module ``agent_framework``). The single +interception point is ``agent_framework.FunctionTool.invoke`` — the async method +through which every function tool (``@tool``-decorated callables and +``FunctionTool(...)`` instances) executes — so governance applies regardless of +whether the user wires any framework middleware. + +See :mod:`agent_assembly.adapters.microsoft_agent_framework.patch` for the hook +mechanics and ADR-0001 for the broader hook architecture. +""" + +from agent_assembly.adapters.microsoft_agent_framework.adapter import ( + MicrosoftAgentFrameworkAdapter, +) +from agent_assembly.adapters.microsoft_agent_framework.patch import ( + MicrosoftAgentFrameworkPatch, +) + +__all__ = ["MicrosoftAgentFrameworkAdapter", "MicrosoftAgentFrameworkPatch"] diff --git a/agent_assembly/adapters/microsoft_agent_framework/adapter.py b/agent_assembly/adapters/microsoft_agent_framework/adapter.py new file mode 100644 index 0000000..d37fdd0 --- /dev/null +++ b/agent_assembly/adapters/microsoft_agent_framework/adapter.py @@ -0,0 +1,62 @@ +"""Microsoft Agent Framework adapter.""" + +from __future__ import annotations + +import importlib.util + +from agent_assembly.adapters.base import FrameworkAdapter, GovernanceInterceptor +from agent_assembly.adapters.microsoft_agent_framework.patch import ( + MicrosoftAgentFrameworkPatch, +) + + +class MicrosoftAgentFrameworkAdapter(FrameworkAdapter): + """Adapter for Microsoft Agent Framework governance hook installation.""" + + def __init__(self, *, process_agent_id: str | None = None) -> None: + self._process_agent_id = process_agent_id + self._patch: MicrosoftAgentFrameworkPatch | None = None + + @property + def process_agent_id(self) -> str | None: + return self._process_agent_id + + @process_agent_id.setter + def process_agent_id(self, value: str | None) -> None: + self._process_agent_id = value + + def get_framework_name(self) -> str: + return "microsoft_agent_framework" + + def get_supported_versions(self) -> list[str]: + # Microsoft Agent Framework reached its first stable line at 1.x + # (``agent-framework`` / ``agent-framework-core`` 1.9.0 at time of + # writing). Pin to the 1.x major; the 2.x surface is unverified. + return [">=1.0.0,<2.0"] + + def is_available(self) -> bool: + """Detect the importable ``agent_framework`` package. + + WHY override (AAASM-3528 lesson): the framework name + (``microsoft_agent_framework``) deliberately differs from the importable + module (``agent_framework``). The base ``is_available`` imports + ``get_framework_name()``, which would never resolve here and make the + adapter silently report unavailable. Detect the real top-level module so + the hook genuinely activates when the framework is installed. + """ + try: + return importlib.util.find_spec("agent_framework") is not None + except (ImportError, ValueError): + return False + + def register_hooks(self, interceptor: GovernanceInterceptor) -> None: + self._patch = MicrosoftAgentFrameworkPatch( + callback_handler=interceptor, + process_agent_id=self._process_agent_id, + ) + self._patch.apply() + + def unregister_hooks(self) -> None: + if self._patch is not None: + self._patch.revert() + self._patch = None diff --git a/agent_assembly/adapters/microsoft_agent_framework/patch.py b/agent_assembly/adapters/microsoft_agent_framework/patch.py new file mode 100644 index 0000000..3066605 --- /dev/null +++ b/agent_assembly/adapters/microsoft_agent_framework/patch.py @@ -0,0 +1,355 @@ +"""Microsoft Agent Framework patch module. + +The interception point is ``agent_framework.FunctionTool.invoke`` — the single +async coroutine through which **every** function tool executes. Both the +``@agent_framework.tool`` decorator and direct ``FunctionTool(...)`` construction +produce ``FunctionTool`` instances, and agents/workflows dispatch tool calls +through ``FunctionTool.invoke``. Patching that one method therefore governs all +tool execution without requiring the user to register any framework middleware +(which would be opt-in and bypassable). + +Governance contract (mirrors the CrewAI / Pydantic AI adapters): + +- A ``deny`` verdict raises ``PolicyViolationError`` *before* the wrapped + function runs, so a denied tool never executes its side effects. +- A ``pending`` verdict waits for an approval decision and then allows or denies. +- An ``allow`` verdict runs the original ``invoke`` and records the result. +- Under ``enforce`` posture an unknown / malformed verdict fails closed (deny); + under ``observe`` / ``disabled`` it fails open (allow). +""" + +from __future__ import annotations + +import importlib as importlib +import inspect +from collections.abc import Mapping +from dataclasses import dataclass +from functools import wraps +from typing import TYPE_CHECKING, Any, Literal + +from agent_assembly.adapters.crewai.patch import ( + _get_pending_tool_approval_timeout_seconds as _resolve_pending_timeout_seconds, +) +from agent_assembly.adapters.crewai.patch import ( + _interceptor_enforces, +) +from agent_assembly.adapters.crewai.patch import ( + _normalize_decision as _normalize_governance_decision, +) +from agent_assembly.core.spawn import _SPAWN_CTX, SpawnContext, spawn_context_scope + +if TYPE_CHECKING: + from agent_assembly.exceptions import PolicyViolationError + +_ORIGINAL_FUNCTION_TOOL_INVOKE = "_agent_assembly_original_maf_function_tool_invoke" +_TOOLS_PATCHED_FLAG = "_agent_assembly_maf_function_tool_patched" +_PROCESS_AGENT_ID: str | None = None +_MAX_AUDIT_RESULT_CHARS = 2000 + + +@dataclass(slots=True) +class MicrosoftAgentFrameworkPatch: + """Applies Microsoft Agent Framework runtime monkey-patching hooks.""" + + callback_handler: Any + process_agent_id: str | None = None + + def apply(self) -> bool: + """Patch ``FunctionTool.invoke`` and return whether a hook was installed. + + Returns ``False`` (a no-op) when ``agent_framework`` is not importable or + does not expose a callable ``FunctionTool.invoke``, rather than raising. + """ + set_process_agent_id(self.process_agent_id) + + function_tool_cls = _load_function_tool_class() + if function_tool_cls is None: + set_process_agent_id(None) + return False + + return _apply_function_tool_invoke_patch(function_tool_cls, self.callback_handler) + + def revert(self) -> None: + """Revert the ``FunctionTool.invoke`` patch when available.""" + function_tool_cls = _load_function_tool_class() + if function_tool_cls is not None: + _revert_function_tool_invoke_patch(function_tool_cls) + set_process_agent_id(None) + return None + + +def _load_function_tool_class() -> type[Any] | None: + try: + module = importlib.import_module("agent_framework") + except ImportError: + return None + + function_tool_cls = getattr(module, "FunctionTool", None) + if isinstance(function_tool_cls, type): + return function_tool_cls + return None + + +def set_process_agent_id(agent_id: str | None) -> None: + global _PROCESS_AGENT_ID + _PROCESS_AGENT_ID = agent_id + + +def _get_process_agent_id() -> str | None: + if isinstance(_PROCESS_AGENT_ID, str) and _PROCESS_AGENT_ID: + return _PROCESS_AGENT_ID + return None + + +def _coerce_args_mapping(value: Any) -> dict[str, Any] | None: + """Return a plain dict for a Pydantic model / mapping argument, else ``None``.""" + model_dump = getattr(value, "model_dump", None) + if callable(model_dump): + dumped = model_dump() + if isinstance(dumped, Mapping): + return dict(dumped) + + if isinstance(value, Mapping): + return dict(value) + + return None + + +def _serialize_tool_args( + arguments: Any, + context: Any, + direct_kwargs: dict[str, Any], +) -> dict[str, Any]: + """Collapse the several ``FunctionTool.invoke`` argument channels into one dict. + + ``invoke`` accepts arguments via ``arguments=`` (a Pydantic model or mapping), + via ``context.arguments``, or as direct keyword arguments. The governance + check needs a single ``dict`` view of the call's inputs regardless of which + channel the caller used. + """ + mapped = _coerce_args_mapping(arguments) + if mapped is not None: + return mapped + + if context is not None: + mapped = _coerce_args_mapping(getattr(context, "arguments", None)) + if mapped is not None: + return mapped + + if direct_kwargs: + return dict(direct_kwargs) + + if arguments is not None: + return {"value": str(arguments)} + + return {} + + +def _resolve_agent_id(context: Any) -> str | None: + if context is not None: + for attr in ("agent_id", "assembly_agent_id"): + candidate = getattr(context, attr, None) + if isinstance(candidate, str) and candidate: + return candidate + return _get_process_agent_id() + + +def _current_spawn_depth() -> int: + current = _SPAWN_CTX.get() + return (current.depth + 1) if current is not None else 1 + + +def _normalize_decision( + decision: object, + *, + enforce: bool = False, +) -> tuple[Literal["allow", "deny", "pending"], str | None]: + return _normalize_governance_decision(decision, enforce=enforce) + + +def _get_pending_tool_approval_timeout_seconds(callback_handler: Any) -> int: + return _resolve_pending_timeout_seconds(callback_handler) + + +def _build_denied_error(tool_name: str, reason: str | None) -> PolicyViolationError: + from agent_assembly.exceptions import PolicyViolationError + + reason_text = reason or "No reason provided." + return PolicyViolationError(f"Tool '{tool_name}' blocked by governance policy: {reason_text}") + + +def _build_pending_rejected_error(tool_name: str, reason: str | None) -> PolicyViolationError: + from agent_assembly.exceptions import PolicyViolationError + + reason_text = reason or "No reason provided." + return PolicyViolationError(f"Tool '{tool_name}' rejected during approval: {reason_text}") + + +async def _invoke_async_tool_check( + callback_handler: Any, + *, + tool_name: str, + tool_args: dict[str, Any], + agent_id: str | None, +) -> object: + method = getattr(callback_handler, "check_tool_start", None) + if not callable(method): + return {"status": "allow"} + + result = method( + serialized={"name": tool_name}, + input_str=str(tool_args), + tool_name=tool_name, + args=tool_args, + agent_id=agent_id, + ) + if inspect.isawaitable(result): + return await result + return result + + +async def _wait_for_async_tool_approval( + callback_handler: Any, + *, + tool_name: str, + timeout_seconds: int, + tool_args: dict[str, Any], + agent_id: str | None, +) -> object: + method = getattr(callback_handler, "wait_for_tool_approval", None) + if not callable(method): + return {"status": "deny", "reason": "Approval handler is unavailable."} + + result = method( + serialized={"name": tool_name}, + input_str=str(tool_args), + tool_name=tool_name, + timeout_seconds=timeout_seconds, + args=tool_args, + agent_id=agent_id, + ) + if inspect.isawaitable(result): + return await result + return result + + +def _truncate_result_for_audit(result: object) -> str: + return str(result)[:_MAX_AUDIT_RESULT_CHARS] + + +async def _record_async_tool_result( + callback_handler: Any, + *, + tool_name: str, + result: object, + agent_id: str | None, +) -> None: + record_method = getattr(callback_handler, "record_result", None) + if callable(record_method): + recorded = record_method( + tool_name=tool_name, + result=_truncate_result_for_audit(result), + agent_id=agent_id, + ) + if inspect.isawaitable(recorded): + await recorded + return None + + tool_end_method = getattr(callback_handler, "on_tool_end", None) + if callable(tool_end_method): + recorded = tool_end_method( + output=_truncate_result_for_audit(result), + tool_name=tool_name, + agent_id=agent_id, + ) + if inspect.isawaitable(recorded): + await recorded + return None + + +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 + + original_invoke = getattr(function_tool_cls, "invoke", None) + if not callable(original_invoke): + return False + + enforce = _interceptor_enforces(callback_handler) + + @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") + } + tool_args = _serialize_tool_args(arguments, context, direct_kwargs) + agent_id = _resolve_agent_id(context) + + decision = await _invoke_async_tool_check( + callback_handler, + tool_name=tool_name, + tool_args=tool_args, + agent_id=agent_id, + ) + status, reason = _normalize_decision(decision, enforce=enforce) + is_pending_flow = False + if status == "pending": + is_pending_flow = True + timeout_seconds = _get_pending_tool_approval_timeout_seconds(callback_handler) + final_decision = await _wait_for_async_tool_approval( + callback_handler, + tool_name=tool_name, + timeout_seconds=timeout_seconds, + tool_args=tool_args, + agent_id=agent_id, + ) + status, reason = _normalize_decision(final_decision, enforce=enforce) + + if status == "deny": + if is_pending_flow: + raise _build_pending_rejected_error(tool_name, reason) + raise _build_denied_error(tool_name, reason) + + spawn_ctx = SpawnContext( + parent_agent_id=agent_id or "", + depth=_current_spawn_depth(), + spawned_by_tool=tool_name, + delegation_reason=f"tool:{tool_name}", + ) + with spawn_context_scope(spawn_ctx): + result = await original_invoke(self, *args, **kwargs) + + await _record_async_tool_result( + callback_handler, + tool_name=tool_name, + result=result, + agent_id=agent_id, + ) + return result + + setattr(function_tool_cls, _ORIGINAL_FUNCTION_TOOL_INVOKE, original_invoke) + function_tool_cls.invoke = patched_invoke + setattr(function_tool_cls, _TOOLS_PATCHED_FLAG, True) + return True + + +def _revert_function_tool_invoke_patch(function_tool_cls: type[Any]) -> None: + if not vars(function_tool_cls).get(_TOOLS_PATCHED_FLAG, False): + return None + + original_invoke = vars(function_tool_cls).get(_ORIGINAL_FUNCTION_TOOL_INVOKE, None) + if callable(original_invoke): + function_tool_cls.invoke = original_invoke + + if _ORIGINAL_FUNCTION_TOOL_INVOKE in vars(function_tool_cls): + delattr(function_tool_cls, _ORIGINAL_FUNCTION_TOOL_INVOKE) + if _TOOLS_PATCHED_FLAG in vars(function_tool_cls): + delattr(function_tool_cls, _TOOLS_PATCHED_FLAG) + return None diff --git a/agent_assembly/adapters/registry.py b/agent_assembly/adapters/registry.py index 25994c3..578c129 100644 --- a/agent_assembly/adapters/registry.py +++ b/agent_assembly/adapters/registry.py @@ -14,6 +14,9 @@ from agent_assembly.adapters.langgraph.adapter import LangGraphAdapter from agent_assembly.adapters.llamaindex.adapter import LlamaIndexAdapter from agent_assembly.adapters.mcp.adapter import MCPAdapter +from agent_assembly.adapters.microsoft_agent_framework.adapter import ( + MicrosoftAgentFrameworkAdapter, +) from agent_assembly.adapters.openai_agents.adapter import OpenAIAgentsAdapter from agent_assembly.adapters.pydantic_ai.adapter import PydanticAIAdapter from agent_assembly.adapters.smolagents.adapter import SmolagentsAdapter @@ -32,6 +35,7 @@ "smolagents": 6, "agno": 6, "llama_index.core": 6, + "microsoft_agent_framework": 6, "mcp": 99, } @@ -78,6 +82,7 @@ def __init__(self) -> None: SmolagentsAdapter(), AgnoAdapter(), LlamaIndexAdapter(), + MicrosoftAgentFrameworkAdapter(), MCPAdapter(), ] for adapter in builtin_adapters: diff --git a/docs/compatibility/frameworks.md b/docs/compatibility/frameworks.md index 7d7e0b7..578a9a5 100644 --- a/docs/compatibility/frameworks.md +++ b/docs/compatibility/frameworks.md @@ -54,6 +54,7 @@ to work but are not continuously tested. | OpenAI Agents | `agents` | `agent_assembly.adapters.openai_agents` | `>=0.1.0` | `0.17.x` | | Smolagents | `smolagents` | `agent_assembly.adapters.smolagents` | `>=1.0.0,<2.0.0` | `1.26.x` | | Agno | `agno` | `agent_assembly.adapters.agno` | `>=2.0.0` (patches `FunctionCall.execute` / `aexecute`) | `2.6.x` | +| Microsoft Agent Framework | `agent_framework` | `agent_assembly.adapters.microsoft_agent_framework` | `>=1.0.0,<2.0` | `1.9.x` — see [note below](#microsoft-agent-framework-version-range) | !!! note "Adapter present vs. example present" Every framework above has an adapter that is implemented and registered — @@ -109,6 +110,23 @@ which Agno surfaces to the model as a tool error. An `allow` runs the body and r result. The **tested line is `agno>=2.0.0`** (currently `2.6.x`), installed by the SDK's dev/test dependency group and exercised by the `importorskip`-guarded integration test. +### Microsoft Agent Framework version range + +Microsoft's unified **Agent Framework** ships on PyPI as `agent-framework` but imports +as the top-level module **`agent_framework`** — so the adapter's `is_available()` probes +`agent_framework`, not its framework name. The single interception point is +`agent_framework.FunctionTool.invoke`, the async coroutine through which **every** +function tool executes (both `@agent_framework.tool`-decorated callables and direct +`FunctionTool(...)` instances). Patching that one method governs all tool execution +without requiring the user to register any framework middleware. + +The **tested line is `agent-framework>=1.9.0`** — the first stable 1.x line. The adapter +pins `>=1.0.0,<2.0`; the 2.x surface is unverified. The package is **pre-release-heavy**: +several of its sub-distributions (`agent-framework-azure-ai-search`, etc.) are published +as pre-releases, so an installer that does not allow pre-releases (notably `uv`) needs an +explicit pre-release opt-in (e.g. `uv pip install --prerelease=allow agent-framework`). +`pip install agent-framework` resolves them without a flag. + ## Declaring frameworks in your own project The frameworks above are **not** runtime dependencies of `agent-assembly` — installing diff --git a/docs/examples/framework-support.md b/docs/examples/framework-support.md index 11e0831..faa1bf1 100644 --- a/docs/examples/framework-support.md +++ b/docs/examples/framework-support.md @@ -47,6 +47,7 @@ repository. | OpenAI Agents | `agent_assembly.adapters.openai_agents` | ✅ Validated — see [OpenAI Agents SDK](openai-agents-sdk.md). | | Pydantic AI | `agent_assembly.adapters.pydantic_ai` | ✅ Validated — see [Pydantic AI](pydantic-ai.md). | | Google ADK | `agent_assembly.adapters.google_adk` | ✅ Validated — see [Google ADK](google-adk.md). | +| Microsoft Agent Framework | `agent_assembly.adapters.microsoft_agent_framework` | ✅ Validated — governs `agent_framework.FunctionTool.invoke`; see the [`microsoft-agent-framework-tool-policy`](https://github.com/ai-agent-assembly/agent-assembly-examples/tree/master/python/microsoft-agent-framework-tool-policy) example. | | LlamaIndex | _no native adapter_ | ✅ Validated (manual wrapper) — see [LlamaIndex — manual tool policy](llamaindex-tool-policy.md); governs `FunctionTool` calls via `GovernedToolRunner`. | | MCP servers | `agent_assembly.adapters.mcp` | ⏳ Planned — adapter ships; a curated example is not yet vendored. | diff --git a/test/unit/adapters/microsoft_agent_framework/test_microsoft_agent_framework_adapter.py b/test/unit/adapters/microsoft_agent_framework/test_microsoft_agent_framework_adapter.py new file mode 100644 index 0000000..116f431 --- /dev/null +++ b/test/unit/adapters/microsoft_agent_framework/test_microsoft_agent_framework_adapter.py @@ -0,0 +1,81 @@ +"""Unit tests for the Microsoft Agent Framework adapter contract.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from agent_assembly.adapters.microsoft_agent_framework.adapter import ( + MicrosoftAgentFrameworkAdapter, +) + + +def test_framework_name_and_versions() -> None: + adapter = MicrosoftAgentFrameworkAdapter() + assert adapter.get_framework_name() == "microsoft_agent_framework" + versions = adapter.get_supported_versions() + assert versions == [">=1.0.0,<2.0"] + # Contract validation must pass for a registerable adapter. + adapter.validate_registration() + + +def test_is_available_detects_agent_framework_not_framework_name(monkeypatch: pytest.MonkeyPatch) -> None: + """``is_available`` must probe ``agent_framework`` — not the framework name. + + The framework name (``microsoft_agent_framework``) is intentionally not the + importable module (``agent_framework``). Probing the name would always miss + and silently disable governance (AAASM-3528). + """ + adapter = MicrosoftAgentFrameworkAdapter() + + probed: list[str] = [] + + def fake_find_spec(name: str) -> Any: + probed.append(name) + return object() if name == "agent_framework" else None + + monkeypatch.setattr( + "agent_assembly.adapters.microsoft_agent_framework.adapter.importlib.util.find_spec", + fake_find_spec, + ) + assert adapter.is_available() is True + assert probed == ["agent_framework"] + + +def test_is_available_false_when_absent(monkeypatch: pytest.MonkeyPatch) -> None: + adapter = MicrosoftAgentFrameworkAdapter() + + monkeypatch.setattr( + "agent_assembly.adapters.microsoft_agent_framework.adapter.importlib.util.find_spec", + lambda _name: None, + ) + assert adapter.is_available() is False + + +def test_is_available_handles_module_not_found(monkeypatch: pytest.MonkeyPatch) -> None: + adapter = MicrosoftAgentFrameworkAdapter() + + def raise_mnfe(_name: str) -> Any: + raise ModuleNotFoundError("agent_framework") + + monkeypatch.setattr( + "agent_assembly.adapters.microsoft_agent_framework.adapter.importlib.util.find_spec", + raise_mnfe, + ) + assert adapter.is_available() is False + + +def test_process_agent_id_round_trips() -> None: + adapter = MicrosoftAgentFrameworkAdapter(process_agent_id="agent-7") + assert adapter.process_agent_id == "agent-7" + adapter.set_process_agent_id("agent-9") + assert adapter.process_agent_id == "agent-9" + + +def test_registered_as_builtin_adapter() -> None: + from agent_assembly.adapters.registry import AdapterRegistry + + registry = AdapterRegistry() + assert "microsoft_agent_framework" in registry._registered + assert registry._registered["microsoft_agent_framework"].get_framework_name() == "microsoft_agent_framework" diff --git a/test/unit/adapters/microsoft_agent_framework/test_microsoft_agent_framework_patch.py b/test/unit/adapters/microsoft_agent_framework/test_microsoft_agent_framework_patch.py new file mode 100644 index 0000000..0cd073c --- /dev/null +++ b/test/unit/adapters/microsoft_agent_framework/test_microsoft_agent_framework_patch.py @@ -0,0 +1,288 @@ +"""Unit tests for the Microsoft Agent Framework adapter patch. + +These tests prove **real governance**, not a no-op (the AAASM-3528 lesson): + +- A fake ``FunctionTool`` exercises the patch wiring without requiring the + framework to be installed (so the suite runs in plain CI). +- A *negative control* records that the wrapped function actually runs when the + verdict is ``allow`` and is *prevented* when the verdict is ``deny`` — a no-op + patch would fail both halves. +- A guarded section drives the **real** ``agent_framework.FunctionTool`` when the + package is importable, proving the chosen hook point genuinely intercepts. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest + +from agent_assembly.adapters.microsoft_agent_framework import patch as maf_patch +from agent_assembly.exceptions import PolicyViolationError + + +class _AllowInterceptor: + _enforce = True + + def __init__(self) -> None: + self.checked: list[dict[str, Any]] = [] + self.recorded: list[dict[str, Any]] = [] + + def check_tool_start(self, **kwargs: Any) -> dict[str, str]: + self.checked.append(kwargs) + return {"status": "allow"} + + def record_result(self, **kwargs: Any) -> None: + self.recorded.append(kwargs) + + +class _DenyInterceptor: + _enforce = True + + def check_tool_start(self, **kwargs: Any) -> dict[str, str]: + del kwargs + return {"status": "deny", "reason": "denied by policy"} + + +class _PendingThenDenyInterceptor: + _enforce = True + + def check_tool_start(self, **kwargs: Any) -> dict[str, str]: + del kwargs + return {"status": "pending"} + + def wait_for_tool_approval(self, **kwargs: Any) -> dict[str, str]: + del kwargs + return {"status": "deny", "reason": "rejected at approval"} + + +def _install_fake_agent_framework(monkeypatch: pytest.MonkeyPatch, side_effects: list[Any]) -> type[Any]: + """Install a fake ``agent_framework`` exposing a ``FunctionTool`` with ``invoke``.""" + + class FakeFunctionTool: + name = "fake_tool" + + async def invoke(self, *, arguments: Any = None, **kwargs: Any) -> str: + del kwargs + side_effects.append(("ran", arguments)) + return "real-result" + + fake_module = SimpleNamespace(FunctionTool=FakeFunctionTool) + + def fake_import_module(module_name: str) -> object: + if module_name == "agent_framework": + return fake_module + raise ImportError(module_name) + + monkeypatch.setattr(maf_patch.importlib, "import_module", fake_import_module) + return FakeFunctionTool + + +def test_apply_is_idempotent_and_revert_restores(monkeypatch: pytest.MonkeyPatch) -> None: + side_effects: list[Any] = [] + FakeTool = _install_fake_agent_framework(monkeypatch, side_effects) + original_invoke = FakeTool.invoke + + patcher = maf_patch.MicrosoftAgentFrameworkPatch(_AllowInterceptor()) + assert patcher.apply() is True + patched_ref = FakeTool.invoke + assert FakeTool.invoke is not original_invoke + assert getattr(FakeTool, maf_patch._TOOLS_PATCHED_FLAG, False) is True + + # Second apply is a no-op: the patched method is not re-wrapped. + assert patcher.apply() is True + assert FakeTool.invoke is patched_ref + + patcher.revert() + assert FakeTool.invoke is original_invoke + assert getattr(FakeTool, maf_patch._TOOLS_PATCHED_FLAG, False) is False + assert maf_patch._get_process_agent_id() is None + + +@pytest.mark.asyncio +async def test_allow_runs_tool_and_records_result(monkeypatch: pytest.MonkeyPatch) -> None: + side_effects: list[Any] = [] + FakeTool = _install_fake_agent_framework(monkeypatch, side_effects) + interceptor = _AllowInterceptor() + + patcher = maf_patch.MicrosoftAgentFrameworkPatch(interceptor) + assert patcher.apply() is True + + result = await FakeTool().invoke(arguments={"city": "Seattle"}) + + # Real governance: the wrapped function actually executed (negative control). + assert side_effects == [("ran", {"city": "Seattle"})] + assert result == "real-result" + # The check observed the real tool name + args, and the result was recorded. + assert interceptor.checked[0]["tool_name"] == "fake_tool" + assert interceptor.checked[0]["args"] == {"city": "Seattle"} + assert interceptor.recorded[0]["tool_name"] == "fake_tool" + + patcher.revert() + + +@pytest.mark.asyncio +async def test_deny_blocks_tool_and_prevents_side_effect(monkeypatch: pytest.MonkeyPatch) -> None: + side_effects: list[Any] = [] + FakeTool = _install_fake_agent_framework(monkeypatch, side_effects) + + patcher = maf_patch.MicrosoftAgentFrameworkPatch(_DenyInterceptor()) + assert patcher.apply() is True + + with pytest.raises(PolicyViolationError, match="blocked by governance policy"): + await FakeTool().invoke(arguments={"city": "Seattle"}) + + # The wrapped function must NOT have run — a no-op patch would let it through. + assert side_effects == [] + + patcher.revert() + + +@pytest.mark.asyncio +async def test_pending_then_deny_rejects_at_approval(monkeypatch: pytest.MonkeyPatch) -> None: + side_effects: list[Any] = [] + FakeTool = _install_fake_agent_framework(monkeypatch, side_effects) + + patcher = maf_patch.MicrosoftAgentFrameworkPatch(_PendingThenDenyInterceptor()) + assert patcher.apply() is True + + with pytest.raises(PolicyViolationError, match="rejected during approval"): + await FakeTool().invoke(arguments={"x": 1}) + + assert side_effects == [] + + patcher.revert() + + +@pytest.mark.asyncio +async def test_unknown_verdict_fails_closed_under_enforce(monkeypatch: pytest.MonkeyPatch) -> None: + side_effects: list[Any] = [] + FakeTool = _install_fake_agent_framework(monkeypatch, side_effects) + + class _MalformedInterceptor: + _enforce = True + + def check_tool_start(self, **kwargs: Any) -> dict[str, str]: + del kwargs + return {"status": "garbage"} + + patcher = maf_patch.MicrosoftAgentFrameworkPatch(_MalformedInterceptor()) + assert patcher.apply() is True + + with pytest.raises(PolicyViolationError): + await FakeTool().invoke(arguments={"x": 1}) + assert side_effects == [] + + patcher.revert() + + +@pytest.mark.asyncio +async def test_unknown_verdict_fails_open_without_enforce(monkeypatch: pytest.MonkeyPatch) -> None: + side_effects: list[Any] = [] + FakeTool = _install_fake_agent_framework(monkeypatch, side_effects) + + class _ObserveInterceptor: + # No ``_enforce`` attribute -> fail-open (observe / disabled posture). + def check_tool_start(self, **kwargs: Any) -> dict[str, str]: + del kwargs + return {"status": "garbage"} + + patcher = maf_patch.MicrosoftAgentFrameworkPatch(_ObserveInterceptor()) + assert patcher.apply() is True + + result = await FakeTool().invoke(arguments={"x": 1}) + assert result == "real-result" + assert side_effects == [("ran", {"x": 1})] + + patcher.revert() + + +def test_apply_is_noop_when_framework_absent(monkeypatch: pytest.MonkeyPatch) -> None: + def raise_import_error(module_name: str) -> object: + raise ImportError(module_name) + + monkeypatch.setattr(maf_patch.importlib, "import_module", raise_import_error) + assert maf_patch._load_function_tool_class() is None + assert maf_patch.MicrosoftAgentFrameworkPatch(_AllowInterceptor()).apply() is False + + +def test_load_returns_none_for_non_type_attribute(monkeypatch: pytest.MonkeyPatch) -> None: + fake_module = SimpleNamespace(FunctionTool=object()) + + def fake_import_module(module_name: str) -> object: + if module_name == "agent_framework": + return fake_module + raise ImportError(module_name) + + monkeypatch.setattr(maf_patch.importlib, "import_module", fake_import_module) + assert maf_patch._load_function_tool_class() is None + + +def test_serialize_tool_args_channels() -> None: + class _Model: + def model_dump(self) -> dict[str, Any]: + return {"a": 1} + + # arguments= via a pydantic-like model + assert maf_patch._serialize_tool_args(_Model(), None, {}) == {"a": 1} + # arguments= via a plain mapping + assert maf_patch._serialize_tool_args({"b": 2}, None, {}) == {"b": 2} + # via context.arguments + ctx = SimpleNamespace(arguments={"c": 3}) + assert maf_patch._serialize_tool_args(None, ctx, {}) == {"c": 3} + # via direct kwargs + assert maf_patch._serialize_tool_args(None, None, {"d": 4}) == {"d": 4} + # empty + assert maf_patch._serialize_tool_args(None, None, {}) == {} + + +# --- Real-framework governance proof (skips when agent-framework is absent) --- + + +@pytest.mark.asyncio +async def test_real_agent_framework_function_tool_is_governed() -> None: + """Drive the REAL ``agent_framework.FunctionTool`` to prove a genuine hook. + + This is the decisive negative control against a fake adapter: with the + actual package installed, a denied verdict must both raise and stop the + wrapped function's side effect, while an allowed verdict must run it. + """ + af = pytest.importorskip("agent_framework", reason="agent-framework not installed") + from agent_assembly.adapters.microsoft_agent_framework import ( + MicrosoftAgentFrameworkAdapter, + ) + + side: list[tuple[str, str]] = [] + + @af.tool # type: ignore[misc] # framework decorator is untyped + def write_note(path: str, content: str) -> str: + """Write a note (records a side effect).""" + side.append((path, content)) + return f"wrote {path}" + + adapter = MicrosoftAgentFrameworkAdapter() + + # Negative control: ungoverned call runs the side effect. + side.clear() + await write_note.invoke(arguments={"path": "/a", "content": "x"}) + assert side == [("/a", "x")] + + try: + adapter.register(_DenyInterceptor()) + side.clear() + with pytest.raises(PolicyViolationError): + await write_note.invoke(arguments={"path": "/a", "content": "x"}) + assert side == [] # deny prevented the side effect + finally: + adapter.unregister_hooks() + + interceptor = _AllowInterceptor() + try: + adapter.register(interceptor) + side.clear() + await write_note.invoke(arguments={"path": "/b", "content": "y"}) + assert side == [("/b", "y")] # allow ran the tool + assert interceptor.recorded # and recorded the result + finally: + adapter.unregister_hooks()