[AAASM-4734] 🐛 (adapters): Fail closed on unrecognized verdict and missing interceptor

agent_assembly/adapters/google_adk/patch.py

@@ -16,6 +16,7 @@
     _get_pending_tool_approval_timeout_seconds as _resolve_pending_timeout_seconds,
 )
 from agent_assembly.adapters.crewai.patch import (
+    _interceptor_enforces,
     _missing_interceptor_decision,
 )
 from agent_assembly.adapters.crewai.patch import (
@@ -171,6 +172,7 @@ def _apply_tool_run_async_patch(tool_cls: type[Any], callback_handler: Any) -> N
         return None
 
     original_run_async = tool_cls.run_async
+    enforce = _interceptor_enforces(callback_handler)
 
     @wraps(original_run_async)
     async def patched_run_async(self: Any, *, args: Any, tool_context: Any, **kwargs: Any) -> Any:
@@ -186,7 +188,7 @@ async def patched_run_async(self: Any, *, args: Any, tool_context: Any, **kwargs
             agent_id=agent_id,
             run_id=run_id,
         )
-        status, reason = _normalize_decision(decision)
+        status, reason = _normalize_decision(decision, enforce=enforce)
         is_pending_flow = False
         if status == "pending":
             is_pending_flow = True
@@ -199,7 +201,7 @@ async def patched_run_async(self: Any, *, args: Any, tool_context: Any, **kwargs
                 agent_id=agent_id,
                 run_id=run_id,
             )
-            status, reason = _normalize_decision(final_decision)
+            status, reason = _normalize_decision(final_decision, enforce=enforce)
 
         if status == "deny":
             if is_pending_flow:
@@ -291,8 +293,10 @@ def _serialize_tool_args(args: Any) -> dict[str, Any]:
 
 def _normalize_decision(
     decision: object,
+    *,
+    enforce: bool = False,
 ) -> tuple[Literal["allow", "deny", "pending"], str | None]:
-    return _normalize_governance_decision(decision)
+    return _normalize_governance_decision(decision, enforce=enforce)
 
 
 async def _invoke_async_tool_check(

agent_assembly/adapters/mcp/patch.py

@@ -18,6 +18,9 @@
 from agent_assembly.adapters.crewai.patch import (
     _interceptor_enforces as _resolve_interceptor_enforces,
 )
+from agent_assembly.adapters.crewai.patch import (
+    _missing_interceptor_decision,
+)
 from agent_assembly.adapters.crewai.patch import (
     _normalize_decision as _normalize_governance_decision,
 )
@@ -141,7 +144,7 @@ async def _invoke_async_tool_check(
     target = _resolve_governance_target(callback_handler)
     method = getattr(target, "check_tool_start", None)
     if not callable(method):
-        return {"status": "allow"}
+        return _missing_interceptor_decision(callback_handler)
 
     result = method(
         serialized={"name": tool_name},

agent_assembly/adapters/openai_agents/patch.py

@@ -34,6 +34,10 @@
 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,
+    _missing_interceptor_decision,
+)
 from agent_assembly.adapters.crewai.patch import (
     _normalize_decision as _normalize_governance_decision,
 )
@@ -306,8 +310,10 @@ def _resolve_agent_id(ctx: Any) -> str | None:
 
 def _normalize_decision(
     decision: object,
+    *,
+    enforce: bool = False,
 ) -> tuple[Literal["allow", "deny", "pending"], str | None]:
-    return _normalize_governance_decision(decision)
+    return _normalize_governance_decision(decision, enforce=enforce)
 
 
 def _resolve_governance_target(callback_handler: Any) -> Any:
@@ -328,7 +334,7 @@ async def _invoke_async_tool_check(
     target = _resolve_governance_target(callback_handler)
     method = getattr(target, "check_tool_start", None)
     if not callable(method):
-        return {"status": "allow"}
+        return _missing_interceptor_decision(callback_handler)
 
     result = method(
         serialized={"name": tool_name},
@@ -477,6 +483,8 @@ def _wrap_on_invoke_tool(tool_obj: Any, callback_handler: Any) -> None:
     if getattr(original_invoke, _WRAPPED_INVOKE_FLAG, False):
         return None
 
+    enforce = _interceptor_enforces(callback_handler)
+
     @wraps(original_invoke)
     async def governed_invoke(ctx: Any, tool_input: Any) -> Any:
         tool_name = str(getattr(tool_obj, "name", tool_obj.__class__.__name__))
@@ -491,7 +499,7 @@ async def governed_invoke(ctx: Any, tool_input: Any) -> Any:
                 agent_id=agent_id,
                 ctx=ctx,
             )
-            status, reason = _normalize_decision(decision)
+            status, reason = _normalize_decision(decision, enforce=enforce)
             is_pending_flow = False
             if status == "pending":
                 is_pending_flow = True
@@ -504,7 +512,7 @@ async def governed_invoke(ctx: Any, tool_input: Any) -> Any:
                     agent_id=agent_id,
                     ctx=ctx,
                 )
-                status, reason = _normalize_decision(final_decision)
+                status, reason = _normalize_decision(final_decision, enforce=enforce)
 
             if status == "deny":
                 blocked_result = _build_tool_deny_error(

agent_assembly/adapters/pydantic_ai/patch.py

@@ -16,6 +16,7 @@
     _get_pending_tool_approval_timeout_seconds as _resolve_pending_timeout_seconds,
 )
 from agent_assembly.adapters.crewai.patch import (
+    _interceptor_enforces,
     _missing_interceptor_decision,
 )
 from agent_assembly.adapters.crewai.patch import (
@@ -294,6 +295,8 @@ def _apply_tool_run_patch(tool_cls: type[Any], callback_handler: Any) -> bool:
     if not callable(original_run):
         return False
 
+    enforce = _interceptor_enforces(callback_handler)
+
     @wraps(original_run)
     async def patched_run(self: Any, ctx: Any, args: Any, **kwargs: Any) -> Any:
         tool_name = str(getattr(self, "name", self.__class__.__name__))
@@ -308,7 +311,7 @@ async def patched_run(self: Any, ctx: Any, args: Any, **kwargs: Any) -> Any:
             agent_id=agent_id,
             run_id=run_id,
         )
-        status, reason = _normalize_decision(decision)
+        status, reason = _normalize_decision(decision, enforce=enforce)
         is_pending_flow = False
         if status == "pending":
             is_pending_flow = True
@@ -321,7 +324,7 @@ async def patched_run(self: Any, ctx: Any, args: Any, **kwargs: Any) -> Any:
                 agent_id=agent_id,
                 run_id=run_id,
             )
-            status, reason = _normalize_decision(final_decision)
+            status, reason = _normalize_decision(final_decision, enforce=enforce)
 
         if status == "deny":
             if is_pending_flow:
@@ -384,6 +387,8 @@ def _apply_toolset_call_tool_patch(toolset_cls: type[Any], callback_handler: Any
     if not callable(original_call_tool):
         return False
 
+    enforce = _interceptor_enforces(callback_handler)
+
     @wraps(original_call_tool)
     async def patched_call_tool(self: Any, name: Any, tool_args: Any, ctx: Any, tool: Any, **kwargs: Any) -> Any:
         tool_name = str(name)
@@ -398,7 +403,7 @@ async def patched_call_tool(self: Any, name: Any, tool_args: Any, ctx: Any, tool
             agent_id=agent_id,
             run_id=run_id,
         )
-        status, reason = _normalize_decision(decision)
+        status, reason = _normalize_decision(decision, enforce=enforce)
         is_pending_flow = False
         if status == "pending":
             is_pending_flow = True
@@ -411,7 +416,7 @@ async def patched_call_tool(self: Any, name: Any, tool_args: Any, ctx: Any, tool
                 agent_id=agent_id,
                 run_id=run_id,
             )
-            status, reason = _normalize_decision(final_decision)
+            status, reason = _normalize_decision(final_decision, enforce=enforce)
 
         if status == "deny":
             if is_pending_flow:
@@ -503,8 +508,10 @@ def _serialize_tool_args(args: Any) -> dict[str, Any]:
 
 def _normalize_decision(
     decision: object,
+    *,
+    enforce: bool = False,
 ) -> tuple[Literal["allow", "deny", "pending"], str | None]:
-    return _normalize_governance_decision(decision)
+    return _normalize_governance_decision(decision, enforce=enforce)
 
 
 async def _invoke_async_tool_check(

agent_assembly/core/assembly.py

@@ -21,6 +21,7 @@
     resolve_gateway_url,
 )
 from agent_assembly.core.runtime_interceptor import (
+    _local_posture_is_enforce,
     _native_core_available,
     build_governance_interceptor,
     connect_runtime_client,
@@ -394,8 +395,8 @@ def _register_agent_with_gateway(
     ``register`` raises, the failure is no longer silent: a loud
     :func:`_warn_agent_unregistered` fires and ``False`` is returned (AAASM-4547).
     Init still proceeds so the proxy / eBPF layers stay authoritative — except
-    under ``enforce``, where a ``register`` failure propagates so a misconfigured
-    gateway fails init closed.
+    under an enforce posture (the ``None`` default or explicit ``enforce``), where a
+    ``register`` failure propagates so a misconfigured gateway fails init closed.
     """
     if runtime_client is None:
         if native_available:
@@ -418,7 +419,7 @@ def _register_agent_with_gateway(
             parent_agent_id=parent_agent_id,
         )
     except Exception as error:
-        if enforcement_mode == "enforce":
+        if _local_posture_is_enforce(enforcement_mode):
             raise
         _warn_agent_unregistered(f"registration failed: {error}")
         return False

test/unit/adapters/enforce_helpers.py

@@ -0,0 +1,47 @@
+"""Shared stand-ins for the AAASM-4734 fail-closed-under-enforce regression tests.
+
+Every governed adapter must DENY (not fall open) under an enforce posture for two
+inputs that previously slipped through: a malformed/unrecognized verdict, and an
+interceptor exposing no ``check_tool_start``. The interceptor stand-ins and the
+parametrization list are shared here so the four adapters' tests exercise both
+scenarios without copy-pasting the same arrange/assert into each file (which
+SonarCloud otherwise flags as duplicated lines).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+class EnforcingUnknownInterceptor:
+    """Enforcing interceptor whose ``check_tool_start`` returns a malformed verdict.
+
+    ``_enforce = True`` marks the fail-closed posture; ``None`` is an unrecognized
+    verdict that must be denied rather than silently allowed.
+    """
+
+    _enforce = True
+
+    async def check_tool_start(self, **kwargs: object) -> object:
+        del kwargs
+        return None
+
+
+class EnforcingHandlerWithoutCheck:
+    """Enforcing handler that exposes no ``check_tool_start`` at all.
+
+    The missing-method path must fail closed under enforce instead of the former
+    hardcoded allow.
+    """
+
+    _enforce = True
+
+
+# Both fail-closed scenarios as interceptor-factory params. Adapters that cover
+# both cases (openai_agents / pydantic_ai / google_adk) parametrize their deny
+# test over this list; mcp already had the malformed-verdict test, so it uses the
+# missing-check handler directly.
+ENFORCE_DENY_CASES = [
+    pytest.param(EnforcingUnknownInterceptor, id="unknown_verdict"),
+    pytest.param(EnforcingHandlerWithoutCheck, id="missing_check_tool_start"),
+]

test/unit/adapters/google_adk/test_google_adk_patch.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from test.unit.adapters.enforce_helpers import ENFORCE_DENY_CASES
 from types import SimpleNamespace
 from typing import Any
 
@@ -461,3 +462,23 @@ async def test_revert_restores_concrete_function_tool_override(
     assert FakeFunctionTool.run_async is original_function
     assert getattr(FakeBaseTool, google_adk_patch._TOOLS_PATCHED_FLAG, False) is False
     assert getattr(FakeFunctionTool, google_adk_patch._TOOLS_PATCHED_FLAG, False) is False
+
+
+# --- AAASM-4734: fail closed on unrecognized verdict / missing interceptor ---
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("interceptor_factory", ENFORCE_DENY_CASES)
+async def test_denies_under_enforce(
+    monkeypatch: pytest.MonkeyPatch,
+    interceptor_factory: type,
+) -> None:
+    FakeBaseTool = _install_fake_google_adk_modules(monkeypatch)
+
+    patcher = google_adk_patch.GoogleADKPatch(interceptor_factory())
+    assert patcher.apply() is True
+
+    tool = FakeBaseTool()
+    tool_context = SimpleNamespace(invocation_context=None)
+    with pytest.raises(PolicyViolationError):
+        await tool.run_async(args={"step": 1}, tool_context=tool_context)

test/unit/adapters/mcp/test_patch.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from test.unit.adapters.enforce_helpers import EnforcingHandlerWithoutCheck
 from types import SimpleNamespace
 from typing import Any
 
@@ -298,3 +299,22 @@ async def check_tool_start(self, **kwargs: object) -> object:
 
     result = await FakeClientSession().call_tool(name="allowed_tool", arguments={"ok": True})
     assert result["name"] == "allowed_tool"
+
+
+# --- AAASM-4734: missing check_tool_start must fail closed under enforce ---
+
+
+@pytest.mark.asyncio
+async def test_missing_check_tool_start_blocks_tool_under_enforce(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Previously this path hardcoded allow; under enforce a handler exposing no
+    # check_tool_start must block instead.
+    FakeClientSession = _install_fake_mcp_module(monkeypatch)
+
+    patcher = mcp_patch.MCPClientPatch(EnforcingHandlerWithoutCheck(), process_agent_id="agent-9")
+    assert patcher.apply() is True
+
+    session = FakeClientSession()
+    with pytest.raises(MCPToolBlockedError):
+        await session.call_tool("some_tool", {"q": "x"})

test/unit/adapters/openai_agents/test_patch.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+from test.unit.adapters.enforce_helpers import ENFORCE_DENY_CASES
 from types import SimpleNamespace
 from typing import Any
 
@@ -514,3 +515,27 @@ async def on_tool_end(self, **kwargs: object) -> None:
     )
 
     assert observed
+
+
+# --- AAASM-4734: fail closed on unrecognized verdict / missing interceptor ---
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("interceptor_factory", ENFORCE_DENY_CASES)
+async def test_denies_under_enforce(
+    monkeypatch: pytest.MonkeyPatch,
+    interceptor_factory: type,
+) -> None:
+    function_tool_cls = _install_fake_openai_agents_module(monkeypatch)
+
+    patcher = openai_patch.OpenAIAgentsPatch(callback_handler=interceptor_factory())
+    assert patcher.apply() is True
+
+    tool = function_tool_cls(name="governed_tool")
+    result = await tool.on_invoke_tool(SimpleNamespace(agent_id="a"), "{}")
+
+    # A fail-open patch would return the tool's real dict output; under enforce
+    # both an unrecognized verdict and a missing interceptor must be blocked
+    # (returned as a deny string).
+    assert isinstance(result, str)
+    assert "blocked by governance policy" in result