fix(evaluators): budget R2 -- percent-encoding, fail-closed snapshot, dunder guard

R2 findings:

- _sanitize_scope_value: percent-encode |/= instead of replacing with _
  (was causing key collisions between "a|b" and "a_b")
- max_buckets fail-closed: spent_usd/spent_tokens now 0.0/0 (not recorded,
  previously reported current-call-only values misleading callers)
- _extract_by_path: narrowed guard from startswith("_") to startswith("__")
  (single-underscore dict keys are legitimate data fields)
- Fixed tautological test assertion in test_scope_key_injection_pipe
- Added 3 tests: no-collision, single-underscore access, NaN/Inf cost

57 budget tests, 287 total evaluator tests passing.

Author: amabito

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

@@ -42,8 +42,12 @@ def _derive_period_key(window: str | None) -> str:
 
 
 def _sanitize_scope_value(val: str) -> str:
-    """Remove pipe and equals from scope values to prevent key injection."""
-    return val.replace("|", "_").replace("=", "_")
+    """Percent-encode pipe and equals in scope values to prevent key injection.
+
+    Uses percent-encoding (not replacement) to preserve round-trip
+    distinction: "a|b" and "a_b" remain different keys.
+    """
+    return val.replace("%", "%25").replace("|", "%7C").replace("=", "%3D")
 
 
 def _build_scope_key(
@@ -68,7 +72,7 @@ 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("_"):
+        if part.startswith("__"):
             return None
         if isinstance(current, dict):
             current = current.get(part)

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

@@ -139,10 +139,12 @@ def record_and_check(
             bucket = self._buckets.get(key)
             if bucket is None:
                 if len(self._buckets) >= self._max_buckets:
-                    # Fail-closed: treat as exceeded to prevent OOM
+                    # Fail-closed: treat as exceeded to prevent OOM.
+                    # spent fields are 0 because no bucket exists for
+                    # this scope+period -- the call is rejected, not recorded.
                     return BudgetSnapshot(
-                        spent_usd=cost_usd,
-                        spent_tokens=input_tokens + output_tokens,
+                        spent_usd=0.0,
+                        spent_tokens=0,
                         limit_usd=limit_usd,
                         limit_tokens=limit_tokens,
                         utilization=1.0,

evaluators/builtin/tests/budget/test_budget.py

@@ -412,15 +412,26 @@ def test_exceeded_at_exact_limit_tokens(self) -> None:
     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
+        parts = key.split("|")
+        assert len(parts) == 2, f"Expected 2 parts, got {len(parts)}: {parts}"
+        assert "ch=admin" not in parts  # injected component not a segment
+
+    def test_scope_key_no_collision(self) -> None:
+        """Percent-encoding must keep 'a|b' distinct from 'a_b'."""
+        key1 = _build_scope_key({}, "uid", {"uid": "a|b"})
+        key2 = _build_scope_key({}, "uid", {"uid": "a_b"})
+        assert key1 != key2
 
     def test_max_buckets_prevents_oom(self) -> None:
         store = InMemoryBudgetStore(max_buckets=5)
+        exceeded_count = 0
         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
+            if snap.exceeded:
+                exceeded_count += 1
+                assert snap.spent_usd == 0.0  # not recorded, fail-closed
         assert len(store._buckets) == 5
+        assert exceeded_count == 5  # scopes 5-9 rejected
 
     @pytest.mark.asyncio
     async def test_per_without_metadata_skips_rule(self) -> None:
@@ -440,3 +451,15 @@ 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
+
+    def test_extract_by_path_allows_single_underscore(self) -> None:
+        """Single-underscore keys in data dicts are legitimate."""
+        from agent_control_evaluators.budget.evaluator import _extract_by_path
+        assert _extract_by_path({"_metadata": {"tokens": 42}}, "_metadata.tokens") == 42
+
+    def test_extract_cost_rejects_nan_inf(self) -> None:
+        """NaN and Inf must not pass through as cost values."""
+        from agent_control_evaluators.budget.evaluator import _extract_cost
+        assert _extract_cost({"c": float("nan")}, "c") is None
+        assert _extract_cost({"c": float("inf")}, "c") is None
+        assert _extract_cost({"c": float("-inf")}, "c") is None