diff --git a/contributing/samples/authority_routing/README.md b/contributing/samples/authority_routing/README.md new file mode 100644 index 0000000..f9c69b0 --- /dev/null +++ b/contributing/samples/authority_routing/README.md @@ -0,0 +1,32 @@ +# Authority Routing sample + +Demonstrates `AuthorityRoutingPlugin`: it classifies the **authority** of each +user turn as `ADVISE` / `EXECUTE` / `DEFER` / `STOP` and enforces that posture +on every tool call. Only `EXECUTE` authorizes side effects; a deterministic +keyword guard can only ever *tighten* the model router's verdict (strictest +posture wins), and the plugin fails **closed**. + +## Run + +```bash +python contributing/samples/authority_routing/main.py +``` + +Expected output: + +``` +[ADVISE ] DENIED edit_file <- Review the retry settings in config.yaml. +[EXECUTE] ALLOWED edit_file <- Rename foo to bar in utils.py. +[DEFER ] DENIED shell <- Deploy the release to production. +``` + +The third case shows the guard at work: even if the router had returned +`EXECUTE`, the word "deploy" trips the irreversibility tripwire and the floor +is raised to `DEFER`. + +## Wiring a real model router + +`main.py` uses a trivial keyword `CallableRouter` so the sample runs with no API +key. For real use, implement `PostureRouter.route()` with a model call (a Gemini +sketch is at the bottom of `main.py`) and pass it as `router=`. Parse the model +output against a schema and fail closed (return `DEFER`) if it is unparseable. diff --git a/contributing/samples/authority_routing/main.py b/contributing/samples/authority_routing/main.py new file mode 100644 index 0000000..46391a1 --- /dev/null +++ b/contributing/samples/authority_routing/main.py @@ -0,0 +1,139 @@ +# 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. + +"""Runnable example for AuthorityRoutingPlugin. + +Shows the plugin composing a model router with the deterministic guard without +needing a live model key: the router here is a tiny keyword classifier wrapped +in ``CallableRouter``. Swap it for a Gemini-backed router (sketch at the bottom) +to get real authority classification. + +Run:: + + python contributing/samples/authority_routing/main.py +""" + +from __future__ import annotations + +import asyncio + +from google.adk_community.plugins import ( + AuthorityRoutingPlugin, + CallableRouter, + Posture, + PostureVerdict, +) + + +def keyword_router(request: str, context: str) -> PostureVerdict: + """A trivial stand-in router (no model call). + + A real deployment replaces this with a model that reads the request and the + standing context. The deterministic guard tightens whatever this returns. + """ + low = request.lower() + if any(w in low for w in ("what would", "how would", "review", "suggest")): + return PostureVerdict(Posture.ADVISE, "", "request reads as advisory") + return PostureVerdict(Posture.EXECUTE, "the named target", "action requested") + + +class _Tool: + def __init__(self, name: str) -> None: + self.name = name + + +class _Ctx: + def __init__(self, state: dict) -> None: + self.state = state + + +class _Session: + def __init__(self, state: dict) -> None: + self.state = state + + +class _Inv: + def __init__(self, state: dict) -> None: + self.session = _Session(state) + + +class _Part: + def __init__(self, text: str) -> None: + self.text = text + + +class _Content: + def __init__(self, text: str) -> None: + self.parts = [_Part(text)] + + +async def main() -> None: + plugin = AuthorityRoutingPlugin( + router=CallableRouter(keyword_router), + read_only_tool_names={"read_file"}, + ) + + cases = [ + ("Review the retry settings in config.yaml.", "edit_file"), + ("Rename foo to bar in utils.py.", "edit_file"), + ("Deploy the release to production.", "shell"), + ] + + for request, tool_name in cases: + state: dict = {} + await plugin.on_user_message_callback( + invocation_context=_Inv(state), + user_message=_Content(request), + ) + decision = state["authority:posture"] + result = await plugin.before_tool_callback( + tool=_Tool(tool_name), + tool_args={}, + tool_context=_Ctx(state), + ) + verdict = "ALLOWED" if result is None else "DENIED" + print(f"[{decision['posture']:7}] {verdict:7} {tool_name:10} <- {request}") + + +# --- Sketch: a Gemini-backed router (requires google-genai + an API key) ------ +# +# from google import genai +# from google.adk_community.plugins import PostureRouter, PostureVerdict, Posture +# +# class GeminiPostureRouter(PostureRouter): +# def __init__(self, model: str = "gemini-2.5-flash") -> None: +# self._client = genai.Client() +# self._model = model +# +# async def route(self, request: str, context: str) -> PostureVerdict: +# resp = self._client.models.generate_content( +# model=self._model, +# contents=f"STANDING CONTEXT:\n{context}\n\nUSER REQUEST:\n{request}", +# config={ +# "system_instruction": ( +# "Classify the authority of the request as ADVISE, EXECUTE, " +# "DEFER, or STOP. When torn, pick the stricter posture." +# ), +# "response_mime_type": "application/json", +# "response_schema": PostureVerdict, +# }, +# ) +# parsed = resp.parsed +# if parsed is None: # fail closed on unparseable output +# return PostureVerdict(Posture.DEFER, "", "router output unparseable") +# return parsed + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/google/adk_community/plugins/__init__.py b/src/google/adk_community/plugins/__init__.py index 2e4b2ee..4cd5821 100644 --- a/src/google/adk_community/plugins/__init__.py +++ b/src/google/adk_community/plugins/__init__.py @@ -15,6 +15,14 @@ from google.adk_community.plugins.agent_governance_plugin import ( AgentGovernancePlugin, ) +from google.adk_community.plugins.authority_routing_plugin import ( + AuthorityRoutingPlugin, + CallableRouter, + Posture, + PostureRouter, + PostureVerdict, + stricter, +) from google.adk_community.plugins.taxonomy import ( DefaultSkillPolicy, SkillPolicy, @@ -27,11 +35,17 @@ __all__ = [ "AgentGovernancePlugin", + "AuthorityRoutingPlugin", + "CallableRouter", "DefaultSkillPolicy", + "Posture", + "PostureRouter", + "PostureVerdict", "SkillPolicy", "TaxonomyPipeline", "TaxonomyPlugin", "TaxonomyRegistry", "TaxonomyResolver", "TaxonomyTerm", + "stricter", ] 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..33b014a --- /dev/null +++ b/src/google/adk_community/plugins/authority_routing_plugin.py @@ -0,0 +1,400 @@ +# 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-based tool routing (ADVISE / EXECUTE / DEFER / STOP). + +Tool schemas define what an agent *can* reach and permission systems define +what calls are *allowed*, but a large class of real agent failures happens +inside those boundaries: the user asked for advice and the agent executed; the +user authorized one bounded edit and the agent expanded into a dependent +workflow; an approval was clearly required and the agent proceeded anyway. + +This plugin classifies the *authority* of each user turn into one of four +postures and enforces it on every tool call: + +* ``ADVISE`` - the user wants a recommendation; no side effects authorized. +* ``EXECUTE`` - action authorized within an explicit, bounded scope. +* ``DEFER`` - authorization is ambiguous or a required approval is missing. +* ``STOP`` - the work is disallowed by policy, safety, or a prior refusal. + +It composes two layers under one rule -- **the strictest posture wins**: + +1. A pluggable model **router** (:class:`PostureRouter`) classifies the turn. + The router is best-effort and can be argued out of a posture; it can only + ever be *tightened* by the guard, never loosened. +2. A deterministic **guard** (plain keyword code) that a model cannot be talked + past. It raises an irreversibility / pending-approval floor the router + output is composed against. + +The design fails **closed**: an unparseable or errored routing decision, or a +tool call that runs before any turn was routed, denies side effects rather than +allowing them. + +Ported from a production authority-routing pattern, also proposed to the +Anthropic cookbook (anthropics/claude-cookbooks#787). The model router there is +Anthropic-backed; here it is a provider-agnostic ABC so any model (Gemini via +``google-genai``, or otherwise) can back it. +""" + +from __future__ import annotations + +import abc +import asyncio +import inspect +import logging +from collections import deque +from dataclasses import dataclass +from enum import Enum +from typing import Any, Awaitable, Callable, Optional, Sequence + +from google.adk.agents.invocation_context import InvocationContext +from google.adk.plugins.base_plugin import BasePlugin +from google.adk.tools.base_tool import BaseTool +from google.adk.tools.tool_context import ToolContext +from google.genai import types + +logger = logging.getLogger(__name__) + + +class Posture(str, Enum): + """Authority posture for a user turn.""" + + ADVISE = "ADVISE" + EXECUTE = "EXECUTE" + DEFER = "DEFER" + STOP = "STOP" + + +# Strictness ladder: a guard may only move the posture UP this ladder, never +# down. EXECUTE is the only posture that authorizes side effects. +_STRICTNESS = { + Posture.EXECUTE: 0, + Posture.ADVISE: 1, + Posture.DEFER: 2, + Posture.STOP: 3, +} + + +def stricter(a: Posture, b: Posture) -> Posture: + """Return the most-restrictive of two postures (most-restrictive-wins).""" + return a if _STRICTNESS[a] >= _STRICTNESS[b] else b + + +@dataclass +class PostureVerdict: + """A router's authority classification for a single user turn. + + Attributes: + posture: The classified authority posture. + scope: For ``EXECUTE``, the explicit boundary of the authorized action; + otherwise a short reason for the chosen posture. Advisory only -- + it is surfaced to the agent and logged, not string-matched. + rationale: Human-readable justification for the verdict. + """ + + posture: Posture + scope: str = "" + rationale: str = "" + + +# Default deterministic tripwires. Anything in a request that touches deletion, +# sending, payments, credentials, or deployment lowers the floor to DEFER even +# if the router said EXECUTE. +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", +) + +# Markers in the standing context that signal an approval is not yet resolved. +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 PostureRouter(abc.ABC): + """Pluggable authority router. + + Implement :meth:`route` with any model backend. Keep it best-effort: the + plugin composes the router output against a deterministic guard that can + only tighten it, and fails closed if :meth:`route` raises. + """ + + @abc.abstractmethod + async def route(self, request: str, context: str) -> PostureVerdict: + """Classify a user turn's authority. + + Args: + request: The user's request text for this turn. + context: The standing context (prior authorizations, approval + state) the plugin was configured to read. + + Returns: + A :class:`PostureVerdict`. Implementations that parse structured + model output should fail closed (return ``DEFER``) on parse errors + rather than raising, though a raised exception is also handled by + the plugin as a closed failure. + """ + + +class CallableRouter(PostureRouter): + """Adapt a plain callable into a :class:`PostureRouter`. + + The callable receives ``(request, context)`` and returns a + :class:`PostureVerdict`. Sync callables run in a thread executor so they + never block the event loop. + """ + + def __init__( + self, + fn: Callable[[str, str], PostureVerdict | Awaitable[PostureVerdict]], + ) -> None: + self._fn = fn + + async def route(self, request: str, context: str) -> PostureVerdict: + result = self._fn(request, context) + if inspect.isawaitable(result): + return await result + loop = asyncio.get_running_loop() + return await loop.run_in_executor(None, lambda: result) + + +class AuthorityRoutingPlugin(BasePlugin): + """Enforce a per-turn authority posture on every tool call. + + On each user message the plugin routes an authority posture (via the + optional model router, tightened by the deterministic guard) and stores it + in session state. Before every tool call it enforces that posture: only + ``EXECUTE`` authorizes side effects; ``ADVISE`` / ``DEFER`` / ``STOP`` + short-circuit the tool with a denial response (per the ADK plugin + contract), unless the tool is on the read-only allowlist. + + Args: + router: The authority router. If ``None``, the plugin runs in + guard-only mode: the base posture is ``EXECUTE`` and only the + deterministic floor can tighten it. A router is strongly + recommended for anything beyond keyword tripwires. + standing_context_key: Session-state key the plugin reads the standing + context from (prior authorizations, approval state). Missing or + non-string values are treated as empty context. + posture_state_key: Session-state key the plugin writes the routed + decision to and reads it back from at tool time. + read_only_tool_names: Tool names that gather information without side + effects. These remain permitted under ``ADVISE`` and ``DEFER`` (an + agent may still read to advise or to decide what to ask), but are + blocked under ``STOP``. + extra_irreversible_keywords: Additional request substrings that lower + the deterministic floor to ``DEFER``. + pending_approval_markers: Override the default context markers that + force ``DEFER``. ``None`` keeps the defaults. + fail_closed: If ``True`` (default), a routing error or a tool call that + runs before any turn was routed denies side effects. If ``False``, + such cases fall back to the guard-only base (``EXECUTE``). + audit_log_size: Max entries kept in the in-memory audit log (FIFO). + + Example:: + + from google.adk_community.plugins import ( + AuthorityRoutingPlugin, CallableRouter, PostureVerdict, Posture, + ) + + def my_router(request, context): + # call Gemini (or any model) and parse a PostureVerdict + ... + return PostureVerdict(posture=Posture.ADVISE, scope="", rationale="...") + + plugin = AuthorityRoutingPlugin( + router=CallableRouter(my_router), + read_only_tool_names={"read_file", "list_dir"}, + ) + runner = Runner(agent=my_agent, plugins=[plugin], ...) + """ + + def __init__( + self, + router: Optional[PostureRouter] = None, + *, + standing_context_key: str = "authority:standing_context", + posture_state_key: str = "authority:posture", + read_only_tool_names: Optional[Sequence[str]] = None, + extra_irreversible_keywords: Optional[Sequence[str]] = None, + pending_approval_markers: Optional[Sequence[str]] = None, + fail_closed: bool = True, + audit_log_size: int = 1000, + ) -> None: + super().__init__(name="authority_routing") + self._router = router + self._standing_context_key = standing_context_key + self._posture_state_key = posture_state_key + self._read_only = set(read_only_tool_names or ()) + self._irreversible = tuple(DEFAULT_IRREVERSIBLE_KEYWORDS) + tuple( + extra_irreversible_keywords or () + ) + self._pending_markers = tuple( + DEFAULT_PENDING_APPROVAL_MARKERS + if pending_approval_markers is None + else pending_approval_markers + ) + self._fail_closed = fail_closed + self._audit: deque[dict[str, Any]] = deque(maxlen=audit_log_size) + + # -- deterministic guard ------------------------------------------------- + + def deterministic_floor(self, request: str, context: str) -> tuple[Posture, str]: + """Return the strictest posture the guard requires, with the reason. + + Pending-approval markers in the context and irreversibility keywords in + the request each raise the floor to ``DEFER``. With no concern found the + floor is ``EXECUTE`` (no floor raised). + """ + low_req, low_ctx = request.lower(), 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: + if kw in low_req: + return Posture.DEFER, f"irreversibility tripwire: '{kw}'" + return Posture.EXECUTE, "no deterministic concerns" + + # -- routing on each user turn ------------------------------------------ + + async def on_user_message_callback( + self, + *, + invocation_context: InvocationContext, + user_message: types.Content, + ) -> Optional[types.Content]: + """Route and store the authority posture for this turn.""" + request = _extract_text(user_message) + state = invocation_context.session.state + raw_ctx = state.get(self._standing_context_key, "") + context = raw_ctx if isinstance(raw_ctx, str) else "" + + # Base posture from the router (best-effort), or EXECUTE in guard-only + # mode. A raised router is a closed failure. + if self._router is None: + verdict = PostureVerdict( + posture=Posture.EXECUTE, + scope="", + rationale="guard-only mode (no router configured)", + ) + else: + try: + verdict = await self._router.route(request, context) + except Exception as exc: # noqa: BLE001 - router is untrusted + floor_posture = Posture.DEFER if self._fail_closed else Posture.EXECUTE + verdict = PostureVerdict( + posture=floor_posture, + scope="", + rationale=f"router raised {type(exc).__name__}: {exc}", + ) + logger.warning( + "Authority router raised; failing %s: %s", + "closed" if self._fail_closed else "open", exc, + ) + + floor, floor_reason = self.deterministic_floor(request, context) + final = stricter(verdict.posture, floor) + + decision = { + "posture": final.value, + "scope": verdict.scope, + "router_posture": verdict.posture.value, + "router_rationale": verdict.rationale, + "floor": floor.value, + "floor_reason": floor_reason, + } + state[self._posture_state_key] = decision + self._audit.append({"event": "route", **decision}) + logger.info( + "Authority posture=%s (router=%s, floor=%s: %s)", + final.value, verdict.posture.value, floor.value, floor_reason, + ) + return None + + # -- enforcement before each tool call ---------------------------------- + + async def before_tool_callback( + self, + *, + tool: BaseTool, + tool_args: dict[str, Any], + tool_context: ToolContext, + ) -> Optional[dict]: + """Allow the tool under ``EXECUTE``; otherwise short-circuit it.""" + decision = tool_context.state.get(self._posture_state_key) + + if not isinstance(decision, dict) or "posture" not in decision: + # No turn was routed (e.g. tool fired before any user message was + # classified). Fail closed by default. + if self._fail_closed: + return self._deny( + tool, posture=Posture.DEFER.value, scope="", + reason="no authority decision on record for this turn", + ) + return None + + posture = decision["posture"] + scope = decision.get("scope", "") + + if posture == Posture.EXECUTE.value: + return None + + if posture in (Posture.ADVISE.value, Posture.DEFER.value) and tool.name in self._read_only: + self._audit.append( + {"event": "allow_read_only", "tool": tool.name, "posture": posture} + ) + return None + + return self._deny( + tool, posture=posture, scope=scope, + reason=decision.get("floor_reason") or decision.get("router_rationale", ""), + ) + + def _deny(self, tool: BaseTool, *, posture: str, scope: str, reason: str) -> dict: + self._audit.append( + {"event": "deny", "tool": tool.name, "posture": posture, "reason": reason} + ) + logger.warning( + "Authority denied tool '%s' under posture %s: %s", + tool.name, posture, reason, + ) + return { + "error": "authority_denied", + "posture": posture, + "scope": scope, + "reason": reason, + "message": ( + f"Tool '{tool.name}' was not executed: the authority router " + f"classified this turn as {posture}, which does not authorize " + f"side-effecting actions. Only EXECUTE authorizes tool side effects." + ), + } + + @property + def audit_log(self) -> list[dict[str, Any]]: + """A copy of the in-memory audit log (oldest first).""" + return list(self._audit) + + +def _extract_text(content: Optional[types.Content]) -> str: + """Join the text parts of a Content into a single string.""" + if content is None or not getattr(content, "parts", None): + return "" + texts = [p.text for p in content.parts if getattr(p, "text", None)] + return "\n".join(texts) diff --git a/tests/plugins/test_authority_routing_plugin.py b/tests/plugins/test_authority_routing_plugin.py new file mode 100644 index 0000000..e2352d7 --- /dev/null +++ b/tests/plugins/test_authority_routing_plugin.py @@ -0,0 +1,283 @@ +# 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. + +The plugin's ADK integration surface is small (a state dict on the invocation +context and on the tool context), so these tests use light fakes instead of a +full ADK runtime. The deterministic guard is tested directly; the model router +is stubbed, so no live model key is required. +""" + +from __future__ import annotations + +from typing import Any, Optional + +import pytest + +from google.adk_community.plugins.authority_routing_plugin import ( + AuthorityRoutingPlugin, + CallableRouter, + Posture, + PostureRouter, + PostureVerdict, + stricter, +) + + +# --------------------------------------------------------------------------- +# Fakes for the small slice of ADK the plugin touches +# --------------------------------------------------------------------------- + + +class _FakeTool: + def __init__(self, name: str) -> None: + self.name = name + + +class _FakeSession: + def __init__(self, state: dict[str, Any]) -> None: + self.state = state + + +class _FakeInvocationContext: + def __init__(self, state: dict[str, Any]) -> None: + self.session = _FakeSession(state) + + +class _FakeToolContext: + def __init__(self, state: dict[str, Any]) -> None: + self.state = state + + +class _FakePart: + def __init__(self, text: Optional[str]) -> None: + self.text = text + + +class _FakeContent: + def __init__(self, *texts: str) -> None: + self.parts = [_FakePart(t) for t in texts] + + +class _StubRouter(PostureRouter): + """A router that always returns a fixed verdict (or raises).""" + + def __init__( + self, verdict: Optional[PostureVerdict] = None, raises: bool = False + ) -> None: + self._verdict = verdict + self._raises = raises + + async def route(self, request: str, context: str) -> PostureVerdict: + if self._raises: + raise RuntimeError("boom") + assert self._verdict is not None + return self._verdict + + +async def _route( + plugin: AuthorityRoutingPlugin, state: dict, request: str +) -> None: + await plugin.on_user_message_callback( + invocation_context=_FakeInvocationContext(state), + user_message=_FakeContent(request), + ) + + +async def _tool(plugin: AuthorityRoutingPlugin, state: dict, name: str): + return await plugin.before_tool_callback( + tool=_FakeTool(name), + tool_args={}, + tool_context=_FakeToolContext(state), + ) + + +# --------------------------------------------------------------------------- +# Composition + guard (pure, no async) +# --------------------------------------------------------------------------- + + +def test_plugin_name() -> None: + assert AuthorityRoutingPlugin().name == "authority_routing" + + +def test_stricter_is_most_restrictive_wins() -> None: + assert stricter(Posture.EXECUTE, Posture.ADVISE) == Posture.ADVISE + assert stricter(Posture.ADVISE, Posture.DEFER) == Posture.DEFER + assert stricter(Posture.DEFER, Posture.STOP) == Posture.STOP + assert stricter(Posture.EXECUTE, Posture.EXECUTE) == Posture.EXECUTE + + +def test_floor_irreversible_keyword_forces_defer() -> None: + plugin = AuthorityRoutingPlugin() + posture, reason = plugin.deterministic_floor("please delete the table", "") + assert posture == Posture.DEFER + assert "delete" in reason + + +def test_floor_pending_marker_in_context_forces_defer() -> None: + plugin = AuthorityRoutingPlugin() + posture, reason = plugin.deterministic_floor( + "apply the change", "this is pending approval from the owner" + ) + assert posture == Posture.DEFER + assert "pending approval" in reason + + +def test_floor_clean_request_is_execute() -> None: + plugin = AuthorityRoutingPlugin() + posture, _ = plugin.deterministic_floor("rename this local variable", "") + assert posture == Posture.EXECUTE + + +def test_extra_irreversible_keywords_extend_the_guard() -> None: + plugin = AuthorityRoutingPlugin(extra_irreversible_keywords=["frobnicate"]) + posture, reason = plugin.deterministic_floor("frobnicate the widget", "") + assert posture == Posture.DEFER + assert "frobnicate" in reason + + +# --------------------------------------------------------------------------- +# Routing + enforcement +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_guard_only_mode_allows_clean_execute() -> None: + plugin = AuthorityRoutingPlugin() # no router + state: dict[str, Any] = {} + await _route(plugin, state, "rename a local variable") + assert state["authority:posture"]["posture"] == "EXECUTE" + assert await _tool(plugin, state, "edit_file") is None + + +@pytest.mark.asyncio +async def test_advise_denies_side_effecting_tool() -> None: + plugin = AuthorityRoutingPlugin( + router=_StubRouter( + PostureVerdict(Posture.ADVISE, "", "user wants advice") + ) + ) + state: dict[str, Any] = {} + await _route(plugin, state, "what would you change about the retry settings?") + result = await _tool(plugin, state, "edit_file") + assert result is not None + assert result["error"] == "authority_denied" + assert result["posture"] == "ADVISE" + + +@pytest.mark.asyncio +async def test_guard_tightens_execute_to_defer_on_irreversible() -> None: + # Router says EXECUTE, but the request text trips the irreversibility wire. + plugin = AuthorityRoutingPlugin( + router=_StubRouter(PostureVerdict(Posture.EXECUTE, "the config", "ok")) + ) + state: dict[str, Any] = {} + await _route(plugin, state, "deploy the config to production") + assert state["authority:posture"]["posture"] == "DEFER" + result = await _tool(plugin, state, "shell") + assert result is not None + assert result["posture"] == "DEFER" + + +@pytest.mark.asyncio +async def test_execute_clean_allows_tool() -> None: + plugin = AuthorityRoutingPlugin( + router=_StubRouter(PostureVerdict(Posture.EXECUTE, "one file", "bounded")) + ) + state: dict[str, Any] = {} + await _route(plugin, state, "rename foo to bar in utils") + assert await _tool(plugin, state, "edit_file") is None + + +@pytest.mark.asyncio +async def test_router_exception_fails_closed_to_defer() -> None: + plugin = AuthorityRoutingPlugin(router=_StubRouter(raises=True)) + state: dict[str, Any] = {} + await _route(plugin, state, "rename a variable") + assert state["authority:posture"]["posture"] == "DEFER" + assert (await _tool(plugin, state, "edit_file"))["error"] == "authority_denied" + + +@pytest.mark.asyncio +async def test_router_exception_fails_open_when_configured() -> None: + plugin = AuthorityRoutingPlugin( + router=_StubRouter(raises=True), fail_closed=False + ) + state: dict[str, Any] = {} + await _route(plugin, state, "rename a variable") + assert state["authority:posture"]["posture"] == "EXECUTE" + assert await _tool(plugin, state, "edit_file") is None + + +@pytest.mark.asyncio +async def test_read_only_tool_allowed_under_advise() -> None: + plugin = AuthorityRoutingPlugin( + router=_StubRouter(PostureVerdict(Posture.ADVISE, "", "advice")), + read_only_tool_names={"read_file"}, + ) + state: dict[str, Any] = {} + await _route(plugin, state, "review this module") + assert await _tool(plugin, state, "read_file") is None + assert (await _tool(plugin, state, "edit_file")) is not None + + +@pytest.mark.asyncio +async def test_read_only_tool_blocked_under_stop() -> None: + plugin = AuthorityRoutingPlugin( + router=_StubRouter(PostureVerdict(Posture.STOP, "", "disallowed")), + read_only_tool_names={"read_file"}, + ) + state: dict[str, Any] = {} + await _route(plugin, state, "do the disallowed thing") + assert (await _tool(plugin, state, "read_file")) is not None + + +@pytest.mark.asyncio +async def test_tool_without_routed_decision_fails_closed() -> None: + plugin = AuthorityRoutingPlugin() + result = await _tool(plugin, {}, "edit_file") + assert result is not None + assert result["error"] == "authority_denied" + + +@pytest.mark.asyncio +async def test_tool_without_routed_decision_allowed_when_open() -> None: + plugin = AuthorityRoutingPlugin(fail_closed=False) + assert await _tool(plugin, {}, "edit_file") is None + + +@pytest.mark.asyncio +async def test_callable_router_wraps_sync_callable() -> None: + def sync_router(request: str, context: str) -> PostureVerdict: + return PostureVerdict(Posture.ADVISE, "", "sync") + + plugin = AuthorityRoutingPlugin(router=CallableRouter(sync_router)) + state: dict[str, Any] = {} + await _route(plugin, state, "review this") + assert state["authority:posture"]["posture"] == "ADVISE" + + +@pytest.mark.asyncio +async def test_audit_log_records_route_and_deny() -> None: + plugin = AuthorityRoutingPlugin( + router=_StubRouter(PostureVerdict(Posture.ADVISE, "", "advice")) + ) + state: dict[str, Any] = {} + await _route(plugin, state, "review this") + await _tool(plugin, state, "edit_file") + events = [e["event"] for e in plugin.audit_log] + assert "route" in events + assert "deny" in events