fix: address review — run_in_executor, bounded audit, consistent style

- Wrap verify() in loop.run_in_executor() to avoid blocking (matches
  AgentGovernancePlugin pattern)
- Remove unused tool_args from _log_denial/_log_allow
- Use collections.deque(maxlen=) for bounded audit log (default 1000)
- Move import json to top of file
- Use Optional[] consistently instead of mixing | None
- Fix docstring: plugin enforces permission checks at execution time,
  delegation narrowing is enforced at credential issuance

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: saneGuy

src/google/adk_community/plugins/delegation_auth_plugin.py

@@ -12,14 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""ADK plugin for delegation-scoped agent authorization.
+"""ADK plugin for permission-scoped agent authorization.
 
 Verifies that an agent holds a valid credential with sufficient permissions
-before executing a tool. Credentials are scoped: a delegated agent can only
-use a subset of its parent's permissions, never more.
+before executing a tool. Uses a pluggable verifier interface so any identity
+system (DID, JWT, ZKP, API keys, etc.) can be wired in.
 
-The plugin uses a pluggable verifier interface. A structural verifier is
-included for development; plug in a real verifier (DID, JWT, ZKP, etc.)
+A structural verifier is included for development; plug in a real verifier
 for production.
 
 Example::
@@ -34,6 +33,10 @@
 
 from __future__ import annotations
 
+import asyncio
+import collections
+import functools
+import json
 import logging
 import time
 from abc import ABC, abstractmethod
@@ -46,6 +49,8 @@
 
 logger = logging.getLogger(__name__)
 
+_MAX_AUDIT_ENTRIES_DEFAULT = 1000
+
 
 # ---------------------------------------------------------------------------
 # Pluggable verifier interface
@@ -68,6 +73,10 @@ class CredentialVerifier(ABC):
 
     Implement this to plug in any identity system: DID verification,
     JWT validation, ZKP proof checking, API key lookup, etc.
+
+    Note: ``verify()`` is called in a thread executor to avoid blocking
+    the event loop. Implementations may perform synchronous I/O (network
+    calls, file reads) safely.
     """
 
     @abstractmethod
@@ -87,12 +96,10 @@ class StructuralVerifier(CredentialVerifier):
     """Development verifier that checks credential structure only.
 
     Accepts any JSON-parseable credential with the required fields.
-    NOT for production — use a real verifier (DID, JWT, ZKP, etc.).
+    NOT for production -- use a real verifier (DID, JWT, etc.).
     """
 
     def verify(self, credential: str) -> VerificationResult:
-        import json
-
         try:
             data = json.loads(credential)
         except (json.JSONDecodeError, TypeError):
@@ -125,32 +132,23 @@ def verify(self, credential: str) -> VerificationResult:
         )
 
 
-# ---------------------------------------------------------------------------
-# Per-tool permission mapping
-# ---------------------------------------------------------------------------
-
-# Default: every tool requires these permissions. Override with
-# tool_permissions to set per-tool requirements.
-_DEFAULT_PERMISSIONS: set[str] = {"read_data"}
-
-
 # ---------------------------------------------------------------------------
 # Plugin
 # ---------------------------------------------------------------------------
 
 class DelegationAuthPlugin(BasePlugin):
-    """ADK plugin that enforces delegation-scoped authorization.
+    """ADK plugin that enforces permission-scoped authorization.
 
     Before each tool call, the plugin:
     1. Reads the agent's credential from session state
-    2. Verifies it using the configured verifier
+    2. Verifies it using the configured verifier (in a thread executor)
     3. Checks that the credential's permissions cover the tool's requirements
     4. Blocks execution if any check fails
 
-    Delegation scoping: credentials can only narrow permissions, never expand.
-    If agent A delegates to agent B, B's permission set is always a subset
-    of A's. The verifier enforces this at credential issuance; this plugin
-    enforces it at execution time.
+    Permission scoping: this plugin enforces permission checks at execution
+    time. Delegation scope narrowing (ensuring child agents can only hold a
+    subset of parent permissions) is enforced at credential issuance by the
+    identity system, not by this plugin.
 
     Args:
         required_permissions: Default permissions required for any tool call.
