fix(evaluators): budget evaluator R1 -- security hardening + 6 adversarial tests

3-body review findings:

Security:
- Sanitize pipe/equals in scope key metadata values (injection prevention)
- Add max_buckets=100K to InMemoryBudgetStore (OOM prevention, fail-closed)
- Block dunder attribute access in _extract_by_path
- Add math.isfinite guard on extracted cost values
- Skip per-user rules when per field missing from metadata (was collapsing
  per-user budgets into global bucket)

Correctness:
- Changed exceeded check from > to >= (utilization=100% now triggers exceeded)
- Removed unused BudgetSnapshot import from evaluator.py

Tests (6 adversarial):
- Exact limit boundary (USD and tokens)
- Scope key injection via pipe character
- max_buckets OOM prevention
- per-field missing skips rule
- dunder path rejection

54 budget tests, 284 total evaluator tests passing.

Author: amabito

evaluators/builtin/src/agent_control_evaluators/budget/evaluator.py

@@ -8,13 +8,14 @@
 from __future__ import annotations
 
 import logging
+import math
 from datetime import datetime, timezone
 from typing import Any
 
 from agent_control_evaluators._base import Evaluator, EvaluatorMetadata
 from agent_control_evaluators._registry import register_evaluator
 from agent_control_evaluators.budget.config import BudgetEvaluatorConfig, BudgetLimitRule
-from agent_control_evaluators.budget.store import BudgetSnapshot, InMemoryBudgetStore
+from agent_control_evaluators.budget.store import InMemoryBudgetStore
 from agent_control_models import EvaluatorResult
 
 logger = logging.getLogger(__name__)
@@ -40,6 +41,11 @@ def _derive_period_key(window: str | None) -> str:
     return ""
 
 
