[AAASM-5665] 🐛 (adapters): Emit an audit record when a tool call is denied

agent_assembly/adapters/_shared/tool_governance.py

@@ -4,8 +4,9 @@
 framework-specific hook points, but the governance logic they run once a tool
 call is intercepted is identical: serialize the args, ask the interceptor for a
 verdict, honour a ``pending`` approval round-trip, deny by raising when the
-verdict is ``deny``, otherwise run the original inside a spawn-context scope and
-record the result. That shared body — previously duplicated verbatim in both
+verdict is ``deny``, otherwise run the original inside a spawn-context scope.
+Either way the outcome is recorded through the audit hook before the flow ends
+(AAASM-5665). That shared body — previously duplicated verbatim in both
 adapters (the cross-file duplication SonarCloud flagged on PR #269, AAASM-4746) —
 lives here so each adapter keeps only its framework-specific glue.
 
@@ -19,6 +20,7 @@
 
 from __future__ import annotations
 
+import contextlib
 import inspect
 from collections.abc import Callable, Mapping
 from typing import TYPE_CHECKING, Any, Literal
@@ -39,6 +41,11 @@
 
 _MAX_AUDIT_RESULT_CHARS = 2000
 
+_KEYWORD_PARAMETER_KINDS = (
+    inspect.Parameter.KEYWORD_ONLY,
+    inspect.Parameter.POSITIONAL_OR_KEYWORD,
+)
+
 
 def _current_spawn_depth() -> int:
     current = _SPAWN_CTX.get()
@@ -127,21 +134,67 @@ def _truncate_result_for_audit(result: object) -> str:
     return str(result)[:_MAX_AUDIT_RESULT_CHARS]
 
 
+def _accepts_keyword(method: Any, name: str) -> bool:
+    """Whether ``method`` can be called with the ``name`` keyword.
+
+    The audit hook is duck-typed — adapters and user code supply their own
+    ``record_result`` / ``on_tool_end`` — and every existing implementation was
+    written against the four-keyword call, so passing a new keyword
+    unconditionally would raise ``TypeError`` on all of them. The denial flag is
+    therefore offered only to handlers that can receive it (an explicit
+    parameter or a ``**kwargs`` catch-all); the rest still get the record, just
+    without the flag.
+    """
+    try:
+        signature = inspect.signature(method)
+    except (TypeError, ValueError):
+        # C-implemented callables expose no introspectable signature. Fall back
+        # to the narrow call so the record is still emitted.
+        return False
+    for parameter in signature.parameters.values():
+        if parameter.kind is inspect.Parameter.VAR_KEYWORD:
+            return True
+        if parameter.name == name and parameter.kind in _KEYWORD_PARAMETER_KINDS:
+            return True
+    return False
+
+
 async def _record_async_tool_result(
     callback_handler: Any,
     *,
     tool_name: str,
     result: object,
     agent_id: str | None,
     run_id: str | None,
+    denied: bool = False,
 ) -> None:
+    """Offer the outcome of one governed tool call to the audit hook.
+
+    Called for a denied call as well as an executed one (AAASM-5665). On the
+    denied path ``result`` carries the denial message and ``denied`` is ``True``
+    so a handler that understands the flag can tell "denied before execution"
+    apart from a tool that ran and returned that same text.
+
+    Whether anything is recorded depends entirely on the ``callback_handler``.
+    Both hooks are duck-typed, and on the interceptor the SDK builds today
+    *neither resolves*: ``RuntimeQueryInterceptor`` defines only
+    ``check_tool_start`` and delegates the rest to ``GatewayClient``, whose
+    surface has no ``record_result`` and no ``on_tool_end``. So on the shipped
+    path this function finds no hook and emits nothing — for allowed calls as
+    much as denied ones — leaving tool outcomes Unmeasured in audit evidence
+    (ADR 0033 §6). A caller that supplies its own handler does get the record;
+    wiring a sink into the SDK's own interceptor is a separate capability.
+    """
+    denial_flag = {"denied": denied} if denied else {}
+
     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,
             run_id=run_id,
+            **(denial_flag if _accepts_keyword(record_method, "denied") else {}),
         )
         if inspect.isawaitable(recorded):
             await recorded
