Fix critical governance bypass and LLM classifier gaps

CHANGELOG.md

@@ -23,6 +23,58 @@ in beta until the v1.0 stability criteria documented in
   documented in [docs/telemetry.md](docs/telemetry.md), and all payload
   construction is auditable in `src/agentegrity/core/_telemetry_props.py`.
 
+### Security
+
+- **[Critical] Tool arguments could shadow the fields governance matches
+  on.** The `pre_tool_use` buffer entry was built by spreading the
+  agent's own arguments over the trusted fields
+  (`{"tool": tool_name, "type": "tool_call", **tool_input}`). That entry
+  becomes `context["action"]`, so an argument named `tool` overwrote the
+  real tool name and GOV-001's sensitive-tool gate evaluated the forgery:
+  a prompt-injected agent calling `payment_execute` with
+  `{"tool": "noop"}` skipped approval entirely. Under `enforce=True` the
+  dangerous tool ran; in observe-only mode the signed attestation
+  recorded `governance: pass` for a check that never happened. The same
+  trick overrode `type`. Untrusted arguments now nest under
+  `action["arguments"]`, where no agent-controlled key can collide with
+  a trusted field.
+- **[High] The LLM classifier skipped the two multi-agent channels.**
+  The regex scanner covers seven channels; `AdversarialLLMLayer.aevaluate`
+  fed only five to the classifier, so `shared_memory` and
+  `broadcast_messages` reached neither strong detector — the regex
+  taxonomy scores ~0 on the action-oriented injections that travel
+  through them. A compromised peer's injection into shared memory passed
+  clean, which is the cross-agent cascade (T-CASCADE) the v0.8 channels
+  exist to defend. Both channels are now classified, using the same
+  content keys and channel labels as the regex scanner.
+
+### Changed
+
+- **LLM classification cost is bounded.** `aevaluate` issued one
+  sequential API call per target with no ceiling, while the multi-agent
+  buffers cap at 1000 entries per channel — a latency tail and a
+  cost-amplification lever on attacker-writable content. Calls now run
+  with bounded concurrency (`max_concurrency`, default 8) and a
+  per-evaluation target ceiling (`max_targets_per_evaluation`, default
+  200). Truncation is logged and reported as
+  `details["llm_classifier"]["targets_dropped"]`, so a capped scan never
+  reads as full coverage. Verdicts are cached per (channel, content
+  hash) so the growing buffers don't re-bill already-classified text;
+  fail-open verdicts are never cached, since they record an outage
+  rather than a judgment.
+
+### Migration from 0.8.1
+
+- **Tool-call actions nest their arguments.** `context["action"]` for a
+  tool call is now
+  `{"tool": …, "type": "tool_call", "arguments": {…}}` instead of the
+  arguments spread flat. Custom `PolicyRule` conditions reading
+  call-derived keys off the top level must read them from
+  `action["arguments"]`. The built-in GOV-003 financial rule was updated
+  accordingly: it reads `action["arguments"]["amount"]` and no longer
+  honors a top-level `amount`. Rules matching only `tool` or `type` are
+  unaffected.
+
 ## [0.8.1] - 2026-07-01
 
 A security-hardening release. It closes every finding from a full

src/agentegrity/adapters/base.py

@@ -712,7 +712,12 @@ def _handle_pre_tool_use(self, data: dict[str, Any]) -> dict[str, Any]:
 
         self._append_capped(
             self._buffer.tool_calls,
-            {"tool": tool_name, "type": "tool_call", **tool_input},
+            # Agent-supplied arguments are untrusted and live under
+            # "arguments" — never spread flat, or an argument named
+            # "tool"/"type" would shadow the fields governance rules
+            # (GOV-001's sensitive-tool gate) match on. Mirrors the
+            # nesting the decision-record path already uses.
+            {"tool": tool_name, "type": "tool_call", "arguments": tool_input},
             "tool_calls",
         )
         self._buffer.tool_usage[tool_name] += 1

src/agentegrity/layers/adversarial_llm.py

@@ -30,9 +30,12 @@
 
 from __future__ import annotations
 
+import asyncio
+import hashlib
 import json
 import logging
 import os
+from collections import OrderedDict
 from dataclasses import dataclass
 from typing import Any
 
@@ -46,6 +49,17 @@
 
 DEFAULT_MODEL = "claude-haiku-4-5-20251001"
 
