diff --git a/contributing/samples/authority_routing/README.md b/contributing/samples/authority_routing/README.md new file mode 100644 index 0000000..23cdfc7 --- /dev/null +++ b/contributing/samples/authority_routing/README.md @@ -0,0 +1,149 @@ +# Authority routing plugin for Google ADK + +An ADK plugin that decides an **authority posture** — ADVISE / EXECUTE / DEFER / +STOP — for every tool call *before* it runs. Tool schemas and permissions govern +*access*; this plugin governs *authorization*: whether the agent was allowed to +act at all. + +It is the layer where two common agent failures live: + +- **advice becomes action** — the user asked for a recommendation and the agent + executed anyway; +- **scope expansion** — one bounded edit ballooned into a dependent workflow; +- **missing approval** — an approval was clearly required and the agent proceeded. + +## Install + +```bash +pip install google-adk-community +``` + +No extra dependencies — the plugin is self-contained. + +## The four postures + +| Posture | Meaning | +| --------- | ------------------------------------------------------------- | +| `ADVISE` | Produce recommendations only; no tool side effects. | +| `EXECUTE` | Act, within an explicitly bounded scope. | +| `DEFER` | Pause and request authorization when scope/approval is unclear. | +| `STOP` | Refuse or block when the requested work is not allowed. | + +## How it works + +Two stages produce the final posture: + +1. **Posture router** (optional, caller-supplied) — typically a model call that + classifies the request and returns a `PostureVerdict`. Its verdict is + *advisory*. +2. **Deterministic guard** — an irreversibility keyword tripwire (scanning the + tool name and argument values) plus an approval-state check on session + context. It composes with the router under **most-restrictive-wins**: plain + code can only make the posture *stricter*, never looser. + +The plugin is **fail-closed**: a router that raises or returns an unparseable +verdict defaults to `DEFER` — an unreadable authorization is not an +authorization. With **no router configured**, the plugin runs guard-only: a +fully deterministic, dependency-free layer that `DEFER`s irreversible or +approval-pending tool calls and lets everything else `EXECUTE`. + +When the final posture is anything other than `EXECUTE`, `before_tool_callback` +returns a block dict that short-circuits the tool (per the ADK plugin contract): + +```json +{ + "error": "authority_blocked", + "posture": "DEFER", + "reason": "deterministic guard: irreversibility tripwire: 'delete'", + "scope": "" +} +``` + +## Usage + +```python +from google.adk.agents import Agent +from google.adk.runners import Runner +from google.adk.sessions import InMemorySessionService +from google.adk_community.plugins import ( + AuthorityRequest, + AuthorityRoutingPlugin, + Posture, + PostureVerdict, +) + +# Guard-only: no model, fully deterministic. +plugin = AuthorityRoutingPlugin() + +# Or with a model-backed router: +def route(req: AuthorityRequest) -> PostureVerdict: + # ...call your model, parse a structured verdict... + return PostureVerdict(posture=Posture.EXECUTE, scope="update one row") + +plugin = AuthorityRoutingPlugin(posture_router=route) + +runner = Runner( + agent=Agent(name="my_agent", model="gemini-2.0-flash"), + plugins=[plugin], + app_name="my-app", + session_service=InMemorySessionService(), +) +``` + +The standing context the guard scans for approval markers is read from session +state under `authority_context` (configurable via `context_state_key`). + +## Run the sample + +```bash +python contributing/samples/authority_routing/main.py +``` + +Live output (no API key, no network — the demo router is a deterministic stub; +tool call logs go to stderr): + +``` +Runner created with authority-routing plugin enabled. + +CASE TOOL POSTURE OUTCOME +------------------------------------------------------------------------------------------------ +authorized read — router EXECUTE, guard clear web_search EXECUTE tool runs +advice requested — agent must NOT execute update_ticket ADVISE blocked — user asked for a recommendation, not an action +irreversible call — guard overrides EXECUTE delete_records DEFER blocked — deterministic guard: irreversibility tripwire: 'delete' +approval pending — guard floors to DEFER send_email DEFER blocked — deterministic guard: approval-state marker in context: 'awaiting approval' +disallowed — STOP survives strictest-wins read_vault STOP blocked — blocked by standing policy +``` + +Row 3 is the key case: the router authorized `EXECUTE`, but the tool call is +irreversible, so the deterministic guard floored it to `DEFER`. Row 5 shows the +other direction — the router's `STOP` survives even though the guard raised no +concern of its own. + +## Configuration + +| Argument | Default | Purpose | +| -------------------------- | -------------------- | -------------------------------------------------- | +| `posture_router` | `None` (guard-only) | Callable classifying an `AuthorityRequest`. | +| `context_state_key` | `"authority_context"`| Session-state key scanned for approval markers. | +| `irreversible_keywords` | built-in list | Override the tripwire keywords. | +| `pending_approval_markers` | built-in list | Override the approval-state markers. | +| `fail_closed` | `True` | On bad router output: `True` → DEFER, `False` → ignore router. | + +## Relationship to other governance plugins + +This plugin is complementary to `AgentGovernancePlugin` (policy-as-code +allow/deny) and the HITL approval gateway. Those answer *"is this tool call +permitted by policy?"*; this one answers *"was the agent authorized to act at +all, and if so, within what scope?"* — and it degrades to a useful deterministic +default with no external dependency and no model. + +## Provenance + +The four-posture pattern, the most-restrictive-wins composition, and the +tripwire come from a live 24/7 multi-agent system and were first published as an +Anthropic cookbook notebook. Design and validation are the author's; +implementation was drafted with AI assistance, then reviewed and tested by hand. + +## Links + +- [ADK Plugin docs](https://google.github.io/adk-docs/plugins/) diff --git a/contributing/samples/authority_routing/main.py b/contributing/samples/authority_routing/main.py new file mode 100644 index 0000000..e487d73 --- /dev/null +++ b/contributing/samples/authority_routing/main.py @@ -0,0 +1,167 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Example: Google ADK agent with authority-posture routing. + +Demonstrates: +1. Wiring AuthorityRoutingPlugin into an ADK Runner. +2. How the deterministic guard (irreversibility tripwire + approval-state + check) alone, with no model, already blocks unauthorized tool calls. +3. How an optional posture router composes with the guard under + most-restrictive-wins. + +Run it directly — it needs no API key and no network: + + python contributing/samples/authority_routing/main.py +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass + +from google.adk.agents import Agent +from google.adk.runners import Runner +from google.adk.sessions import InMemorySessionService + +from google.adk_community.plugins import AuthorityRequest +from google.adk_community.plugins import AuthorityRoutingPlugin +from google.adk_community.plugins import Posture +from google.adk_community.plugins import PostureVerdict + + +def demo_router(request: AuthorityRequest) -> PostureVerdict: + """A tiny illustrative router (stands in for a model call). + + Real deployments pass the request to an LLM and parse a structured verdict. + Here we key off markers the caller left in context so the demo is fully + deterministic: + + * ``policy: denied`` -> STOP (the work is disallowed) + * ``wants: advice`` -> ADVISE (recommendation only, no side effects) + * otherwise -> EXECUTE (authorized to act) + """ + context = request.context.lower() + if "policy: denied" in context: + return PostureVerdict( + posture=Posture.STOP, rationale="blocked by standing policy" + ) + if "wants: advice" in context: + return PostureVerdict( + posture=Posture.ADVISE, + rationale="user asked for a recommendation, not an action", + ) + return PostureVerdict( + posture=Posture.EXECUTE, scope=f"call {request.action} as requested" + ) + + +def create_governed_runner() -> Runner: + """Create an ADK runner whose tool calls pass through authority routing.""" + plugin = AuthorityRoutingPlugin(posture_router=demo_router) + + agent = Agent( + name="authority_routed_agent", + model="gemini-2.0-flash", + instruction="You are an assistant with an authorization layer.", + ) + + return Runner( + agent=agent, + plugins=[plugin], + app_name="authority-routing-demo", + session_service=InMemorySessionService(), + ) + + +# --------------------------------------------------------------------------- +# A live walkthrough of the decision layer (no model, no network). +# --------------------------------------------------------------------------- + + +@dataclass +class _Tool: + name: str + + +class _Ctx: + + def __init__(self, context: str = ""): + self.state = {"authority_context": context} + + +@dataclass +class _Case: + label: str + tool: str + args: dict + context: str + + +CASES = [ + _Case( + "authorized read — router EXECUTE, guard clear", + "web_search", + {"query": "weather in Paris tomorrow"}, + context="wants: action", + ), + _Case( + "advice requested — agent must NOT execute", + "update_ticket", + {"id": 42, "status": "closed"}, + context="wants: advice on whether to close ticket 42", + ), + _Case( + "irreversible call — guard overrides EXECUTE", + "delete_records", + {"table": "customers", "where": "all"}, + context="wants: action", + ), + _Case( + "approval pending — guard floors to DEFER", + "send_email", + {"to": "vip@corp.com", "body": "the numbers"}, + context="wants: action; note: awaiting approval from finance", + ), + _Case( + "disallowed — STOP survives strictest-wins", + "read_vault", + {"path": "prod credential store"}, + context="wants: action; policy: denied for credential stores", + ), +] + + +async def walkthrough() -> None: + plugin = AuthorityRoutingPlugin(posture_router=demo_router) + print(f"{'CASE':<46}{'TOOL':<17}{'POSTURE':<9}OUTCOME") + print("-" * 96) + for case in CASES: + decision = await plugin.before_tool_callback( + tool=_Tool(case.tool), + tool_args=case.args, + tool_context=_Ctx(case.context), + ) + if decision is None: + posture, outcome = "EXECUTE", "tool runs" + else: + posture = decision["posture"] + outcome = f"blocked — {decision['reason']}" + print(f"{case.label:<46}{case.tool:<17}{posture:<9}{outcome}") + + +if __name__ == "__main__": + runner = create_governed_runner() + print("Runner created with authority-routing plugin enabled.\n") + asyncio.run(walkthrough()) diff --git a/src/google/adk_community/plugins/__init__.py b/src/google/adk_community/plugins/__init__.py index 2e4b2ee..36bc2b4 100644 --- a/src/google/adk_community/plugins/__init__.py +++ b/src/google/adk_community/plugins/__init__.py @@ -12,21 +12,27 @@ # See the License for the specific language governing permissions and # limitations under the License. -from google.adk_community.plugins.agent_governance_plugin import ( - AgentGovernancePlugin, -) -from google.adk_community.plugins.taxonomy import ( - DefaultSkillPolicy, - SkillPolicy, - TaxonomyPipeline, - TaxonomyPlugin, - TaxonomyRegistry, - TaxonomyResolver, - TaxonomyTerm, -) +from google.adk_community.plugins.agent_governance_plugin import AgentGovernancePlugin +from google.adk_community.plugins.authority_routing_plugin import AuthorityRequest +from google.adk_community.plugins.authority_routing_plugin import AuthorityRoutingPlugin +from google.adk_community.plugins.authority_routing_plugin import Posture +from google.adk_community.plugins.authority_routing_plugin import PostureVerdict +from google.adk_community.plugins.authority_routing_plugin import stricter +from google.adk_community.plugins.taxonomy import DefaultSkillPolicy +from google.adk_community.plugins.taxonomy import SkillPolicy +from google.adk_community.plugins.taxonomy import TaxonomyPipeline +from google.adk_community.plugins.taxonomy import TaxonomyPlugin +from google.adk_community.plugins.taxonomy import TaxonomyRegistry +from google.adk_community.plugins.taxonomy import TaxonomyResolver +from google.adk_community.plugins.taxonomy import TaxonomyTerm __all__ = [ "AgentGovernancePlugin", + "AuthorityRequest", + "AuthorityRoutingPlugin", + "Posture", + "PostureVerdict", + "stricter", "DefaultSkillPolicy", "SkillPolicy", "TaxonomyPipeline", diff --git a/src/google/adk_community/plugins/authority_routing_plugin.py b/src/google/adk_community/plugins/authority_routing_plugin.py new file mode 100644 index 0000000..0d5049c --- /dev/null +++ b/src/google/adk_community/plugins/authority_routing_plugin.py @@ -0,0 +1,423 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""ADK plugin for authority-posture routing (ADVISE / EXECUTE / DEFER / STOP). + +Tool schemas and permissions govern *access* — which tools exist and who may +call them. This plugin governs *authorization* — whether the agent was allowed +to act at all before any tool runs. That is the layer where "the user asked for +advice and the agent executed" and "one bounded edit became a dependent +workflow" failures actually live. + +Every proposed tool call is classified into one of four postures: + +======== ================================================================= +Posture Meaning +======== ================================================================= +ADVISE Produce recommendations only; no tool side effects are authorized. +EXECUTE Act, within an explicitly bounded scope. +DEFER Pause and request authorization when scope or approval is unclear. +STOP Refuse or block when the requested work is not allowed. +======== ================================================================= + +Two stages decide the final posture: + +1. A **posture router** (optional, caller-supplied ``Callable``) — typically a + model call — that classifies the request into a posture with a scope + statement. Its verdict is *advisory*. +2. A **deterministic guard** — an irreversibility keyword tripwire plus an + approval-state check — that composes with the router under + **most-restrictive-wins**. Plain code can only make the posture *stricter*, + never looser: if the router says EXECUTE but the guard trips, the floor is + DEFER; if the router says STOP and the guard finds nothing, it stays STOP. + +The design is **fail-closed**: an unparseable or raised router verdict defaults +to DEFER, because an unreadable authorization is not an authorization. With no +router configured, the plugin runs guard-only — a fully deterministic, +dependency-free authorization layer that DEFERs irreversible or approval-pending +tool calls and lets everything else EXECUTE. + +Requires: no extra dependencies. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import enum +import logging +from typing import Any +from typing import Callable +from typing import Optional +from typing import Sequence +from typing import Union + +from google.adk.plugins.base_plugin import BasePlugin +from google.adk.tools.base_tool import BaseTool +from google.adk.tools.tool_context import ToolContext + +logger = logging.getLogger("google_adk_community." + __name__) + + +class Posture(str, enum.Enum): + """The four authority postures, ordered loosest to strictest.""" + + ADVISE = "ADVISE" + EXECUTE = "EXECUTE" + DEFER = "DEFER" + STOP = "STOP" + + +# Strictness ladder: a guard may only move a posture UP this ladder, never down. +# EXECUTE is the *loosest* (act freely within scope); STOP is the strictest. +_STRICTNESS: dict[Posture, int] = { + Posture.EXECUTE: 0, + Posture.ADVISE: 1, + Posture.DEFER: 2, + Posture.STOP: 3, +} + + +def stricter(a: Posture, b: Posture) -> Posture: + """Most-restrictive-wins composition: returns the stricter of two postures.""" + return a if _STRICTNESS[a] >= _STRICTNESS[b] else b + + +@dataclass +class PostureVerdict: + """A validated authority decision. + + Attributes: + posture: The chosen posture. + scope: For EXECUTE, the explicit boundary of the authorized action; + otherwise a short statement of why this posture was chosen. An EXECUTE + verdict without a scope is how bounded edits balloon into workflows. + rationale: Free-text explanation, useful for audit trails. + """ + + posture: Posture + scope: str = "" + rationale: str = "" + + +@dataclass +class AuthorityRequest: + """The unit a posture router is asked to authorize: one proposed tool call. + + Attributes: + action: The tool name being proposed. + tool_args: The arguments the agent wants to call the tool with. + context: Standing context (pending approvals, prior refusals, scope of the + current task) pulled from session state. The guard scans this for + approval-state markers. + """ + + action: str + tool_args: dict[str, Any] + context: str + + +# A router maps a request to a posture. It may return a full PostureVerdict, a +# bare Posture, or a string naming a posture ("EXECUTE"); anything else is +# treated as unparseable and fails closed. +PostureRouter = Callable[ + [AuthorityRequest], Union[PostureVerdict, Posture, str] +] + + +# The tripwire is deliberately blunt. A false DEFER costs a clarifying question; +# a false EXECUTE can cost an irreversible action. Bias toward the question. +DEFAULT_IRREVERSIBLE_KEYWORDS: tuple[str, ...] = ( + "delete", + "drop", + "wipe", + "truncate", + "rm -", + "overwrite", + "purge", + "send", + "email", + "publish", + "post", + "deploy", + "merge", + "push", + "release", + "pay", + "refund", + "charge", + "transfer", + "wire", + "password", + "secret", + "token", + "credential", + "api key", + ".env", + "revoke", + "uninstall", + "reset", +) + +DEFAULT_PENDING_APPROVAL_MARKERS: tuple[str, ...] = ( + "pending approval", + "awaiting approval", + "not yet approved", + "approval required", + "do not proceed", + "don't do this yet", + "hold off", + "waiting for sign-off", +) + + +class AuthorityRoutingPlugin(BasePlugin): + """ADK plugin that enforces an authority posture before tool execution. + + On every tool call the plugin computes a final posture by composing an + optional caller-supplied posture router with a deterministic guard under + most-restrictive-wins. When the final posture is anything other than EXECUTE, + the call is short-circuited with a structured block response (per the ADK + plugin contract); an EXECUTE posture returns ``None`` and the tool proceeds. + + Args: + posture_router: Optional callable classifying an ``AuthorityRequest`` into a + posture (typically a model call). May return a ``PostureVerdict``, a + ``Posture``, or a posture name as a string. If omitted, the plugin runs + guard-only and every non-tripped call is treated as EXECUTE. + context_state_key: Session-state key whose string value is treated as the + standing context scanned for approval-state markers. Defaults to + ``"authority_context"``. + irreversible_keywords: Override the tripwire keyword list. The tool name and + argument values are scanned (case-insensitively) for these substrings; a + hit forces the floor to DEFER. + pending_approval_markers: Override the approval-state marker list scanned in + the context string; a hit forces the floor to DEFER. + fail_closed: When ``True`` (default), a router that raises or returns an + unparseable verdict yields DEFER. When ``False``, such a router is ignored + (posture EXECUTE) and a warning is logged. + name: Plugin name registered with ADK. Defaults to ``"authority_routing"``. + + Example:: + + from google.adk_community.plugins import AuthorityRoutingPlugin + + # Guard-only: no model, fully deterministic. + plugin = AuthorityRoutingPlugin() + runner = Runner(agent=my_agent, plugins=[plugin], ...) + + # With a model-backed router: + def route(req): + verdict = my_model_call(req.action, req.tool_args, req.context) + return PostureVerdict(posture=..., scope=...) + + plugin = AuthorityRoutingPlugin(posture_router=route) + """ + + def __init__( + self, + *, + posture_router: Optional[PostureRouter] = None, + context_state_key: str = "authority_context", + irreversible_keywords: Optional[Sequence[str]] = None, + pending_approval_markers: Optional[Sequence[str]] = None, + fail_closed: bool = True, + name: str = "authority_routing", + ) -> None: + super().__init__(name=name) + self._posture_router = posture_router + self._context_state_key = context_state_key + self._irreversible_keywords = tuple( + kw.lower() + for kw in ( + irreversible_keywords + if irreversible_keywords is not None + else DEFAULT_IRREVERSIBLE_KEYWORDS + ) + ) + self._pending_markers = tuple( + m.lower() + for m in ( + pending_approval_markers + if pending_approval_markers is not None + else DEFAULT_PENDING_APPROVAL_MARKERS + ) + ) + self._fail_closed = fail_closed + + # -- verdict coercion -------------------------------------------------------- + + def _coerce_verdict(self, raw: Any) -> Optional[PostureVerdict]: + """Normalize a router return value into a PostureVerdict, or None if bad.""" + if isinstance(raw, PostureVerdict): + if isinstance(raw.posture, Posture): + return raw + # A PostureVerdict carrying a stringy posture — coerce the inner value. + coerced = self._coerce_posture(raw.posture) + if coerced is None: + return None + return PostureVerdict( + posture=coerced, scope=raw.scope, rationale=raw.rationale + ) + if isinstance(raw, Posture): + return PostureVerdict(posture=raw) + if isinstance(raw, str): + coerced = self._coerce_posture(raw) + if coerced is None: + return None + return PostureVerdict(posture=coerced) + return None + + @staticmethod + def _coerce_posture(value: Any) -> Optional[Posture]: + if isinstance(value, Posture): + return value + if isinstance(value, str): + try: + return Posture(value.strip().upper()) + except ValueError: + return None + return None + + # -- deterministic guard ----------------------------------------------------- + + def _deterministic_floor( + self, action_probe: str, context: str + ) -> tuple[Posture, str]: + """Return the strictest posture the guard requires, with the reason. + + EXECUTE means "no floor raised". The guard never returns ADVISE or STOP; + those are the router's judgment calls, not deterministic ones. + """ + low_ctx = context.lower() + for marker in self._pending_markers: + if marker in low_ctx: + return Posture.DEFER, f"approval-state marker in context: '{marker}'" + for kw in self._irreversible_keywords: + if kw in action_probe: + return Posture.DEFER, f"irreversibility tripwire: '{kw}'" + return Posture.EXECUTE, "no deterministic concerns" + + # -- ADK lifecycle ----------------------------------------------------------- + + async def before_tool_callback( + self, + *, + tool: BaseTool, + tool_args: dict[str, Any], + tool_context: ToolContext, + ) -> Optional[dict]: + """Decide the authority posture for a tool call before it runs. + + Returns ``None`` to let the tool proceed (posture EXECUTE), or a structured + block dict to short-circuit execution (posture ADVISE / DEFER / STOP). + """ + context = self._read_context(tool_context) + request = AuthorityRequest( + action=tool.name, tool_args=tool_args, context=context + ) + + # 1. Router verdict (advisory). Fail closed on error / unparseable output. + verdict = self._route(request) + + # 2. Deterministic guard floor. + action_probe = self._action_probe(tool.name, tool_args) + floor, floor_reason = self._deterministic_floor(action_probe, context) + + # 3. Compose: strictest wins. + final = stricter(verdict.posture, floor) + + if final is Posture.EXECUTE: + return None + + reason = self._explain(final, verdict, floor, floor_reason) + logger.warning( + "[%s] Blocked tool '%s': posture=%s (%s)", + self.name, + tool.name, + final.value, + reason, + ) + return { + "error": "authority_blocked", + "posture": final.value, + "reason": reason, + "scope": verdict.scope, + } + + # -- helpers ----------------------------------------------------------------- + + def _read_context(self, tool_context: ToolContext) -> str: + state = getattr(tool_context, "state", None) + if state is None: + return "" + try: + value = state.get(self._context_state_key, "") + except Exception: # pragma: no cover - defensive against exotic state objs + return "" + return value if isinstance(value, str) else str(value) + + @staticmethod + def _action_probe(tool_name: str, tool_args: dict[str, Any]) -> str: + """Flatten the proposed call into one lowercase string for tripwire scans.""" + parts = [tool_name] + parts.extend(str(v) for v in tool_args.values()) + return " ".join(parts).lower() + + def _route(self, request: AuthorityRequest) -> PostureVerdict: + """Run the router with fail-closed handling; EXECUTE if no router.""" + if self._posture_router is None: + return PostureVerdict( + posture=Posture.EXECUTE, scope="", rationale="no router" + ) + + try: + raw = self._posture_router(request) + except Exception as exc: # noqa: BLE001 - any router failure must fail closed + return self._router_fallback(f"router raised: {exc!r}") + + verdict = self._coerce_verdict(raw) + if verdict is None: + return self._router_fallback(f"router output unparseable: {raw!r}") + return verdict + + def _router_fallback(self, reason: str) -> PostureVerdict: + if self._fail_closed: + logger.warning("[%s] Failing closed to DEFER: %s", self.name, reason) + return PostureVerdict(posture=Posture.DEFER, scope="", rationale=reason) + logger.warning( + "[%s] Router ignored (fail_closed=False), posture EXECUTE: %s", + self.name, + reason, + ) + return PostureVerdict(posture=Posture.EXECUTE, scope="", rationale=reason) + + @staticmethod + def _explain( + final: Posture, + verdict: PostureVerdict, + floor: Posture, + floor_reason: str, + ) -> str: + if final is floor and floor is not verdict.posture: + # The guard raised the posture above the router's verdict. + return f"deterministic guard: {floor_reason}" + if verdict.rationale: + return verdict.rationale + if verdict.scope: + return verdict.scope + return { + Posture.ADVISE: "request authorized advice only, not tool execution", + Posture.DEFER: "authorization ambiguous or approval pending", + Posture.STOP: "requested action is not permitted", + }.get(final, floor_reason) diff --git a/tests/plugins/test_authority_routing_plugin.py b/tests/plugins/test_authority_routing_plugin.py new file mode 100644 index 0000000..1a1abc3 --- /dev/null +++ b/tests/plugins/test_authority_routing_plugin.py @@ -0,0 +1,424 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for AuthorityRoutingPlugin. + +These tests are fully deterministic — no model calls. The optional posture +router is exercised with plain Python callables, so the suite runs with no +credentials and no network. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from google.adk_community.plugins.authority_routing_plugin import AuthorityRequest +from google.adk_community.plugins.authority_routing_plugin import AuthorityRoutingPlugin +from google.adk_community.plugins.authority_routing_plugin import Posture +from google.adk_community.plugins.authority_routing_plugin import PostureVerdict +from google.adk_community.plugins.authority_routing_plugin import stricter + +# --------------------------------------------------------------------------- +# Fake ADK types for testing +# --------------------------------------------------------------------------- + + +class _FakeBaseTool: + + def __init__(self, name: str): + self.name = name + + +class _FakeToolContext: + + def __init__(self, state: dict[str, Any] | None = None): + self.state = state or {} + + +def _make_router(*, posture: Posture, scope: str = "", rationale: str = ""): + def router(_req: AuthorityRequest) -> PostureVerdict: + return PostureVerdict(posture=posture, scope=scope, rationale=rationale) + + return router + + +async def _decide( + plugin: AuthorityRoutingPlugin, tool_name, args=None, state=None +): + return await plugin.before_tool_callback( + tool=_FakeBaseTool(tool_name), + tool_args=args or {}, + tool_context=_FakeToolContext(state), + ) + + +# --------------------------------------------------------------------------- +# Posture ladder +# --------------------------------------------------------------------------- + + +class TestStrictness: + + def test_stricter_picks_higher_rung(self): + assert stricter(Posture.EXECUTE, Posture.DEFER) is Posture.DEFER + assert stricter(Posture.STOP, Posture.EXECUTE) is Posture.STOP + assert stricter(Posture.ADVISE, Posture.DEFER) is Posture.DEFER + + def test_stricter_is_symmetric(self): + assert stricter(Posture.DEFER, Posture.ADVISE) is stricter( + Posture.ADVISE, Posture.DEFER + ) + + def test_full_ladder_order(self): + ladder = [Posture.EXECUTE, Posture.ADVISE, Posture.DEFER, Posture.STOP] + for i in range(len(ladder)): + for j in range(len(ladder)): + expected = ladder[max(i, j)] + assert stricter(ladder[i], ladder[j]) is expected + + +# --------------------------------------------------------------------------- +# Construction +# --------------------------------------------------------------------------- + + +class TestConstruction: + + def test_default_name(self): + assert AuthorityRoutingPlugin().name == "authority_routing" + + def test_custom_name(self): + assert AuthorityRoutingPlugin(name="authz").name == "authz" + + def test_no_dependencies_required(self): + # Constructs with zero configuration and no optional deps installed. + plugin = AuthorityRoutingPlugin() + assert plugin is not None + + +# --------------------------------------------------------------------------- +# Guard-only mode (no router) +# --------------------------------------------------------------------------- + + +class TestGuardOnly: + + @pytest.mark.asyncio + async def test_clean_call_executes(self): + plugin = AuthorityRoutingPlugin() + result = await _decide(plugin, "web_search", {"query": "weather"}) + assert result is None + + @pytest.mark.asyncio + async def test_irreversible_tool_name_defers(self): + plugin = AuthorityRoutingPlugin() + result = await _decide(plugin, "delete_record", {"id": 7}) + assert result is not None + assert result["error"] == "authority_blocked" + assert result["posture"] == "DEFER" + assert "tripwire" in result["reason"] + + @pytest.mark.asyncio + async def test_irreversible_arg_value_defers(self): + plugin = AuthorityRoutingPlugin() + # Benign tool name, but the argument reveals a destructive command. + result = await _decide(plugin, "run_shell", {"cmd": "rm -rf build"}) + assert result is not None + assert result["posture"] == "DEFER" + assert "rm -" in result["reason"] + + @pytest.mark.asyncio + async def test_approval_marker_in_context_defers(self): + plugin = AuthorityRoutingPlugin() + result = await _decide( + plugin, + "write_file", + {"path": "notes.txt"}, + state={"authority_context": "This task is pending approval from ops."}, + ) + assert result is not None + assert result["posture"] == "DEFER" + assert "approval-state marker" in result["reason"] + + @pytest.mark.asyncio + async def test_approval_marker_takes_precedence_over_tripwire(self): + # Both fire; the approval marker is checked first, but either way DEFER. + plugin = AuthorityRoutingPlugin() + result = await _decide( + plugin, + "delete_record", + {"id": 1}, + state={"authority_context": "hold off — awaiting approval"}, + ) + assert result["posture"] == "DEFER" + assert "approval-state marker" in result["reason"] + + +# --------------------------------------------------------------------------- +# Router mode +# --------------------------------------------------------------------------- + + +class TestRouter: + + @pytest.mark.asyncio + async def test_router_execute_allows(self): + plugin = AuthorityRoutingPlugin( + posture_router=_make_router(posture=Posture.EXECUTE, scope="one row") + ) + result = await _decide(plugin, "update_row", {"id": 2}) + assert result is None + + @pytest.mark.asyncio + async def test_router_advise_blocks(self): + plugin = AuthorityRoutingPlugin( + posture_router=_make_router( + posture=Posture.ADVISE, rationale="user asked for a review" + ) + ) + result = await _decide(plugin, "update_row", {"id": 2}) + assert result is not None + assert result["posture"] == "ADVISE" + assert result["reason"] == "user asked for a review" + + @pytest.mark.asyncio + async def test_router_stop_blocks(self): + plugin = AuthorityRoutingPlugin( + posture_router=_make_router(posture=Posture.STOP) + ) + result = await _decide(plugin, "read_public_doc", {}) + assert result is not None + assert result["posture"] == "STOP" + + @pytest.mark.asyncio + async def test_router_scope_surfaced_when_no_rationale(self): + plugin = AuthorityRoutingPlugin( + posture_router=_make_router( + posture=Posture.DEFER, scope="scope unclear" + ) + ) + result = await _decide(plugin, "read_doc", {}) + assert result["scope"] == "scope unclear" + + +# --------------------------------------------------------------------------- +# Most-restrictive-wins composition +# --------------------------------------------------------------------------- + + +class TestComposition: + + @pytest.mark.asyncio + async def test_guard_overrides_router_execute(self): + # Router says EXECUTE, but the tool call is irreversible -> DEFER. + plugin = AuthorityRoutingPlugin( + posture_router=_make_router(posture=Posture.EXECUTE, scope="one file") + ) + result = await _decide(plugin, "delete_file", {"path": "prod.db"}) + assert result is not None + assert result["posture"] == "DEFER" + assert "deterministic guard" in result["reason"] + + @pytest.mark.asyncio + async def test_router_stop_survives_clean_guard(self): + # Router says STOP; guard finds nothing. The guard never loosens -> STOP. + plugin = AuthorityRoutingPlugin( + posture_router=_make_router(posture=Posture.STOP) + ) + result = await _decide(plugin, "web_search", {"query": "hi"}) + assert result["posture"] == "STOP" + + @pytest.mark.asyncio + async def test_router_advise_survives_clean_guard(self): + plugin = AuthorityRoutingPlugin( + posture_router=_make_router(posture=Posture.ADVISE) + ) + result = await _decide(plugin, "web_search", {"query": "hi"}) + assert result["posture"] == "ADVISE" + + +# --------------------------------------------------------------------------- +# Fail-closed behavior +# --------------------------------------------------------------------------- + + +class TestFailClosed: + + @pytest.mark.asyncio + async def test_router_raises_defaults_to_defer(self): + def boom(_req): + raise RuntimeError("model timeout") + + plugin = AuthorityRoutingPlugin(posture_router=boom) + result = await _decide(plugin, "web_search", {}) + assert result is not None + assert result["posture"] == "DEFER" + + @pytest.mark.asyncio + async def test_router_unparseable_defaults_to_defer(self): + plugin = AuthorityRoutingPlugin(posture_router=lambda _req: 42) + result = await _decide(plugin, "web_search", {}) + assert result["posture"] == "DEFER" + + @pytest.mark.asyncio + async def test_router_bad_string_defaults_to_defer(self): + plugin = AuthorityRoutingPlugin(posture_router=lambda _req: "MAYBE") + result = await _decide(plugin, "web_search", {}) + assert result["posture"] == "DEFER" + + @pytest.mark.asyncio + async def test_fail_open_ignores_bad_router(self): + def boom(_req): + raise RuntimeError("model timeout") + + plugin = AuthorityRoutingPlugin(posture_router=boom, fail_closed=False) + # Router ignored -> EXECUTE, and a clean guard -> allow. + result = await _decide(plugin, "web_search", {}) + assert result is None + + @pytest.mark.asyncio + async def test_fail_open_still_respects_guard(self): + def boom(_req): + raise RuntimeError("model timeout") + + plugin = AuthorityRoutingPlugin(posture_router=boom, fail_closed=False) + # Router ignored, but the guard still trips on an irreversible call. + result = await _decide(plugin, "delete_file", {"path": "x"}) + assert result["posture"] == "DEFER" + + +# --------------------------------------------------------------------------- +# Verdict coercion +# --------------------------------------------------------------------------- + + +class TestVerdictCoercion: + + @pytest.mark.asyncio + async def test_router_returns_bare_posture(self): + plugin = AuthorityRoutingPlugin(posture_router=lambda _req: Posture.ADVISE) + result = await _decide(plugin, "web_search", {}) + assert result["posture"] == "ADVISE" + + @pytest.mark.asyncio + async def test_router_returns_posture_string(self): + plugin = AuthorityRoutingPlugin(posture_router=lambda _req: "stop") + result = await _decide(plugin, "web_search", {}) + assert result["posture"] == "STOP" + + @pytest.mark.asyncio + async def test_router_returns_verdict_with_string_posture(self): + # A hand-built verdict whose posture slipped through as a string. + plugin = AuthorityRoutingPlugin( + posture_router=lambda _req: PostureVerdict( + posture="defer", scope="s" # type: ignore[arg-type] + ) + ) + result = await _decide(plugin, "web_search", {}) + assert result["posture"] == "DEFER" + assert result["scope"] == "s" + + +# --------------------------------------------------------------------------- +# Customization +# --------------------------------------------------------------------------- + + +class TestCustomization: + + @pytest.mark.asyncio + async def test_custom_irreversible_keywords(self): + plugin = AuthorityRoutingPlugin(irreversible_keywords=["yeet"]) + # Default keyword no longer trips... + assert await _decide(plugin, "delete_file", {"path": "x"}) is None + # ...but the custom one does. + result = await _decide(plugin, "yeet_file", {"path": "x"}) + assert result["posture"] == "DEFER" + + @pytest.mark.asyncio + async def test_custom_approval_markers(self): + plugin = AuthorityRoutingPlugin(pending_approval_markers=["frozen"]) + result = await _decide( + plugin, + "web_search", + {}, + state={"authority_context": "this workflow is frozen"}, + ) + assert result["posture"] == "DEFER" + + @pytest.mark.asyncio + async def test_custom_context_state_key(self): + plugin = AuthorityRoutingPlugin(context_state_key="approvals") + result = await _decide( + plugin, + "web_search", + {}, + state={"approvals": "approval required"}, + ) + assert result["posture"] == "DEFER" + + @pytest.mark.asyncio + async def test_router_receives_request_fields(self): + seen = {} + + def router(req: AuthorityRequest): + seen["action"] = req.action + seen["args"] = req.tool_args + seen["context"] = req.context + return Posture.EXECUTE + + plugin = AuthorityRoutingPlugin(posture_router=router) + await _decide( + plugin, "update_row", {"id": 9}, state={"authority_context": "ctx"} + ) + assert seen == {"action": "update_row", "args": {"id": 9}, "context": "ctx"} + + +# --------------------------------------------------------------------------- +# Block response shape +# --------------------------------------------------------------------------- + + +class TestBlockShape: + + @pytest.mark.asyncio + async def test_block_dict_has_all_fields(self): + plugin = AuthorityRoutingPlugin( + posture_router=_make_router( + posture=Posture.STOP, scope="sc", rationale="ra" + ) + ) + result = await _decide(plugin, "web_search", {}) + assert set(result) == {"error", "posture", "reason", "scope"} + assert result["error"] == "authority_blocked" + assert result["posture"] == "STOP" + assert result["reason"] == "ra" + assert result["scope"] == "sc" + + @pytest.mark.asyncio + async def test_missing_state_is_safe(self): + # A tool context without a usable state must not raise. + plugin = AuthorityRoutingPlugin() + + class _NoState: + state = None + + result = await plugin.before_tool_callback( + tool=_FakeBaseTool("web_search"), + tool_args={}, + tool_context=_NoState(), + ) + assert result is None