@@ -154,6 +207,7 @@ async def _record_async_tool_result(
             tool_name=tool_name,
             agent_id=agent_id,
             run_id=run_id,
+            **(denial_flag if _accepts_keyword(tool_end_method, "denied") else {}),
         )
         if inspect.isawaitable(recorded):
             await recorded
@@ -223,9 +277,35 @@ async def run_governed_async_tool(
     # non-decision, not a grant — blocking it here stops it from falling through
     # and running the tool, matching the LangChain handler.
     if status != "allow":
-        if is_pending_flow:
-            raise _build_pending_rejected_error(tool_name, reason)
-        raise _build_denied_error(tool_name, reason)
+        error = (
+            _build_pending_rejected_error(tool_name, reason)
+            if is_pending_flow
+            else _build_denied_error(tool_name, reason)
+        )
+        # Offer the deny to the audit hook before raising (AAASM-5665).
+        # Previously this raised straight past the record call below, so a
+        # denied call could not reach an audit sink even when the caller had
+        # supplied one. See _record_async_tool_result on why the SDK's own
+        # interceptor still resolves no hook, leaving the shipped path
+        # Unmeasured.
+        #
+        # Best-effort, and the guard is load-bearing: the hook is duck-typed
+        # from caller-supplied code, and inserting a call here where none used
+        # to exist would otherwise let a raising handler replace a decided deny
+        # with its own exception — a caller matching on PolicyViolationError
+        # would stop recognising the deny. A decided deny is final regardless of
+        # audit outcome; this repo already settled that for the openai_agents
+        # path under AAASM-4782, so follow it rather than invent a second answer.
+        with contextlib.suppress(Exception):
+            await _record_async_tool_result(
+                callback_handler,
+                tool_name=tool_name,
+                result=str(error),
+                agent_id=agent_id,
+                run_id=run_id,
+                denied=True,
+            )
+        raise error
 
     spawn_ctx = SpawnContext(
         parent_agent_id=agent_id or "",

test/unit/adapters/_shared/test_tool_governance.py

@@ -47,3 +47,143 @@ def invoke_original() -> str:
         )
 
     assert ran == []
+
+
+class _FourKeywordHandler:
+    """A handler written against the pre-AAASM-5665 four-keyword hook.
+
+    No ``denied`` parameter and no ``**kwargs``, so passing the flag to it would
+    raise ``TypeError``. Every in-tree adapter hook and any third-party one
+    predates the flag, which is why the flag is offered conditionally.
+    """
+
+    def __init__(self) -> None:
+        self.calls: list[dict[str, Any]] = []
+
+    def record_result(self, *, tool_name: str, result: str, agent_id: str | None, run_id: str | None) -> None:
+        self.calls.append({"tool_name": tool_name, "result": result, "agent_id": agent_id, "run_id": run_id})
+
+
+class _UnreadableSignatureHook:
+    """A callable whose signature cannot be introspected.
+
+    ``inspect.signature`` raises ``ValueError`` for C-implemented callables
+    (``dict`` and ``type`` do so on CPython), and proxy/wrapper objects
+    reproduce it by raising from ``__signature__``. Either way the SDK cannot
+    prove the callable accepts the flag, so it must fall back to the narrow
+    call rather than risk a ``TypeError`` that would swallow the deny.
+    """
+
+    def __init__(self) -> None:
+        self.calls: list[dict[str, Any]] = []
+
+    def __call__(self, **kwargs: Any) -> None:
+        self.calls.append(kwargs)
+
+    @property
+    def __signature__(self) -> Any:
+        raise ValueError("no signature found for builtin")
+
+
+class _DenyingHandlerMixin:
+    def check_tool_start(self, **kwargs: Any) -> dict[str, str]:
+        del kwargs
+        return {"status": "deny", "reason": "policy forbids this tool"}
+
+
+class _FourKeywordDenyHandler(_DenyingHandlerMixin, _FourKeywordHandler):
+    pass
+
+
+class _UnreadableSignatureDenyHandler(_DenyingHandlerMixin):
+    def __init__(self) -> None:
+        self.record_result = _UnreadableSignatureHook()
+
+
+async def _run_denied(handler: Any) -> None:
+    ran: list[bool] = []
+
+    def invoke_original() -> str:
+        ran.append(True)
+        return "tool-result"
+
+    with pytest.raises(PolicyViolationError, match="policy forbids this tool"):
+        await tool_governance.run_governed_async_tool(
+            handler,
+            enforce=True,
+            tool_name="write_to_disk",
+            tool_args={"path": "/tmp/x"},
+            agent_id="agent-1",
+            run_id="run-1",
+            invoke_original=invoke_original,
+        )
+
+    assert ran == [], "the denied tool body ran"
+
+
+class _RaisingAuditHandler(_DenyingHandlerMixin):
+    """A handler whose audit hook raises — reachable, since the hook is caller-supplied."""
+
+    def __init__(self) -> None:
+        self.attempts = 0
+
+    def record_result(self, **kwargs: Any) -> None:
+        del kwargs
+        self.attempts += 1
+        raise RuntimeError("audit handler exploded")
+
+
+@pytest.mark.asyncio
+async def test_a_raising_audit_handler_does_not_replace_the_policy_denial() -> None:
+    """Reads the exception the governed call actually raises.
+
+    Recording the deny put a caller-supplied hook on a path that previously did
+    not touch it, so a raising handler could substitute its own exception for
+    the denial. A decided deny is final regardless of audit outcome
+    (AAASM-4782); a caller matching on PolicyViolationError must still see one.
+    """
+    handler = _RaisingAuditHandler()
+
+    await _run_denied(handler)
+
+    # Distinguishes "the audit failure was contained" from "the hook was never
+    # reached", which would satisfy the assertion above for the wrong reason.
+    assert handler.attempts == 1
+
+
+@pytest.mark.asyncio
+async def test_a_four_keyword_handler_still_receives_the_deny_record_without_the_flag() -> None:
+    """Reads the arguments the flow passed the handler's own record_result.
+
+    This is the backward-compatibility guarantee: adding ``denied`` must not
+    stop a handler that predates it from being recorded to. If the flag were
+    passed unconditionally this call would raise ``TypeError`` from inside the
+    governance flow and replace the policy denial with an unrelated error.
+    """
+    handler = _FourKeywordDenyHandler()
+
+    await _run_denied(handler)
+
+    assert len(handler.calls) == 1
+    call = handler.calls[0]
+    assert call["tool_name"] == "write_to_disk"
+    assert call["agent_id"] == "agent-1"
+    assert call["run_id"] == "run-1"
+    assert "policy forbids this tool" in call["result"]
+    # The flag is absent rather than False: the handler cannot receive it.
+    assert "denied" not in call
+
+
+@pytest.mark.asyncio
+async def test_a_hook_with_no_readable_signature_still_receives_the_deny_record() -> None:
+    """Reads the keywords the flow passed an unintrospectable callable hook."""
+    handler = _UnreadableSignatureDenyHandler()
+
+    await _run_denied(handler)
+
+    assert len(handler.record_result.calls) == 1
+    call = handler.record_result.calls[0]
+    assert call["tool_name"] == "write_to_disk"
+    assert call["run_id"] == "run-1"
+    # Narrow call: the flag is withheld because acceptance could not be proven.
+    assert "denied" not in call

test/unit/negative_control.py

@@ -124,22 +124,34 @@ def start_network_side_effect() -> NetworkSideEffect:
 
 @dataclass
 class RecordedResult:
-    """One post-execution audit record the governed path emitted."""
+    """One audit record the governed path emitted.
+
+    ``denied`` separates "denied before execution" from a tool that ran and
+    returned the denial text — on the denied path ``result`` carries the
+    policy-violation message, so the flag is what makes the two distinguishable.
+    """
 
     tool_name: str
     agent_id: str | None
     run_id: str | None
     result: str
+    denied: bool = False
 
 
 class AuditRecordingInterceptor:
     """Delegates every governance call to ``inner`` and records the audit hook.
 
     Only the post-execution ``record_result`` hook is added — the authoritative
     verdict still comes from the wrapped interceptor, so the deny under test is
-    the real one. This exists because the SDK's ``GatewayClient`` implements no
-    audit sink of its own (the interceptor is the only one), and AAASM-5529
-    requires deny/allow evidence to carry agent and tool identity.
+    the real one.
+
+    ``record_result`` is a hook the SDK genuinely calls, not a fixture
+    invention: ``_shared.tool_governance._record_async_tool_result`` duck-types
+    it on the callback handler (with an ``on_tool_end`` fallback), and seven
+    adapters route through it. The fixture supplies it because the real
+    ``GatewayClient`` implements no audit sink of its own — it exposes only
+    ``report_edge`` — so without a handler that accepts the hook there is
+    nothing to read the record off.
     """
 
     def __init__(self, inner: Any) -> None:
@@ -156,8 +168,17 @@ def record_result(
         result: str,
         agent_id: str | None = None,
         run_id: str | None = None,
+        denied: bool = False,
     ) -> None:
-        self.records.append(RecordedResult(tool_name=tool_name, agent_id=agent_id, run_id=run_id, result=result))
+        self.records.append(
+            RecordedResult(
+                tool_name=tool_name,
+                agent_id=agent_id,
+                run_id=run_id,
+                result=result,
+                denied=denied,
+            )
+        )
 
     def __getattr__(self, name: str) -> Any:
         return getattr(self._inner, name)

test/unit/test_quickstart_negative_control.py

@@ -273,6 +273,44 @@ def test_the_runtime_saw_the_agent_and_tool_the_deny_was_decided_against(
         assert tool_name == "write_to_disk"
         assert tool_args_of(quickstart.runtime.query_calls[0]) == {"path": str(file_effect.path)}
 
+    def test_a_denied_call_emits_an_audit_record_carrying_the_agent_and_tool(
+        self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect
+    ) -> None:
+        quickstart = _init_quickstart(monkeypatch, decision="deny", reason="policy forbids disk writes")
+        try:
+            outcome = _settle(
+                lambda: quickstart.call(
+                    "write_to_disk", {"path": str(file_effect.path)}, lambda: file_effect.write("denied")
+                )
+            )
+        finally:
+            quickstart.context.shutdown()
+
+        # Absence first, as everywhere else in this suite.
+        assert file_effect.occurred() is False
+        assert isinstance(outcome, ToolExecutionBlockedError)
+
+        # The load-bearing assertion for AAASM-5665, and the one the test above
+        # cannot make: the record handed to the audit hook, not the policy
+        # query and not the raised exception. Before this the deny raised
+        # straight past the hook, so a denied call offered it nothing.
+        #
+        # Scope of the evidence: the record is captured by this fixture's
+        # handler. The interceptor the SDK builds resolves no audit hook at all
+        # (RuntimeQueryInterceptor + GatewayClient expose neither
+        # record_result nor on_tool_end), so tool outcomes are Unmeasured in
+        # audit evidence on the shipped path. What this pins is the governance
+        # flow's call — the part fixable without wiring a sink.
+        assert len(quickstart.interceptor.records) == 1
+        record = quickstart.interceptor.records[0]
+        assert record.tool_name == "write_to_disk"
+        assert record.agent_id == _AGENT_ID
+        assert record.run_id == "run-1"
+        # Distinguishes "denied before execution" from a tool that ran and
+        # returned this same text.
+        assert record.denied is True
+        assert "policy forbids disk writes" in record.result
+
     def test_an_allowed_call_is_recorded_with_the_same_identity(
         self, monkeypatch: pytest.MonkeyPatch, file_effect: FileSideEffect
     ) -> None: