feat(core): batch eval with timeout (rebac-batch-native)

CHANGELOG.md

@@ -5,6 +5,63 @@ All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning 2.0.0](https://semver.org/spec/v2.0.0.html).
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
+## 1.17.0 — 2026-04-05
+
+**Added**
+
+* **SpiceDB native `batch_check`** — async mode now issues a single
+  `BulkCheckPermissions` gRPC call instead of N sequential `CheckPermission`
+  calls.  Results are reassembled in input order from `resp.pairs`.  On RPC
+  error the entire batch returns `[False] * N` (fail-closed).  Sync mode
+  retains the sequential fallback (no bulk endpoint on sync gRPC client).
+
+* **OpenFGA `batch_check` empty-list guard** — `batch_check([])` now returns
+  `[]` immediately without issuing any HTTP request.
+
+* **`evaluate_batch_async` / `evaluate_batch_sync` `timeout` parameter** —
+  optional wall-clock deadline in seconds for the entire batch.  When exceeded,
+  `asyncio.TimeoutError` is raised.  Useful when individual checks call a slow
+  ReBAC provider and you need a bound on total latency.
+
+  ```python
+  decisions = await guard.evaluate_batch_async(requests, timeout=2.0)
+  ```
+
+* **`rbacx_batch_size` metric** — `evaluate_batch_async` now calls
+  `metrics.observe("rbacx_batch_size", N)` after each non-empty batch.
+  Both `PrometheusMetrics` and `OpenTelemetryMetrics` expose a dedicated
+  histogram for this metric.  Useful for capacity planning and TTL tuning.
+
+* **`require_batch_access` FastAPI dependency** — evaluates multiple
+  `(action, resource_type)` pairs in one `evaluate_batch_async` call and
+  returns a `list[Decision]`.  Designed for UI-state endpoints that need
+  to know which actions a user may perform at once.
+
+  ```python
+  from rbacx.adapters.fastapi import require_batch_access
+
+  @app.get("/ui-state")
+  async def ui_state(
+      decisions=Depends(
+          require_batch_access(guard,
+              [("read", "doc"), ("write", "doc"), ("delete", "doc")],
+              build_subject, timeout=2.0)
+      )
+  ):
+      return {"can_read": decisions[0].allowed, "can_write": decisions[1].allowed}
+  ```
+
+## 1.16.2 — 2026-04-05
+
+**Fixed**
+
+* `examples/ai_fastapi_demo/app.py` — `PolicyGenerationError` (invalid JSON
+  from LLM, network error, rate limit, etc.) was not caught inside
+  `_generate_policy`, causing the Starlette lifespan to propagate the
+  exception and kill the application on startup.  The call to
+  `ai.from_schema()` is now wrapped in `try/except Exception`; any failure
+  logs a `WARNING` and falls back to the built-in policy.
+
 ## 1.16.1 — 2026-04-05
 
 **Added**

docs/metrics.md

@@ -7,10 +7,12 @@ RBACX can emit metrics via **Prometheus** or **OpenTelemetry**.
 Use `PrometheusMetrics` sink (requires `prometheus_client`). Exposes:
 - `rbacx_decisions_total{allowed,reason}` — counter of decisions.
 - `rbacx_decision_duration_seconds` — histogram (adapters can observe latency).
+- `rbacx_batch_size` — histogram of `evaluate_batch_*` call sizes (requests per call).
 
 ## OpenTelemetry
 Use `OpenTelemetryMetrics` (requires `opentelemetry-api`). Creates instruments:
 - Counter `rbacx.decisions` (attributes: `allowed`, `reason`).
 - Histogram `rbacx.decision.duration.ms`.
+- Histogram `rbacx_batch_size` (unit: `{request}`) — `evaluate_batch_*` call sizes.
 
-See OpenTelemetry Metrics API and Prometheus client docs for details. 
+See OpenTelemetry Metrics API and Prometheus client docs for details.

docs/performance.md

@@ -14,3 +14,7 @@
   enabled/disabled buttons). Requests run concurrently via `asyncio.gather`,
   so the wall-clock time equals the slowest individual check rather than the
   sum of all checks.