+# Classification is one API call per (channel, text) target, and the
+# adapter's multi-agent buffers cap at 1000 entries per channel. Left
+# unbounded, one evaluation over an attacker-writable channel becomes
+# thousands of sequential calls — a latency tail and a cost-amplification
+# lever. Bound both the parallelism and the per-evaluation target count.
+DEFAULT_MAX_CONCURRENCY = 8
+DEFAULT_MAX_TARGETS = 200
+# Verdict cache bound. A layer instance can outlive many sessions, so
+# the cache needs its own ceiling — oldest entries are evicted first.
+DEFAULT_VERDICT_CACHE_SIZE = 4096
+
 
 @dataclass
 class _LLMConfig:
@@ -98,6 +112,10 @@ class LLMAdversarialAssessment:
     severity: float
     confidence: float
     description: str
+    # True only for fail-open verdicts (network error, missing key,
+    # malformed response). These are outages, not judgments — the
+    # verdict cache must never pin them.
+    failed: bool = False
 
     @classmethod
     def neutral(cls) -> "LLMAdversarialAssessment":
@@ -109,6 +127,7 @@ def neutral(cls) -> "LLMAdversarialAssessment":
             severity=0.0,
             confidence=0.0,
             description="LLM check failed open",
+            failed=True,
         )
 
 
@@ -189,6 +208,18 @@ class AdversarialLLMLayer(AdversarialLayer):
       severity, and confidence. The aggregate ``coherence_score`` is
       recomputed to reflect both signals.
 
+      It covers the same seven channels the regex scanner does,
+      including the v0.8 multi-agent ones (``shared_memory``,
+      ``broadcast_channels``) where the regex taxonomy is weakest.
+
+      Cost is bounded on two axes: at most ``max_concurrency`` calls
+      in flight, and at most ``max_targets_per_evaluation`` targets
+      per pass (truncation is logged and reported in
+      ``details["llm_classifier"]["targets_dropped"]``). Verdicts are
+      cached per (channel, content-hash), so the growing adapter
+      buffers don't re-bill text already classified. Fail-open
+      verdicts are never cached.
+
     Failures are fail-open — the layer reverts to pattern-based
     behaviour on any LLM-side error (network, missing key, malformed
     JSON, timeout).