@@ -163,6 +161,8 @@ class DelegationAuthPlugin(BasePlugin):
             is stored. Defaults to ``"agent_credential"``.
         fail_open: If True, allow tool calls when no credential is present.
             Defaults to False.
+        max_audit_entries: Maximum audit log entries before FIFO eviction.
+            Defaults to 1000.
 
     Example::
 
@@ -177,19 +177,22 @@ class DelegationAuthPlugin(BasePlugin):
 
     def __init__(
         self,
-        required_permissions: set[str] | None = None,
-        tool_permissions: dict[str, set[str]] | None = None,
-        verifier: CredentialVerifier | None = None,
+        required_permissions: Optional[set[str]] = None,
+        tool_permissions: Optional[dict[str, set[str]]] = None,
+        verifier: Optional[CredentialVerifier] = None,
         credential_key: str = "agent_credential",
         fail_open: bool = False,
+        max_audit_entries: int = _MAX_AUDIT_ENTRIES_DEFAULT,
     ) -> None:
         super().__init__(name="delegation_auth")
-        self._required = required_permissions or _DEFAULT_PERMISSIONS
+        self._required = required_permissions or {"read_data"}
         self._tool_permissions = tool_permissions or {}
         self._verifier = verifier or StructuralVerifier()
         self._credential_key = credential_key
         self._fail_open = fail_open
-        self._audit_log: list[dict[str, Any]] = []
+        self._audit_log: collections.deque[dict[str, Any]] = (
+            collections.deque(maxlen=max_audit_entries)
+        )
 
     async def before_tool_callback(
         self,
@@ -215,7 +218,7 @@ async def before_tool_callback(
                     tool.name,
                 )
                 return None
-            self._log_denial(tool.name, "no_credential", tool_args)
+            self._log_denial(tool.name, "no_credential")
             return {
                 "error": "authorization_required",
                 "message": (
@@ -224,20 +227,26 @@ async def before_tool_callback(
                 ),
             }
 
-        # Verify the credential
+        # Verify the credential in a thread executor to avoid blocking
+        loop = asyncio.get_running_loop()
         try:
-            result = self._verifier.verify(credential)
+            result = await loop.run_in_executor(
+                None,
+                functools.partial(self._verifier.verify, credential),
+            )
         except Exception as exc:
             logger.error("Verifier raised for tool %s: %s", tool.name, exc)
-            self._log_denial(tool.name, "verifier_error", tool_args)
+            self._log_denial(tool.name, "verifier_error")
             return {
                 "error": "verification_failed",
                 "message": f"Credential verification error: {exc}",
             }
 
         if not result.valid:
             self._log_denial(
-                tool.name, result.reason or "invalid_credential", tool_args
+                tool.name,
+                result.reason or "invalid_credential",
+                agent_id=result.agent_id,
             )
             return {
                 "error": "authorization_denied",
@@ -254,7 +263,6 @@ async def before_tool_callback(
             self._log_denial(
                 tool.name,
                 f"missing_permissions: {sorted(missing)}",
-                tool_args,
                 agent_id=result.agent_id,
             )
             return {
@@ -265,19 +273,17 @@ async def before_tool_callback(
                 ),
             }
 
-        # Authorized — log and proceed
+        # Authorized
         logger.info(
             "Agent %s authorized for %s (permissions: %s)",
             result.agent_id,
             tool.name,
             sorted(result.permissions),
         )
-        self._log_allow(tool.name, result.agent_id, tool_args)
+        self._log_allow(tool.name, result.agent_id)
         return None
 
-    def _log_allow(
-        self, tool: str, agent_id: str, tool_args: dict[str, Any]
-    ) -> None:
+    def _log_allow(self, tool: str, agent_id: str) -> None:
         self._audit_log.append({
             "action": "allow",
             "tool": tool,
@@ -289,7 +295,6 @@ def _log_denial(
         self,
         tool: str,
         reason: str,
-        tool_args: dict[str, Any],
         agent_id: str = "",
     ) -> None:
         self._audit_log.append({
@@ -303,5 +308,5 @@ def _log_denial(
 
     @property
     def audit_log(self) -> list[dict[str, Any]]:
-        """Read-only access to the audit trail."""
+        """Read-only copy of the audit trail."""
         return list(self._audit_log)