+- **Set `timeout` on batch calls** when individual checks may hit a slow external
+  provider (SpiceDB, OpenFGA). Use `timeout=N` to bound total wall-clock time;
+  `asyncio.TimeoutError` is raised on expiry — catch it and return a safe fallback
+  rather than letting the request hang indefinitely.

docs/web_adapters.md

@@ -39,6 +39,38 @@ async def doc():
     return {"ok": True}
 ```
 
+
+### Batch access check (UI state)
+
+Use `require_batch_access` to evaluate multiple actions in one call — ideal for
+rendering UI elements (show/hide buttons) without N sequential requests:
+
+```python
+from rbacx.adapters.fastapi import require_batch_access
+from rbacx import Subject
+
+def build_subject(request: Request) -> Subject:
+    role = request.headers.get("X-Role", "viewer")
+    return Subject(id="user", roles=[role])
+
+@app.get("/ui-state")
+async def ui_state(
+    decisions=Depends(
+        require_batch_access(
+            guard,
+            [("read", "document"), ("write", "document"), ("delete", "document")],
+            build_subject,
+            timeout=2.0,  # optional deadline for the whole batch
+        )
+    )
+):
+    return {
+        "can_read":   decisions[0].allowed,
+        "can_write":  decisions[1].allowed,
+        "can_delete": decisions[2].allowed,
+    }
+```
+
 ---
 
 ## Flask (decorator)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "rbacx"
-version = "1.16.1"
+version = "1.17.0"
 description = "RBAC/ABAC policy engine for Python with policy sets, condition DSL, and hot reload"
 readme = "README.md"
 requires-python = ">=3.10"

src/rbacx/adapters/fastapi.py

@@ -1,4 +1,4 @@
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Sequence
 
 try:  # Optional dependency boundary
     from fastapi import HTTPException, Request  # type: ignore[import-not-found]
@@ -46,3 +46,74 @@ async def dependency(request: Request) -> None:
         raise HTTPException(status_code=403, detail="Forbidden", headers=headers)
 
     return dependency
+
+
+def require_batch_access(
+    guard: Guard,
+    actions_resources: Sequence[tuple[str, str]],
+    build_subject: Callable,
+    *,
+    timeout: float | None = None,
+) -> Callable:
+    """Return a FastAPI dependency that evaluates multiple access checks in one batch.
+
+    Designed for UI endpoints that need to know which actions are permitted for
+    a given user at once (e.g. to show/hide buttons) — avoids N sequential
+    ``evaluate_async`` calls.
+
+    Args:
+        guard: the :class:`~rbacx.core.engine.Guard` instance.
+        actions_resources: sequence of ``(action, resource_type)`` string pairs
+            that define the checks to perform.  Each pair maps to one
+            :class:`~rbacx.core.model.Decision` in the returned list.
+        build_subject: callable ``(request) -> Subject`` that extracts the
+            subject from the FastAPI request.  Only the subject varies per
+            request; action and resource are taken from *actions_resources*.
+        timeout: optional wall-clock deadline in seconds for the entire batch.
+            ``None`` means no deadline.  When exceeded
+            :class:`asyncio.TimeoutError` propagates as a 500 error.
+
+    Returns:
+        A FastAPI dependency that, when injected, resolves to a
+        ``list[Decision]`` — one entry per ``(action, resource_type)`` pair,
+        in the same order.
+
+    Example::
+
+        from rbacx.adapters.fastapi import require_batch_access
+        from rbacx import Subject, Action, Resource, Context
+
+        def build_subject(request: Request) -> Subject:
+            role = request.headers.get("X-Role", "viewer")
+            return Subject(id="user", roles=[role])
+
+        @app.get("/ui-state")
+        async def ui_state(
+            decisions=Depends(
+                require_batch_access(
+                    guard,
+                    [("read", "document"), ("write", "document"), ("delete", "document")],
+                    build_subject,
+                )
+            )
+        ):
+            return {
+                "can_read":   decisions[0].allowed,
+                "can_write":  decisions[1].allowed,
+                "can_delete": decisions[2].allowed,
+            }
+    """
+    from ..core.model import Action, Context, Resource  # noqa: PLC0415
+
+    async def dependency(request: Request) -> list:
+        if Request is None:  # pragma: no cover
+            raise RuntimeError("fastapi is required for adapters.fastapi")
+
+        subject = build_subject(request)
+        requests = [
+            (subject, Action(action), Resource(type=rtype), Context())
+            for action, rtype in actions_resources
+        ]
+        return await guard.evaluate_batch_async(requests, timeout=timeout)
+
+    return dependency

src/rbacx/core/engine.py

@@ -424,6 +424,7 @@ async def evaluate_batch_async(
         requests: Sequence[tuple[Subject, Action, Resource, Context | None]],
         *,
         explain: bool = False,
+        timeout: float | None = None,
     ) -> list[Decision]:
         """Evaluate multiple access requests concurrently, preserving order.
 
@@ -437,32 +438,59 @@ async def evaluate_batch_async(
                 tuples.  *context* may be ``None``.
             explain: when ``True``, populate :attr:`Decision.trace` on every
                 returned :class:`Decision`.
+            timeout: optional wall-clock deadline in seconds for the **entire
+                batch**.  When exceeded :class:`asyncio.TimeoutError` is raised
+                and no partial results are returned.  ``None`` (default) means
+                no deadline.  Useful when individual checks may call a slow
+                ReBAC provider (e.g. SpiceDB, OpenFGA) and you need a bound on
+                total latency.
 
         Returns:
             List of :class:`Decision` objects, one per request, preserving
             input order.
 
+        Raises:
+            asyncio.TimeoutError: when *timeout* is set and the batch does not
+                complete within the allotted time.
+
         Example::
 
             decisions = await guard.evaluate_batch_async([
                 (subject, Action("read"),   resource1, ctx),
                 (subject, Action("write"),  resource1, ctx),
                 (subject, Action("delete"), resource2, None),
-            ])
+            ], timeout=2.0)
         """
         if not requests:
             return []
-        return list(
-            await asyncio.gather(
-                *[self._evaluate_core_async(s, a, r, c, explain=explain) for s, a, r, c in requests]
-            )
-        )
+
+        coros = [self._evaluate_core_async(s, a, r, c, explain=explain) for s, a, r, c in requests]
+        gathered = asyncio.gather(*coros)
+        if timeout is not None:
+            gathered = asyncio.wait_for(gathered, timeout=timeout)  # type: ignore[assignment]
+
+        results = list(await gathered)
+
+        # Emit batch_size metric so operators can tune pool sizes and TTLs.
+        if self.metrics is not None:
+            try:
+                observe = getattr(self.metrics, "observe", None)
+                if observe is not None:
+                    if inspect.iscoroutinefunction(observe):
+                        await observe("rbacx_batch_size", float(len(requests)))
+                    else:
+                        observe("rbacx_batch_size", float(len(requests)))
+            except Exception:
+                logger.exception("RBACX: metrics.observe(rbacx_batch_size) failed")
+
+        return results
 
     def evaluate_batch_sync(
         self,
         requests: Sequence[tuple[Subject, Action, Resource, Context | None]],
         *,
         explain: bool = False,
+        timeout: float | None = None,
     ) -> list[Decision]:
         """Synchronous wrapper for :meth:`evaluate_batch_async`.
 
@@ -476,10 +504,16 @@ def evaluate_batch_sync(
                 tuples.  *context* may be ``None``.
             explain: when ``True``, populate :attr:`Decision.trace` on every
                 returned :class:`Decision`.
+            timeout: optional wall-clock deadline in seconds passed through to
+                :meth:`evaluate_batch_async`.  ``None`` means no deadline.
 
         Returns:
             List of :class:`Decision` objects, one per request, preserving
             input order.
+
+        Raises:
+            asyncio.TimeoutError: when *timeout* is set and the batch does not
+                complete within the allotted time.
         """
         if not requests:
             return []
@@ -491,10 +525,14 @@ def evaluate_batch_sync(
             loop_running = False
 
         if not loop_running:
-            return asyncio.run(self.evaluate_batch_async(requests, explain=explain))
+            return asyncio.run(
+                self.evaluate_batch_async(requests, explain=explain, timeout=timeout)
+            )
 
         def _runner() -> list[Decision]:
-            return asyncio.run(self.evaluate_batch_async(requests, explain=explain))
+            return asyncio.run(
+                self.evaluate_batch_async(requests, explain=explain, timeout=timeout)
+            )
 
         if Guard._executor is None:
             Guard._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="rbacx-sync")