diff --git a/test/unit/adapters/failopen_conformance.py b/test/unit/adapters/failopen_conformance.py new file mode 100644 index 00000000..c03a27ab --- /dev/null +++ b/test/unit/adapters/failopen_conformance.py @@ -0,0 +1,427 @@ +"""Adapter-driver registry + fault-injection harness for the fail-open conformance matrix. + +This is the systemic answer to the per-round fail-open whack-a-mole (openai_agents +AAASM-4734 → AAASM-4782, langchain AAASM-4790): instead of adding one bespoke +regression test each time a single adapter is caught failing open, every governing +adapter is driven through the *same* fault scenarios under the *same* enforcement +postures, and a completeness check makes a new (or regressed) adapter that ships +without a driver fail the suite. + +The harness reuses the technique the per-adapter tests already proved: build a fake +framework tool whose real body flips a ``ran`` flag, invoke the adapter's *real* +governance wrapper (``_apply_*`` / callback helper) against a fake interceptor whose +behaviour is set per scenario, and report whether the underlying tool actually ran. + +What is asserted (the security invariant): + +* Under an **enforce** posture — explicit ``enforce`` and the ``None``/omitted default, + which :func:`_local_posture_is_enforce` resolves to enforce — every fault (S1–S5), + plus the deny controls (S6–S7), MUST block the tool: it does not run. Only the + explicit allow control (S8) runs. +* Under a **fail-open** posture (``observe`` / ``disabled``) the faults (S1–S5) proceed + by design (the tool runs), preserving the dry-run / hermetic behaviour, while an + authoritative deny (S6–S7) still blocks in every posture. + +The scenarios inject the *raw* fault at the interceptor boundary (no verdict, an +error sentinel, garbage, a missing method, a pending-approval fault, a deny whose +audit sink raises) so the ADAPTER's own normalize + fail-closed logic is what is +under test — not a pre-decided verdict handed to it. +""" + +from __future__ import annotations + +import contextlib +import importlib +from collections.abc import Awaitable, Callable +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from uuid import uuid4 + +from agent_assembly.core.runtime_interceptor import _local_posture_is_enforce +from agent_assembly.exceptions import GatewayError + +# --- Scenarios ------------------------------------------------------------- + +# The interceptor faults the adapter's pre-execution check can hit. Each value is +# the raw fault injected at the ``check_tool_start`` boundary (or its absence); the +# adapter's own ``_normalize_decision`` / ``_missing_interceptor_decision`` is what +# converts it into allow/deny under the active posture. +GATEWAY_UNREACHABLE = "s1_gateway_unreachable" # check returns None (no verdict) +GOVERNANCE_ERROR = "s2_governance_error" # check returns an error-sentinel status +MALFORMED_DECISION = "s3_malformed_decision" # check returns an unrecognized object +MISSING_CHECK_METHOD = "s4_missing_check_method" # interceptor exposes no check_tool_start +PENDING_FAULT = "s5_pending_fault" # pending verdict, then approval faults +DENY_AUDIT_RAISES = "s6_deny_audit_raises" # authoritative deny whose audit sink raises +EXPLICIT_DENY = "s7_explicit_deny" # authoritative deny (control) +EXPLICIT_ALLOW = "s8_explicit_allow" # authoritative allow (control) + +SCENARIOS: tuple[str, ...] = ( + GATEWAY_UNREACHABLE, + GOVERNANCE_ERROR, + MALFORMED_DECISION, + MISSING_CHECK_METHOD, + PENDING_FAULT, + DENY_AUDIT_RAISES, + EXPLICIT_DENY, + EXPLICIT_ALLOW, +) + +# Authoritative-deny scenarios block regardless of posture; the fault scenarios only +# block under an enforce posture (they fail open under observe / disabled). +_AUTHORITATIVE_DENY = frozenset({DENY_AUDIT_RAISES, EXPLICIT_DENY}) + +# --- Enforcement modes ----------------------------------------------------- + +# ``None`` is the advertised default and MUST resolve to the enforce posture +# (AAASM-4130) — the resolution is pinned here through the real production helper so +# an omitted ``enforcement_mode`` can never silently drift to fail-open. +MODES: tuple[tuple[str, str | None], ...] = ( + ("enforce", "enforce"), + ("omitted_none", None), + ("observe", "observe"), + ("disabled", "disabled"), +) + + +def expected_run(scenario: str, *, is_enforce: bool) -> bool: + """The tool-ran outcome the governance contract requires for this cell. + + Args: + scenario: One of :data:`SCENARIOS`. + is_enforce: Whether the active posture is fail-closed (enforce / the ``None`` + default), as resolved by :func:`_local_posture_is_enforce`. + + Returns: + ``True`` when the underlying tool is expected to execute, ``False`` when + governance must block it. + """ + if scenario == EXPLICIT_ALLOW: + return True + if scenario in _AUTHORITATIVE_DENY: + return False + # Injected faults (S1–S5): fail closed under enforce, fail open otherwise. + return not is_enforce + + +# --- Fake interceptor ------------------------------------------------------ + + +class _FaultInterceptor: + """Fake governance interceptor whose ``check_tool_start`` injects a scenario fault. + + ``_enforce`` mirrors the real interceptor flag every adapter reads + (``_interceptor_enforces`` / the LangChain handler's ``_enforce``) so the + adapter's fail-closed-under-enforce path is exercised authentically. + """ + + def __init__(self, scenario: str, *, enforce: bool) -> None: + self._scenario = scenario + self._enforce = enforce + + def check_tool_start(self, **_kwargs: Any) -> object: + scenario = self._scenario + if scenario == GATEWAY_UNREACHABLE: + return None + if scenario == GOVERNANCE_ERROR: + return {"status": "error", "reason": "governance transport failure"} + if scenario == MALFORMED_DECISION: + return object() + if scenario == PENDING_FAULT: + return {"status": "pending", "reason": "awaiting approval"} + if scenario in _AUTHORITATIVE_DENY: + return {"status": "deny", "reason": "denied by policy"} + if scenario == EXPLICIT_ALLOW: + return {"status": "allow"} + raise AssertionError(f"no check_tool_start behaviour for scenario {scenario!r}") + + def wait_for_tool_approval(self, **_kwargs: Any) -> object: + # A faulting approval round-trip returns no authoritative verdict; the + # adapter must fail closed under enforce and open under observe. + return None + + def record_result(self, **_kwargs: Any) -> None: + self._maybe_fail_audit() + + def on_tool_end(self, **_kwargs: Any) -> None: + self._maybe_fail_audit() + + def _maybe_fail_audit(self) -> None: + # AAASM-4782: a deny whose audit sink raises must not let the error re-enter + # the run path and downgrade the decided deny into an allow. + if self._scenario == DENY_AUDIT_RAISES: + raise GatewayError("deny-audit sink unavailable") + + +class _MissingCheckInterceptor: + """Fake interceptor that exposes no ``check_tool_start`` at all (AAASM-4790). + + Every adapter must route the missing-method case through + ``_missing_interceptor_decision`` — deny under enforce, allow otherwise — rather + than silently skipping the pre-execution gate. + """ + + def __init__(self, *, enforce: bool) -> None: + self._enforce = enforce + + def record_result(self, **_kwargs: Any) -> None: + return None + + def on_tool_end(self, **_kwargs: Any) -> None: + return None + + +def make_fake_interceptor(scenario: str, *, enforce: bool) -> object: + """Build the fake interceptor for a scenario under the given enforce posture.""" + if scenario == MISSING_CHECK_METHOD: + return _MissingCheckInterceptor(enforce=enforce) + return _FaultInterceptor(scenario, enforce=enforce) + + +# --- Adapter drivers ------------------------------------------------------- +# +# Each driver builds a fake framework tool whose body flips ``ran[0]`` and invokes the +# adapter's REAL governance wrapper against ``interceptor``. A driver never inspects +# the deny shape (raise vs sentinel return): the only signal is whether the underlying +# body ran, so the harness handles blocked-by-raise uniformly by catching AssemblyError. + +Driver = Callable[[object, list[bool]], Awaitable[None]] + + +async def _drive_crewai(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.crewai import patch as crewai_patch + + class FakeBaseTool: + name = "conformance_tool" + + def run(self, *_args: Any, **_kwargs: Any) -> dict[str, object]: + ran[0] = True + return {"ok": True} + + crewai_patch._apply_basetool_run_patch(FakeBaseTool, interceptor) + FakeBaseTool().run(param="x") + + +async def _drive_agno(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.agno import patch as agno_patch + + class FakeFunctionCall: + def __init__(self) -> None: + self.function = SimpleNamespace(name="conformance_tool") + self.arguments = {"amount": 1} + + def execute(self, *_args: Any, **_kwargs: Any) -> str: + ran[0] = True + return "ok" + + agno_patch._apply_execute_patch(FakeFunctionCall, interceptor) + FakeFunctionCall().execute() + + +async def _drive_haystack(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.haystack import patch as haystack_patch + + class FakeTool: + name = "conformance_tool" + + def invoke(self, **_kwargs: Any) -> dict[str, object]: + ran[0] = True + return {"ok": True} + + haystack_patch._apply_tool_invoke_patch(FakeTool, interceptor) + FakeTool().invoke(param="x") + + +async def _drive_langchain(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.langchain.callback_handler import ( + AssemblyCallbackHandler, + ) + + # LangChain governs via a pre-execution callback: on_tool_start raises to abort the + # tool. The tool "runs" iff the pre-hook returns without blocking. + handler = AssemblyCallbackHandler(interceptor) + handler.on_tool_start({"name": "conformance_tool"}, "input", run_id=uuid4()) + ran[0] = True + + +async def _drive_llamaindex(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.llamaindex import patch as llamaindex_patch + + class _Meta: + def __init__(self, name: str) -> None: + self._name = name + + def get_name(self) -> str: + return self._name + + class FakeFunctionTool: + def __init__(self) -> None: + self.metadata = _Meta("conformance_tool") + + def call(self, *_args: Any, **_kwargs: Any) -> dict[str, object]: + ran[0] = True + return {"ok": True} + + async def acall(self, *_args: Any, **_kwargs: Any) -> dict[str, object]: + ran[0] = True + return {"ok": True} + + llamaindex_patch._apply_tool_call_patch(FakeFunctionTool, interceptor) + FakeFunctionTool().call(param="x") + + +async def _drive_smolagents(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.smolagents import patch as smolagents_patch + + class FakeTool: + name = "conformance_tool" + + def __call__(self, *_args: Any, **_kwargs: Any) -> dict[str, object]: + ran[0] = True + return {"ok": True} + + smolagents_patch._apply_tool_call_patch(FakeTool, interceptor) + FakeTool()(text="x") + + +async def _drive_microsoft_agent_framework(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.microsoft_agent_framework import patch as maf_patch + + class FakeFunctionTool: + name = "conformance_tool" + + async def invoke(self, *_args: Any, **_kwargs: Any) -> str: + ran[0] = True + return "ok" + + maf_patch._apply_function_tool_invoke_patch(FakeFunctionTool, interceptor) + await FakeFunctionTool().invoke(arguments={"amount": 1}) + + +async def _drive_openai_agents(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.openai_agents import patch as openai_patch + + class FakeFunctionTool: + def __init__(self) -> None: + self.name = "conformance_tool" + + async def _invoke(_ctx: Any, _tool_input: Any) -> dict[str, object]: + ran[0] = True + return {"ok": True} + + self.on_invoke_tool = _invoke + + openai_patch._apply_function_tool_patch(FakeFunctionTool, interceptor) + tool = FakeFunctionTool() + await tool.on_invoke_tool(SimpleNamespace(agent_id="conformance-agent"), "{}") + + +async def _drive_pydantic_ai(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.pydantic_ai import patch as pydantic_ai_patch + + class FakeTool: + name = "conformance_tool" + + async def _run(self, _ctx: Any, _args: Any, **_kwargs: Any) -> dict[str, object]: + ran[0] = True + return {"ok": True} + + pydantic_ai_patch._apply_tool_run_patch(FakeTool, interceptor) + ctx = SimpleNamespace( + deps=SimpleNamespace(assembly_agent_id="conformance-agent"), + run_id="run-1", + ) + await FakeTool()._run(ctx, {"topic": "x"}) + + +async def _drive_google_adk(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.google_adk import patch as google_adk_patch + + class FakeBaseTool: + name = "conformance_tool" + + async def run_async(self, *, args: Any, tool_context: Any, **_kwargs: Any) -> dict[str, object]: + del args, tool_context + ran[0] = True + return {"ok": True} + + google_adk_patch._apply_tool_run_async_patch(FakeBaseTool, interceptor) + tool_context = SimpleNamespace( + invocation_context=SimpleNamespace(assembly_agent_id="conformance-agent", invocation_id="run-1"), + ) + await FakeBaseTool().run_async(args={"step": 1}, tool_context=tool_context) + + +async def _drive_mcp(interceptor: object, ran: list[bool]) -> None: + from agent_assembly.adapters.mcp import patch as mcp_patch + + class FakeClientSession: + _server_name = "conformance-server" + + async def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> dict[str, object]: + del name, arguments + ran[0] = True + return {"ok": True} + + mcp_patch._apply_client_session_patch(FakeClientSession, interceptor) + session = FakeClientSession() + await session.call_tool("conformance_tool", {"q": "x"}) + + +# adapter package name → driver. The key MUST equal the directory under +# ``agent_assembly/adapters/`` (the completeness test enforces this both ways). +DRIVERS: dict[str, Driver] = { + "agno": _drive_agno, + "crewai": _drive_crewai, + "google_adk": _drive_google_adk, + "haystack": _drive_haystack, + "langchain": _drive_langchain, + "llamaindex": _drive_llamaindex, + "mcp": _drive_mcp, + "microsoft_agent_framework": _drive_microsoft_agent_framework, + "openai_agents": _drive_openai_agents, + "pydantic_ai": _drive_pydantic_ai, + "smolagents": _drive_smolagents, +} + +# Adapter packages that intentionally have no pre-execution tool-gate driver, each +# with the reason. If one of these grows a tool gate, delete its exemption and add a +# driver; if a brand-new adapter package appears, the completeness test fails until it +# gets a driver or a justified exemption here — that is the anti-whack-a-mole guard. +EXEMPT_ADAPTERS: dict[str, str] = { + "langgraph": "lineage-only: patches node/graph lifecycle for edge emission, no pre-execution tool deny gate", + "_shared": "shared async governance helper reused by other adapters, not a framework adapter itself", +} + + +def discover_adapter_packages() -> set[str]: + """Return the adapter package directories present under ``agent_assembly/adapters/``. + + A package is a subdirectory with an ``__init__.py`` (this includes ``_shared``); + dunder dirs such as ``__pycache__`` are excluded. + """ + adapters_pkg = importlib.import_module("agent_assembly.adapters") + adapters_dir = Path(adapters_pkg.__file__).parent # type: ignore[arg-type] + return { + entry.name + for entry in adapters_dir.iterdir() + if entry.is_dir() and not entry.name.startswith("__") and (entry / "__init__.py").exists() + } + + +async def run_scenario(adapter_name: str, scenario: str, mode: str | None) -> bool: + """Drive one adapter through one scenario under one mode; return whether the tool ran. + + A blocked-by-raise deny (PolicyViolationError / ToolExecutionBlockedError / + MCPToolBlockedError — all ``AssemblyError`` subclasses) is swallowed here because + it is the intended block outcome; the ``ran`` flag is the sole signal. Any other + exception propagates and fails the test. + """ + from agent_assembly.exceptions import AssemblyError + + enforce = _local_posture_is_enforce(mode) + interceptor = make_fake_interceptor(scenario, enforce=enforce) + ran = [False] + with contextlib.suppress(AssemblyError): + await DRIVERS[adapter_name](interceptor, ran) + return ran[0] diff --git a/test/unit/adapters/test_failopen_conformance.py b/test/unit/adapters/test_failopen_conformance.py new file mode 100644 index 00000000..c6cdc9ee --- /dev/null +++ b/test/unit/adapters/test_failopen_conformance.py @@ -0,0 +1,165 @@ +"""Fail-open conformance matrix: every governing adapter × fault scenario × mode. + +This suite is the systemic fix (AAASM-4800) that ends the per-round fail-open +whack-a-mole. Rather than one bespoke regression test per adapter caught failing +open — openai_agents (AAASM-4734 → AAASM-4782), langchain (AAASM-4790) — it drives +*every* adapter that has a pre-execution tool gate through the *same* fault +scenarios under the *same* enforcement postures and asserts one invariant: + + Under an enforce posture (explicit ``enforce`` and the ``None``/omitted default), + a governance fault or deny MUST block the tool. Fail-open is only permitted under + the explicit dry-run postures (``observe`` / ``disabled``). + +Two guards make this hard to regress: + +* The parametrized matrix (:func:`test_failopen_conformance`) covers the exact + cells that previously regressed — see the ``test_regression_*`` guards below for + the named ticket ties. +* The completeness check (:func:`test_every_adapter_has_a_driver`) fails if a new or + regressed adapter package ships without a driver or a justified exemption, so an + ungoverned adapter cannot merge unnoticed. + +The harness, registry, and fake interceptors live in +``test.unit.adapters.failopen_conformance``. +""" + +from __future__ import annotations + +from test.unit.adapters import failopen_conformance as conf + +import pytest + +from agent_assembly.core.runtime_interceptor import _local_posture_is_enforce + + +@pytest.mark.asyncio +@pytest.mark.parametrize("adapter_name", sorted(conf.DRIVERS)) +@pytest.mark.parametrize("scenario", conf.SCENARIOS) +@pytest.mark.parametrize("mode_id,mode_value", conf.MODES, ids=[m[0] for m in conf.MODES]) +async def test_failopen_conformance( + adapter_name: str, + scenario: str, + mode_id: str, + mode_value: str | None, +) -> None: + """No governing adapter fails open under enforce; all pin the dry-run postures. + + ``mode_id`` names the posture for readable parametrize ids; ``mode_value`` is the + ``enforcement_mode`` fed to the real posture resolver. + """ + is_enforce = _local_posture_is_enforce(mode_value) + ran = await conf.run_scenario(adapter_name, scenario, mode_value) + want = conf.expected_run(scenario, is_enforce=is_enforce) + + posture = "enforce" if is_enforce else "fail-open" + verb = "run" if want else "be blocked" + assert ( + ran is want + ), f"{adapter_name} under {mode_id} ({posture}) on {scenario}: tool expected to {verb}, but ran={ran}" + + +def test_every_adapter_has_a_driver() -> None: + """A new/regressed adapter package cannot ship ungoverned (the anti-whack-a-mole guard). + + Every adapter package on disk must be either a governed adapter with a registered + driver or an explicitly justified exemption. Symmetrically, no driver/exemption may + reference a package that no longer exists. + """ + on_disk = conf.discover_adapter_packages() + covered = set(conf.DRIVERS) | set(conf.EXEMPT_ADAPTERS) + + ungoverned = on_disk - covered + assert not ungoverned, ( + "adapter package(s) with no fail-open conformance driver and no exemption: " + f"{sorted(ungoverned)}. Add a driver to failopen_conformance.DRIVERS, or an " + "exemption with a reason to EXEMPT_ADAPTERS." + ) + + stale = covered - on_disk + assert not stale, f"DRIVERS/EXEMPT_ADAPTERS reference package(s) not on disk: {sorted(stale)}" + + +def test_registry_and_exemptions_are_disjoint() -> None: + """An adapter is either driven or exempt — never both (keeps the guard unambiguous).""" + overlap = set(conf.DRIVERS) & set(conf.EXEMPT_ADAPTERS) + assert not overlap, f"adapter(s) both driven and exempt: {sorted(overlap)}" + + +# --- Named regression guards ---------------------------------------------- +# +# These duplicate specific matrix cells on purpose: they tie the fixed bugs to their +# tickets so a reviewer (and `git blame`) can see exactly which regression each pins. + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "scenario", + [ + conf.GATEWAY_UNREACHABLE, + conf.GOVERNANCE_ERROR, + conf.PENDING_FAULT, + conf.DENY_AUDIT_RAISES, + ], +) +async def test_regression_openai_agents_no_fail_open_under_enforce_4782(scenario: str) -> None: + """AAASM-4782/4734: openai_agents must not run the tool on S1/S2/S5/S6 under enforce.""" + ran = await conf.run_scenario("openai_agents", scenario, "enforce") + assert ran is False, f"openai_agents ran the tool under enforce on {scenario} (AAASM-4782 regression)" + + +@pytest.mark.asyncio +async def test_regression_openai_agents_deny_audit_raise_holds_under_observe_4782() -> None: + """AAASM-4782: a deny whose audit sink raises must not downgrade to an allow, even under observe. + + This is the cell that catches the specific bug: with the audit guard removed the + raising record re-enters the run path and the tool executes under the fail-open + posture. The deny is authoritative, so it must block in every posture. + """ + ran = await conf.run_scenario("openai_agents", conf.DENY_AUDIT_RAISES, "observe") + assert ran is False, "openai_agents downgraded a deny to allow when the deny-audit raised (AAASM-4782)" + + +@pytest.mark.asyncio +async def test_regression_openai_agents_raising_check_denies_under_enforce_4782() -> None: + """AAASM-4782: a governance check that RAISES must fail closed under enforce, open under observe. + + Distinct from the return-based faults in the matrix: this exercises the wrapper's + governance-error except-handler directly. + """ + from agent_assembly.exceptions import GatewayError + + class _RaisingCheckInterceptor: + def __init__(self, *, enforce: bool) -> None: + self._enforce = enforce + + def check_tool_start(self, **_kwargs: object) -> object: + raise GatewayError("runtime query failed") + + def record_result(self, **_kwargs: object) -> None: + return None + + def on_tool_end(self, **_kwargs: object) -> None: + return None + + import contextlib + + from agent_assembly.exceptions import AssemblyError + + async def _drive(enforce: bool) -> bool: + ran = [False] + with contextlib.suppress(AssemblyError): + await conf.DRIVERS["openai_agents"](_RaisingCheckInterceptor(enforce=enforce), ran) + return ran[0] + + assert await _drive(enforce=True) is False, "openai_agents ran the tool when the check raised under enforce" + assert await _drive(enforce=False) is True, "openai_agents failed closed under observe (should fail open)" + + +@pytest.mark.asyncio +async def test_regression_langchain_missing_check_method_denies_under_enforce_4790() -> None: + """AAASM-4790: langchain on_tool_start must fail closed when the interceptor has no check method.""" + blocked = await conf.run_scenario("langchain", conf.MISSING_CHECK_METHOD, "enforce") + assert blocked is False, "langchain failed open on a missing check_tool_start under enforce (AAASM-4790)" + + allowed = await conf.run_scenario("langchain", conf.MISSING_CHECK_METHOD, "observe") + assert allowed is True, "langchain failed closed on a missing check_tool_start under observe (should fail open)"