@@ -211,6 +242,9 @@ def __init__(
         block_on_critical: bool = True,
         patterns: list[DetectorPattern] | None = None,
         extra_patterns: list[DetectorPattern] | None = None,
+        max_concurrency: int = DEFAULT_MAX_CONCURRENCY,
+        max_targets_per_evaluation: int = DEFAULT_MAX_TARGETS,
+        verdict_cache_size: int = DEFAULT_VERDICT_CACHE_SIZE,
     ) -> None:
         super().__init__(
             coherence_threshold=coherence_threshold,
@@ -220,6 +254,48 @@ def __init__(
             extra_patterns=extra_patterns,
         )
         self._llm_config = _LLMConfig(api_key=api_key, model=model, timeout=timeout)
+        self._max_concurrency = max(1, max_concurrency)
+        self._max_targets = max(1, max_targets_per_evaluation)
+        # Content-addressed verdict cache. Adapter buffers only grow, so
+        # every evaluation re-presents text already classified in an
+        # earlier pass; re-billing it is pure waste. Keyed on
+        # (channel, sha256(text)) because the same text on a different
+        # channel is a different question. Insertion-ordered, evicted
+        # oldest-first at verdict_cache_size.
+        self._max_cache = max(1, verdict_cache_size)
+        self._verdict_cache: OrderedDict[
+            tuple[str, str], LLMAdversarialAssessment
+        ] = OrderedDict()
+
+    @staticmethod
+    def _cache_key(channel: str, text: str) -> tuple[str, str]:
+        return (channel, hashlib.sha256(text.encode("utf-8")).hexdigest())
+
+    async def _classify_targets(
+        self, targets: list[tuple[str, str]]
+    ) -> list[tuple[tuple[str, str], LLMAdversarialAssessment]]:
+        """Classify targets with bounded concurrency, reusing cached
+        verdicts. Returns (target, verdict) pairs in target order."""
+        semaphore = asyncio.Semaphore(self._max_concurrency)
+
+        async def classify(target: tuple[str, str]) -> LLMAdversarialAssessment:
+            channel, text = target
+            cached = self._verdict_cache.get(self._cache_key(channel, text))
+            if cached is not None:
+                return cached
+            async with semaphore:
+                verdict = await _call_claude_classify(self._llm_config, text)
+            # Never cache a fail-open verdict: it records an outage, not
+            # a judgment, and pinning it would blind every later pass
+            # over the same text.
+            if not verdict.failed:
+                self._verdict_cache[self._cache_key(channel, text)] = verdict
+                while len(self._verdict_cache) > self._max_cache:
+                    self._verdict_cache.popitem(last=False)
+            return verdict
+
+        verdicts = await asyncio.gather(*(classify(t) for t in targets))
+        return list(zip(targets, verdicts))
 
     async def aevaluate(
         self,
@@ -260,6 +336,44 @@ async def aevaluate(
                 )
                 if isinstance(content, str) and content.strip():
                     targets.append(("peer_messages", content))
+        # Multi-agent channels (v0.8). These are the cascade-compromise
+        # surface, and the regex taxonomy scores ~0 on the action-oriented
+        # injections that travel through them, so the LLM classifier must
+        # see them too. Channel labels and content keys match the regex
+        # scanner in AdversarialLayer._detect_channel_threats — the label
+        # asymmetry (broadcast_messages key, broadcast_channels label) is
+        # deliberate: the dedup map below is keyed on the label.
+        topology_ctx = ctx.get("topology_context") or {}
+        for entry in topology_ctx.get("shared_memory", []) or []:
+            if isinstance(entry, dict):
+                content = (
+                    entry.get("content")
+                    or entry.get("summary")
+                    or entry.get("text")
+                )
+                if isinstance(content, str) and content.strip():
+                    targets.append(("shared_memory", content))
+        for entry in topology_ctx.get("broadcast_messages", []) or []:
+            if isinstance(entry, dict):
+                content = entry.get("content") or entry.get("text")
+                if isinstance(content, str) and content.strip():
+                    targets.append(("broadcast_channels", content))
+
+        # Bound the per-evaluation cost. Truncation is logged and
+        # surfaced in details so a capped scan never reads as full
+        # coverage.
+        dropped = 0
+        if len(targets) > self._max_targets:
+            dropped = len(targets) - self._max_targets
+            logger.warning(
+                "adversarial LLM scan truncated: %d of %d targets "
+                "classified (max_targets_per_evaluation=%d); %d unscanned",
+                self._max_targets,
+                len(targets),
+                self._max_targets,
+                dropped,
+            )
+            targets = targets[: self._max_targets]
 
         # Existing threat_types per channel — used to skip duplicate
         # ThreatAssessments when the LLM agrees with the regex.
@@ -271,8 +385,7 @@ async def aevaluate(
             )
 
         llm_threats: list[ThreatAssessment] = []
-        for channel, text in targets:
-            verdict = await _call_claude_classify(self._llm_config, text)
+        for (channel, _text), verdict in await self._classify_targets(targets):
             if not verdict.is_attack:
                 continue
             # Don't duplicate an attack the regex taxonomy already
@@ -292,7 +405,11 @@ async def aevaluate(
             )
 
         if not llm_threats:
-            base_result.details["llm_classifier"] = {"new_threats": 0}
+            base_result.details["llm_classifier"] = {
+                "new_threats": 0,
+                "targets_scanned": len(targets),
+                "targets_dropped": dropped,
+            }
             return base_result
 
         # Append LLM-detected threats and recompute coherence + action.
@@ -304,6 +421,8 @@ async def aevaluate(
         base_result.details["llm_classifier"] = {
             "new_threats": len(llm_threats),
             "new_families": sorted({t.threat_type for t in llm_threats}),
+            "targets_scanned": len(targets),
+            "targets_dropped": dropped,
         }
 
         # Recompute coherence score from the union of pattern + LLM

src/agentegrity/layers/governance.py

@@ -179,8 +179,14 @@ def _rule_code_execution_boundary(
 def _rule_financial_threshold(
     profile: AgentProfile, action: dict[str, Any], context: dict[str, Any]
 ) -> bool:
-    """Flag financial transactions above threshold."""
-    amount = action.get("amount", 0)
+    """Flag financial transactions above threshold.
+
+    Reads ``action["arguments"]["amount"]``. Untrusted call arguments
+    are nested under ``arguments`` so they can never shadow the trusted
+    top-level ``tool``/``type`` fields; callers passing actions directly
+    must use the nested shape.
+    """
+    amount = (action.get("arguments") or {}).get("amount", 0)
     threshold = context.get("financial_threshold", 1000)
     return action.get("type") == "financial" and amount > threshold