+def _sanitize_scope_value(val: str) -> str:
+    """Remove pipe and equals from scope values to prevent key injection."""
+    return val.replace("|", "_").replace("=", "_")
+
+
 def _build_scope_key(
     scope: dict[str, str],
     per: str | None,
@@ -48,19 +54,22 @@ def _build_scope_key(
     """Build a composite scope key from static dimensions and per-field.
 
     Format: "channel=slack|user_id=u1" or "__global__" if empty.
+    Values are sanitized to prevent injection via pipe/equals characters.
     """
     parts: list[str] = []
     for k, v in sorted(scope.items()):
-        parts.append(f"{k}={v}")
+        parts.append(f"{k}={_sanitize_scope_value(v)}")
     if per and per in metadata:
-        parts.append(f"{per}={metadata[per]}")
+        parts.append(f"{per}={_sanitize_scope_value(metadata[per])}")
     return "|".join(parts) if parts else "__global__"
 
 
 def _extract_by_path(data: Any, path: str) -> Any:
     """Extract a value from nested data using dot-notation path."""
     current = data
     for part in path.split("."):
+        if part.startswith("_"):
+            return None
         if isinstance(current, dict):
             current = current.get(part)
         elif hasattr(current, part):
@@ -112,7 +121,7 @@ def _extract_cost(data: Any, cost_path: str | None) -> float | None:
     if data is None or cost_path is None:
         return None
     val = _extract_by_path(data, cost_path)
-    if isinstance(val, (int, float)) and val >= 0:
+    if isinstance(val, (int, float)) and math.isfinite(val) and val >= 0:
         return float(val)
     return None
 
@@ -286,11 +295,16 @@ async def evaluate(self, data: Any) -> EvaluatorResult:
 
 
 def _scope_matches(rule: BudgetLimitRule, metadata: dict[str, str]) -> bool:
-    """Check if rule's static scope dimensions match step metadata.
+    """Check if rule's scope dimensions match step metadata.
 
     An empty scope dict matches everything (global rule).
+    If ``per`` is set but the field is missing from metadata, the rule
+    is skipped to prevent per-user budgets from collapsing into a
+    single global bucket.
     """
     for key, expected in rule.scope.items():
         if metadata.get(key) != expected:
             return False
+    if rule.per and rule.per not in metadata:
+        return False
     return True

evaluators/builtin/src/agent_control_evaluators/budget/store.py

@@ -103,9 +103,12 @@ class InMemoryBudgetStore:
     Redis or Postgres-backed store (separate package).
     """
 
-    def __init__(self) -> None:
+    _DEFAULT_MAX_BUCKETS = 100_000
+
+    def __init__(self, *, max_buckets: int = _DEFAULT_MAX_BUCKETS) -> None:
         self._lock = threading.Lock()
         self._buckets: dict[tuple[str, str], _Bucket] = {}
+        self._max_buckets = max_buckets
 
     def record_and_check(
         self,
@@ -135,6 +138,16 @@ def record_and_check(
         with self._lock:
             bucket = self._buckets.get(key)
             if bucket is None:
+                if len(self._buckets) >= self._max_buckets:
+                    # Fail-closed: treat as exceeded to prevent OOM
+                    return BudgetSnapshot(
+                        spent_usd=cost_usd,
+                        spent_tokens=input_tokens + output_tokens,
+                        limit_usd=limit_usd,
+                        limit_tokens=limit_tokens,
+                        utilization=1.0,
+                        exceeded=True,
+                    )
                 bucket = _Bucket()
                 self._buckets[key] = bucket
             bucket.spent_usd += cost_usd
@@ -146,9 +159,9 @@ def record_and_check(
                 bucket.spent_usd, total_tokens, limit_usd, limit_tokens
             )
             exceeded = False
-            if limit_usd is not None and bucket.spent_usd > limit_usd:
+            if limit_usd is not None and bucket.spent_usd >= limit_usd:
                 exceeded = True
-            if limit_tokens is not None and total_tokens > limit_tokens:
+            if limit_tokens is not None and total_tokens >= limit_tokens:
                 exceeded = True
 
             return BudgetSnapshot(
@@ -185,9 +198,9 @@ def get_snapshot(
                 bucket.spent_usd, total_tokens, limit_usd, limit_tokens
             )
             exceeded = False
-            if limit_usd is not None and bucket.spent_usd > limit_usd:
+            if limit_usd is not None and bucket.spent_usd >= limit_usd:
                 exceeded = True
-            if limit_tokens is not None and total_tokens > limit_tokens:
+            if limit_tokens is not None and total_tokens >= limit_tokens:
                 exceeded = True
             return BudgetSnapshot(
                 spent_usd=bucket.spent_usd,

evaluators/builtin/tests/budget/test_budget.py

@@ -387,3 +387,56 @@ def test_from_dict(self) -> None:
         ev = BudgetEvaluator.from_dict({"limits": [{"limit_usd": 5.0}]})
         assert isinstance(ev, BudgetEvaluator)
         assert ev.config.limits[0].limit_usd == 5.0
+
+
+# ---------------------------------------------------------------------------
+# Security / adversarial tests (R1 findings)
+# ---------------------------------------------------------------------------
+
+
+class TestBudgetAdversarial:
+    """Adversarial tests for budget evaluator security."""
+
+    def test_exceeded_at_exact_limit_usd(self) -> None:
+        """Spending exactly the limit must trigger exceeded (>= not >)."""
+        store = InMemoryBudgetStore()
+        snap = store.record_and_check("s", "p", 0, 0, 1.0, limit_usd=1.0)
+        assert snap.exceeded is True
+        assert snap.utilization == pytest.approx(1.0)
+
+    def test_exceeded_at_exact_limit_tokens(self) -> None:
+        store = InMemoryBudgetStore()
+        snap = store.record_and_check("s", "p", 500, 500, 0.0, limit_tokens=1000)
+        assert snap.exceeded is True
+
+    def test_scope_key_injection_pipe(self) -> None:
+        """Pipe in metadata value must be sanitized, not create new scope dimension."""
+        key = _build_scope_key({"ch": "slack"}, "uid", {"ch": "slack", "uid": "u1|ch=admin"})
+        assert "|ch=admin" not in key.split("|")[-1]  # injected dimension not present
+        assert "u1_ch_admin" in key  # sanitized
+
+    def test_max_buckets_prevents_oom(self) -> None:
+        store = InMemoryBudgetStore(max_buckets=5)
+        for i in range(10):
+            snap = store.record_and_check(f"scope-{i}", "p", 1, 1, 0.01, limit_usd=100.0)
+        # After 5, new buckets are rejected with exceeded=True
+        assert len(store._buckets) == 5
+
+    @pytest.mark.asyncio
+    async def test_per_without_metadata_skips_rule(self) -> None:
+        """per='user_id' but user_id missing -> rule skipped, not global."""
+        from agent_control_evaluators.budget.evaluator import BudgetEvaluator
+        config = BudgetEvaluatorConfig(
+            limits=[{"scope": {}, "per": "user_id", "limit_usd": 1.0}],
+            cost_path="cost",
+            metadata_paths={"user_id": "user_id"},
+        )
+        ev = BudgetEvaluator(config)
+        # No user_id in data -> rule skipped -> not matched (no applicable rules)
+        result = await ev.evaluate({"cost": 999.0})
+        assert result.matched is False
+
+    def test_extract_by_path_rejects_dunder(self) -> None:
+        from agent_control_evaluators.budget.evaluator import _extract_by_path
+        assert _extract_by_path({"a": 1}, "__class__") is None
+        assert _extract_by_path({"a": {"__init__": 1}}, "a.__init__") is None