diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8748a7a..f00f662 100644
--- a/CHANGELOG.md
+++ b/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**
diff --git a/docs/metrics.md b/docs/metrics.md
index 5907bb5..a0fbf22 100644
--- a/docs/metrics.md
+++ b/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.
diff --git a/docs/performance.md b/docs/performance.md
index ef07891..606eb5c 100644
--- a/docs/performance.md
+++ b/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.
diff --git a/docs/web_adapters.md b/docs/web_adapters.md
index b1d2e94..f2e6425 100644
--- a/docs/web_adapters.md
+++ b/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)
diff --git a/pyproject.toml b/pyproject.toml
index c9a8ee1..3538e25 100644
--- a/pyproject.toml
+++ b/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"
diff --git a/src/rbacx/adapters/fastapi.py b/src/rbacx/adapters/fastapi.py
index 1ce0375..98030cd 100644
--- a/src/rbacx/adapters/fastapi.py
+++ b/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
diff --git a/src/rbacx/core/engine.py b/src/rbacx/core/engine.py
index 7cfd6bf..004db98 100644
--- a/src/rbacx/core/engine.py
+++ b/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")
diff --git a/src/rbacx/metrics/otel.py b/src/rbacx/metrics/otel.py
index 1046d69..8620d0a 100644
--- a/src/rbacx/metrics/otel.py
+++ b/src/rbacx/metrics/otel.py
@@ -14,6 +14,7 @@ class OpenTelemetryMetrics(MetricsSink):
     Creates:
       - Counter: rbacx_decisions_total (labels: decision)
       - Histogram: rbacx_decision_seconds (unit: s)
+      - Histogram: rbacx_batch_size (unit: {request}) — evaluate_batch_* call sizes
 
     Notes:
       * OTEL recommends carrying the **unit** in metadata; we also keep `_seconds` in the name
@@ -25,11 +26,13 @@ class OpenTelemetryMetrics(MetricsSink):
     # Explicit attribute annotations for mypy
     _counter: Any | None
     _hist: Any | None
+    _batch_hist: Any | None
 
     def __init__(self) -> None:
         # Ensure attributes always exist
         self._counter = None
         self._hist = None
+        self._batch_hist = None
 
         if get_meter is None:  # pragma: no cover
             return
@@ -59,6 +62,20 @@ def __init__(self) -> None:
         except Exception:  # pragma: no cover
             self._hist = None
 
+        # Batch size histogram
+        try:
+            create_hist = getattr(meter, "create_histogram", None)
+            if create_hist is not None:
+                self._batch_hist = create_hist(
+                    name="rbacx_batch_size",
+                    description="Distribution of evaluate_batch_* call sizes (requests per call).",
+                    unit="{request}",
+                )
+            else:  # pragma: no cover
+                self._batch_hist = None
+        except Exception:  # pragma: no cover
+            self._batch_hist = None
+
     # -- MetricsSink ------------------------------------------------------------
 
     def inc(self, name: str, labels: dict[str, str] | None = None) -> None:
@@ -80,26 +97,28 @@ def inc(self, name: str, labels: dict[str, str] | None = None) -> None:
 
     # ----------------------------- Optional extension --------------------------
     def observe(self, name: str, value: float, labels: dict[str, str] | None = None) -> None:
-        """Optionally record a latency distribution **in seconds**.
+        """Record a value in the appropriate histogram.
 
-        This is a **carcass method** so users can see how to implement it. Guard will call it
-        only if present (checked via ``hasattr``). If no OTEL SDK/pipeline is configured or
-        the histogram wasn't created, this method safely no-ops.
+        Routing:
+          - ``"rbacx_batch_size"`` → ``rbacx_batch_size`` histogram.
+          - Any other *name* → ``rbacx_decision_seconds`` latency histogram.
 
         Parameters
         ----------
         name: str
-            Metric name. Accepted for compatibility and future extensions; ignored here.
+            Metric name used for routing.
         value: float
-            Duration **in seconds** (as exposed by Guard).
+            Value to record.
         labels: dict[str, str] | None
-            OTEL Histogram accepts attributes; we pass them through if present.
+            OTEL Histogram accepts attributes; passed through if present.
         """
-        if self._hist is None:
-            return
         try:
-            # Histogram.record(value, attributes=labels) is the conventional API.
-            self._hist.record(float(value), attributes=dict(labels or {}))
+            if name == "rbacx_batch_size":
+                if self._batch_hist is not None:
+                    self._batch_hist.record(float(value), attributes=dict(labels or {}))
+            else:
+                if self._hist is not None:
+                    self._hist.record(float(value), attributes=dict(labels or {}))
         except Exception:  # pragma: no cover
             __import__("logging").getLogger("rbacx.metrics.otel").debug(
                 "OpenTelemetryMetrics.observe: failed to record histogram", exc_info=True
diff --git a/src/rbacx/metrics/prometheus.py b/src/rbacx/metrics/prometheus.py
index 9527d02..2ec21f3 100644
--- a/src/rbacx/metrics/prometheus.py
+++ b/src/rbacx/metrics/prometheus.py
@@ -14,6 +14,7 @@ class PrometheusMetrics(MetricsSink):
     Exposes:
       - rbacx_decisions_total{decision="allow|deny|..."}
       - rbacx_decision_seconds (Histogram) — optional latency distribution
+      - rbacx_batch_size (Histogram) — distribution of evaluate_batch_* call sizes
 
     Notes:
       * Counter uses the `_total` suffix and latency uses `_seconds` to follow Prometheus/OpenMetrics naming.
@@ -26,11 +27,13 @@ class PrometheusMetrics(MetricsSink):
     # Explicit attribute annotations for mypy
     _counter: Any | None
     _hist: Any | None
+    _batch_hist: Any | None
 
     def __init__(self) -> None:
         # default to None so attributes are always defined
         self._counter = None
         self._hist = None
+        self._batch_hist = None
 
         # create instruments only if the client is available
         if Counter is None or Histogram is None:  # pragma: no cover
@@ -47,6 +50,12 @@ def __init__(self) -> None:
             "rbacx_decision_seconds",
             "RBACX decision evaluation duration in seconds.",
         )
+        # Batch size histogram — number of requests per evaluate_batch_* call
+        self._batch_hist = Histogram(
+            "rbacx_batch_size",
+            "Distribution of rbacx evaluate_batch_* call sizes (number of requests per call).",
+            buckets=(1, 2, 5, 10, 25, 50, 100, 250, 500, 1000),
+        )
 
     # -- MetricsSink ------------------------------------------------------------
 
@@ -69,25 +78,29 @@ def inc(self, name: str, labels: dict[str, str] | None = None) -> None:
 
     # ----------------------------- Optional extension --------------------------
     def observe(self, name: str, value: float, labels: dict[str, str] | None = None) -> None:
-        """Optionally record a latency distribution **in seconds**.
+        """Record a value in the appropriate histogram.
 
-        This is a **carcass method** so users can see how to implement it. Guard will call it
-        only if present (checked via ``hasattr``). If Prometheus is unavailable or the histogram
-        wasn't created, this method safely no-ops.
+        Routing:
+          - ``"rbacx_batch_size"`` → ``rbacx_batch_size`` histogram.
+          - Any other *name* → ``rbacx_decision_seconds`` latency histogram.
 
         Parameters
         ----------
         name: str
-            Metric name. Accepted for compatibility and future extensions; ignored here.
+            Metric name used for routing (see above).
         value: float
-            Duration **in seconds** (as exposed by Guard).
+            Value to record.  For latency use seconds; for batch size use the
+            request count.
         labels: dict[str, str] | None
-            Currently unused (histogram has no labels by default).
+            Currently unused (histograms have no labels by default).
         """
-        if self._hist is None:
-            return
         try:
-            self._hist.observe(float(value))  # seconds
+            if name == "rbacx_batch_size":
+                if self._batch_hist is not None:
+                    self._batch_hist.observe(float(value))
+            else:
+                if self._hist is not None:
+                    self._hist.observe(float(value))
         except Exception:  # pragma: no cover
             __import__("logging").getLogger("rbacx.metrics.prometheus").debug(
                 "PrometheusMetrics.observe: failed to record histogram", exc_info=True
diff --git a/src/rbacx/rebac/openfga.py b/src/rbacx/rebac/openfga.py
index 9262027..a163fc8 100644
--- a/src/rbacx/rebac/openfga.py
+++ b/src/rbacx/rebac/openfga.py
@@ -136,6 +136,16 @@ def batch_check(
         context: dict[str, Any] | None = None,
         authorization_model_id: str | None = None,
     ):
+        """Check multiple (subject, relation, resource) triples via OpenFGA
+        ``/batch-check`` — a single HTTP round-trip for all *triples*.
+
+        Uses ``correlation_id`` per check to reassemble results in input order,
+        since the OpenFGA API does not guarantee response ordering.
+        On any HTTP error all results resolve to ``False`` (fail-closed).
+        """
+        if not triples:
+            return []
+
         # Build request with correlation_id per check (order is not guaranteed in responses)
         checks: list[dict[str, Any]] = []
         corr_ids: list[str] = []
diff --git a/src/rbacx/rebac/spicedb.py b/src/rbacx/rebac/spicedb.py
index 349027d..22f2722 100644
--- a/src/rbacx/rebac/spicedb.py
+++ b/src/rbacx/rebac/spicedb.py
@@ -8,6 +8,8 @@
 # Optional: install via extra rbacx[rebac-spicedb]
 try:
     from authzed.api.v1 import (
+        BulkCheckPermissionRequest,
+        BulkCheckPermissionRequestItem,
         CheckPermissionRequest,
         CheckPermissionResponse,
         Consistency,
@@ -57,6 +59,10 @@ async def CheckPermission(
         self, request: CheckPermissionRequest, timeout: float | None = ...
     ) -> CheckPermissionResponse: ...
 
+    async def BulkCheckPermissions(self, request: Any, timeout: float | None = ...) -> Any:
+        """Optional — available in authzed ≥ 0.9.  Detected via hasattr at runtime."""
+        ...
+
 
 def _dict_to_struct(d: dict[str, Any]) -> Struct:
     s = Struct()
@@ -69,7 +75,9 @@ class SpiceDBChecker(RelationshipChecker):
     """
     ReBAC provider backed by the SpiceDB/Authzed gRPC API.
 
-    - Uses CheckPermission; batch -> sequential calls (gRPC has no one-shot batch).
+    - Single checks use ``CheckPermission``.
+    - Batch checks use ``BulkCheckPermissions`` (one gRPC call for N triples)
+      in async mode; falls back to sequential sync calls otherwise.
     - Consistency: ZedToken (at_least_as_fresh) or fully_consistent.
     - Caveats: pass context as google.protobuf.Struct.
     """
@@ -167,22 +175,91 @@ def batch_check(
         context: dict[str, Any] | None = None,
         zed_token: str | None = None,
     ) -> list[bool] | Any:  # Any = Awaitable[list[bool]]
+        """Check multiple (subject, relation, resource) triples in one call.
+
+        *Async mode* uses ``BulkCheckPermissions`` when available (authzed ≥ 0.9)
+        — a single gRPC round-trip for all *triples*, preserving order.  Falls
+        back to concurrent ``CheckPermission`` calls when the bulk RPC is absent.
+
+        *Sync mode* falls back to sequential ``CheckPermission`` calls (the
+        SpiceDB sync gRPC client does not expose a bulk endpoint).
+
+        On any RPC error the affected item resolves to ``False`` (fail-closed).
+        """
+        if not triples:
+            return []
+
         if self._aclient is not None:
+            aclient = self._aclient
+            # BulkCheckPermissions was added in authzed ≥ 0.9; detect at runtime.
+            _has_bulk = hasattr(aclient, "BulkCheckPermissions")
 
             async def _run() -> list[bool]:
-                out: list[bool] = []
+                # Build consistency once for the whole batch
+                consistency: Any = None
+                if zed_token:
+                    consistency = Consistency(at_least_as_fresh=ZedToken(token=zed_token))
+                elif self.cfg.prefer_fully_consistent:
+                    consistency = Consistency(fully_consistent=True)
+
+                ctx_struct = _dict_to_struct(context) if context else None
+
+                items: list[Any] = []
                 for s, r, o in triples:
-                    out.append(
-                        await self._check_single_async(
-                            s, r, o, context=context, zed_token=zed_token
+                    obj_type, obj_id = (o.split(":", 1) + [""])[:2]
+                    subj_type, subj_id = (s.split(":", 1) + [""])[:2]
+                    items.append(
+                        BulkCheckPermissionRequestItem(
+                            resource=ObjectReference(object_type=obj_type, object_id=obj_id),
+                            permission=r,
+                            subject=SubjectReference(
+                                object=ObjectReference(object_type=subj_type, object_id=subj_id)
+                            ),
+                            context=ctx_struct,
                         )
                     )
-                return out
+
+                req = BulkCheckPermissionRequest(
+                    items=items,
+                    consistency=consistency,
+                )
+
+                if not _has_bulk:
+                    # Older authzed client — fall back to concurrent single checks
+                    import asyncio as _aio
+
+                    results = await _aio.gather(
+                        *[
+                            self._check_single_async(s, r, o, context=context, zed_token=zed_token)
+                            for s, r, o in triples
+                        ]
+                    )
+                    return list(results)
+
+                try:
+                    resp = await aclient.BulkCheckPermissions(req, timeout=self.cfg.timeout_seconds)
+                    # resp.pairs is a list of BulkCheckPermissionPair ordered
+                    # by input position when the API preserves order.
+                    # Each pair has .item (echo) and .item.permissionship.
+                    out: list[bool] = []
+                    for pair in resp.pairs:
+                        allowed = (
+                            pair.item.permissionship
+                            == CheckPermissionResponse.PERMISSIONSHIP_HAS_PERMISSION
+                        )
+                        out.append(allowed)
+                    return out
+                except RpcError as exc:
+                    logger.warning("SpiceDB BulkCheckPermissions RPC error: %s", exc, exc_info=True)
+                    return [False] * len(triples)
+                except Exception:
+                    logger.error("SpiceDB BulkCheckPermissions unexpected error", exc_info=True)
+                    return [False] * len(triples)
 
             return _run()
 
-        # sync fallback
-        return [self.check(s, r, o, context=context, zed_token=zed_token) for (s, r, o) in triples]
+        # sync mode: no bulk gRPC endpoint on sync client — sequential fallback
+        return [self.check(s, r, o, context=context, zed_token=zed_token) for s, r, o in triples]
 
     # -------------- helpers --------------
 
diff --git a/tests/integration/metrics/test_metrics_adapters.py b/tests/integration/metrics/test_metrics_adapters.py
index 10da6cd..c57d652 100644
--- a/tests/integration/metrics/test_metrics_adapters.py
+++ b/tests/integration/metrics/test_metrics_adapters.py
@@ -28,7 +28,7 @@ def labels(self, **kw):
             return _Lbl(self)
 
     class Hst:
-        def __init__(self, name, doc, labelnames=None, registry=None):
+        def __init__(self, name, doc, labelnames=None, registry=None, **kw):
             self.name, self.doc = name, doc
             self.labelnames = tuple(labelnames or [])
             self._vals = []
diff --git a/tests/unit/adapters/fastapi/test_fastapi_headers_new.py b/tests/unit/adapters/fastapi/test_fastapi_headers_new.py
index c23ec6c..3d2f9db 100644
--- a/tests/unit/adapters/fastapi/test_fastapi_headers_new.py
+++ b/tests/unit/adapters/fastapi/test_fastapi_headers_new.py
@@ -4,7 +4,7 @@
 @pytest.mark.asyncio
 async def test_fastapi_dependency_sets_rbacx_headers_on_deny():
     fastapi = pytest.importorskip("fastapi")
-    pytest.importorskip("starlette")
+    pytest.importorskip("starlette.requests")
 
     from starlette.requests import Request
 
diff --git a/tests/unit/adapters/starlette/test_starlette_new_branches.py b/tests/unit/adapters/starlette/test_starlette_new_branches.py
index 9138a88..a9c9ef3 100644
--- a/tests/unit/adapters/starlette/test_starlette_new_branches.py
+++ b/tests/unit/adapters/starlette/test_starlette_new_branches.py
@@ -36,7 +36,7 @@ def test_eval_guard_sync_path_indirect():
     Covers lines 42-43 indirectly via require_access: when guard exposes is_allowed_sync,
     the decorator should take the fast path (allowed).
     """
-    pytest.importorskip("starlette")
+    pytest.importorskip("starlette.requests")
     from starlette.requests import Request
 
     import rbacx.adapters.starlette as st_mod
@@ -77,7 +77,7 @@ async def test_dependency_returns_none_when_allowed_true():
     """
     Covers lines 68-69: dependency should return None (no deny) when allowed=True.
     """
-    pytest.importorskip("starlette")
+    pytest.importorskip("starlette.requests")
     from starlette.requests import Request
 
     import rbacx.adapters.starlette as st_mod
@@ -114,7 +114,7 @@ async def test_async_endpoint_allows_and_calls_handler():
     """
     Covers the tail of 82-94: deny is None on async handler -> returns await handler(request).
     """
-    pytest.importorskip("starlette")
+    pytest.importorskip("starlette.requests")
     from starlette.requests import Request
 
     import rbacx.adapters.starlette as st_mod
@@ -157,7 +157,7 @@ async def test_sync_endpoint_returns_callable_deny_when_dependency_returns_respo
     Covers line 106: when dependency returns an ASGI-callable deny (e.g. a Response),
     the wrapper must 'return deny' directly (no coercion, no threadpool).
     """
-    pytest.importorskip("starlette")
+    pytest.importorskip("starlette.requests")
     from starlette.requests import Request
     from starlette.responses import JSONResponse  # Response objects are ASGI-callable.
 
@@ -222,7 +222,7 @@ async def test_sync_endpoint_calls_real_run_in_threadpool():
     Covers line 107: deny is None on sync handler -> must execute
     `return await run_in_threadpool(handler, request)` (real threadpool call).
     """
-    pytest.importorskip("starlette")
+    pytest.importorskip("starlette.requests")
     from starlette.requests import Request
 
     import rbacx.adapters.starlette as st_mod
diff --git a/tests/unit/adapters/test_fastapi_batch.py b/tests/unit/adapters/test_fastapi_batch.py
new file mode 100644
index 0000000..50c8921
--- /dev/null
+++ b/tests/unit/adapters/test_fastapi_batch.py
@@ -0,0 +1,367 @@
+"""Unit tests for require_batch_access FastAPI dependency."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from rbacx import Guard, Subject
+
+_POLICY_MIXED = {
+    "algorithm": "deny-overrides",
+    "rules": [
+        {"id": "r-read", "effect": "permit", "actions": ["read"], "resource": {"type": "doc"}},
+        {
+            "id": "r-write",
+            "effect": "permit",
+            "actions": ["write"],
+            "resource": {"type": "doc"},
+            "roles": ["editor", "admin"],
+        },
+    ],
+}
+
+_S_VIEWER = Subject(id="u1", roles=["viewer"])
+_S_EDITOR = Subject(id="u2", roles=["editor"])
+
+
+def _fake_request(subject: Subject) -> MagicMock:
+    req = MagicMock()
+    req._subject = subject
+    return req
+
+
+# ---------------------------------------------------------------------------
+# Stubs — avoid FastAPI install
+# ---------------------------------------------------------------------------
+
+
+def _patch_fastapi(monkeypatch):
+    import rbacx.adapters.fastapi as mod
+
+    fake_http_exc = type(
+        "HTTPException",
+        (Exception,),
+        {"__init__": lambda self, status_code=403, detail="", headers=None: None},
+    )
+    monkeypatch.setattr(mod, "HTTPException", fake_http_exc, raising=False)
+    monkeypatch.setattr(mod, "Request", MagicMock, raising=False)
+
+
+# ---------------------------------------------------------------------------
+# Basic permit / deny
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_require_batch_access_returns_decisions(monkeypatch):
+    """require_batch_access resolves to a list of Decision objects."""
+    _patch_fastapi(monkeypatch)
+    from rbacx.adapters.fastapi import require_batch_access
+
+    guard = Guard(_POLICY_MIXED)
+
+    def build_subject(request):
+        return request._subject
+
+    dep = require_batch_access(
+        guard,
+        [("read", "doc"), ("write", "doc")],
+        build_subject,
+    )
+
+    req = _fake_request(_S_VIEWER)
+    decisions = await dep(req)
+    assert len(decisions) == 2
+    assert decisions[0].allowed is True  # viewer can read
+    assert decisions[1].allowed is False  # viewer cannot write
+
+
+@pytest.mark.asyncio
+async def test_require_batch_access_editor_can_write(monkeypatch):
+    """Editor role passes the write check."""
+    _patch_fastapi(monkeypatch)
+    from rbacx.adapters.fastapi import require_batch_access
+
+    guard = Guard(_POLICY_MIXED)
+
+    def build_subject(request):
+        return request._subject
+
+    dep = require_batch_access(
+        guard,
+        [("read", "doc"), ("write", "doc")],
+        build_subject,
+    )
+
+    req = _fake_request(_S_EDITOR)
+    decisions = await dep(req)
+    assert decisions[0].allowed is True
+    assert decisions[1].allowed is True
+
+
+# ---------------------------------------------------------------------------
+# Empty actions_resources
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_require_batch_access_empty_actions(monkeypatch):
+    """Empty actions_resources returns empty list."""
+    _patch_fastapi(monkeypatch)
+    from rbacx.adapters.fastapi import require_batch_access
+
+    guard = Guard(_POLICY_MIXED)
+    dep = require_batch_access(guard, [], lambda req: _S_VIEWER)
+    decisions = await dep(_fake_request(_S_VIEWER))
+    assert decisions == []
+
+
+# ---------------------------------------------------------------------------
+# Order preserved
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_require_batch_access_order_preserved(monkeypatch):
+    """Results are in the same order as actions_resources."""
+    _patch_fastapi(monkeypatch)
+    from rbacx.adapters.fastapi import require_batch_access
+
+    policy = {
+        "algorithm": "deny-overrides",
+        "rules": [
+            {"id": "r1", "effect": "permit", "actions": ["read"], "resource": {"type": "a"}},
+            {"id": "r2", "effect": "deny", "actions": ["write"], "resource": {"type": "b"}},
+            {"id": "r3", "effect": "permit", "actions": ["exec"], "resource": {"type": "c"}},
+        ],
+    }
+    guard = Guard(policy)
+    dep = require_batch_access(
+        guard,
+        [("read", "a"), ("write", "b"), ("exec", "c")],
+        lambda req: Subject(id="u"),
+    )
+    decisions = await dep(_fake_request(Subject(id="u")))
+    assert [d.allowed for d in decisions] == [True, False, True]
+
+
+# ---------------------------------------------------------------------------
+# Timeout propagated
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_require_batch_access_timeout_propagated(monkeypatch):
+    """timeout parameter is passed through to evaluate_batch_async."""
+    _patch_fastapi(monkeypatch)
+    from rbacx.adapters.fastapi import require_batch_access
+
+    guard = Guard(_POLICY_MIXED)
+    observed_timeout: list[float | None] = []
+
+    original = guard.evaluate_batch_async
+
+    async def patched(requests, *, timeout=None, **kw):
+        observed_timeout.append(timeout)
+        return await original(requests, timeout=timeout, **kw)
+
+    guard.evaluate_batch_async = patched  # type: ignore[method-assign]
+
+    dep = require_batch_access(
+        guard,
+        [("read", "doc")],
+        lambda req: _S_VIEWER,
+        timeout=2.5,
+    )
+    await dep(_fake_request(_S_VIEWER))
+    assert observed_timeout == [2.5]
+
+
+# ---------------------------------------------------------------------------
+# Coverage: prometheus/otel batch_hist.observe/record paths
+# ---------------------------------------------------------------------------
+
+
+def test_prometheus_batch_size_observe_routing():
+    """PrometheusMetrics routes 'rbacx_batch_size' to _batch_hist."""
+    import importlib
+    import sys
+    import types
+
+    # Install a fake prometheus_client with a Histogram that accepts **kw
+    fake = types.ModuleType("prometheus_client")
+    observed_batch: list[float] = []
+    observed_latency: list[float] = []
+
+    class FakeHist:
+        def __init__(self, name, doc, **kw):
+            self._name = name
+
+        def observe(self, v):
+            if "batch" in self._name:
+                observed_batch.append(v)
+            else:
+                observed_latency.append(v)
+
+    class FakeCnt:
+        def __init__(self, *a, **kw):
+            pass
+
+        def labels(self, **kw):
+            class C:
+                def inc(self):
+                    pass
+
+            return C()
+
+    fake.Counter = FakeCnt
+    fake.Histogram = FakeHist
+    sys.modules["prometheus_client"] = fake
+
+    try:
+        import rbacx.metrics.prometheus as pm_mod
+
+        importlib.reload(pm_mod)
+        m = pm_mod.PrometheusMetrics()
+        m.observe("rbacx_batch_size", 7.0)
+        m.observe("rbacx_decision_seconds", 0.01)
+        assert observed_batch == [7.0]
+        assert observed_latency == [0.01]
+    finally:
+        sys.modules.pop("prometheus_client", None)
+        importlib.reload(pm_mod)
+
+
+def test_otel_batch_size_observe_routing():
+    """OpenTelemetryMetrics routes 'rbacx_batch_size' to _batch_hist."""
+    import importlib
+    import sys
+    import types
+
+    recorded: dict[str, list] = {"batch": [], "latency": []}
+
+    class FakeHist:
+        def __init__(self, name):
+            self._name = name
+
+        def record(self, v, attributes=None):
+            if "batch" in self._name:
+                recorded["batch"].append(v)
+            else:
+                recorded["latency"].append(v)
+
+    class FakeMeter:
+        def create_counter(self, *a, **kw):
+            class C:
+                def add(self, v, attrs=None):
+                    pass
+
+            return C()
+
+        def create_histogram(self, name, **kw):
+            return FakeHist(name)
+
+    fake_otel = types.ModuleType("opentelemetry.metrics")
+    fake_otel.get_meter = lambda name: FakeMeter()
+    sys.modules["opentelemetry"] = types.ModuleType("opentelemetry")
+    sys.modules["opentelemetry.metrics"] = fake_otel
+
+    try:
+        import rbacx.metrics.otel as otel_mod
+
+        importlib.reload(otel_mod)
+        m = otel_mod.OpenTelemetryMetrics()
+        m.observe("rbacx_batch_size", 5.0)
+        m.observe("rbacx_decision_seconds", 0.002)
+        assert recorded["batch"] == [5.0]
+        assert recorded["latency"] == [0.002]
+    finally:
+        sys.modules.pop("opentelemetry", None)
+        sys.modules.pop("opentelemetry.metrics", None)
+        importlib.reload(otel_mod)
+
+
+# ---------------------------------------------------------------------------
+# Coverage: batch_hist is None (prometheus/otel not installed)
+# ---------------------------------------------------------------------------
+
+
+def test_prometheus_batch_size_observe_batch_hist_none():
+    """observe('rbacx_batch_size') is a no-op when _batch_hist is None (99→exit)."""
+    import importlib
+    import sys
+    import types
+
+    fake = types.ModuleType("prometheus_client")
+
+    class FakeCnt:
+        def __init__(self, *a, **kw):
+            pass
+
+        def labels(self, **kw):
+            class C:
+                def inc(self):
+                    pass
+
+            return C()
+
+    class FakeHist:
+        def __init__(self, name, doc, **kw):
+            self._name = name
+
+        def observe(self, v):
+            pass
+
+    fake.Counter = FakeCnt
+    fake.Histogram = FakeHist
+    sys.modules["prometheus_client"] = fake
+
+    try:
+        import rbacx.metrics.prometheus as pm_mod
+
+        importlib.reload(pm_mod)
+        m = pm_mod.PrometheusMetrics()
+        # Force _batch_hist to None to simulate creation failure
+        m._batch_hist = None
+        # Must not raise — silently no-ops (transition 99→exit)
+        m.observe("rbacx_batch_size", 3.0)
+    finally:
+        sys.modules.pop("prometheus_client", None)
+        importlib.reload(pm_mod)
+
+
+def test_otel_batch_size_observe_batch_hist_none():
+    """observe('rbacx_batch_size') is a no-op when _batch_hist is None (117→exit)."""
+    import importlib
+    import sys
+    import types
+
+    class FakeMeter:
+        def create_counter(self, *a, **kw):
+            class C:
+                def add(self, v, attrs=None):
+                    pass
+
+            return C()
+
+        def create_histogram(self, name, **kw):
+            return None
+
+    fake_otel = types.ModuleType("opentelemetry.metrics")
+    fake_otel.get_meter = lambda name: FakeMeter()
+    sys.modules["opentelemetry"] = types.ModuleType("opentelemetry")
+    sys.modules["opentelemetry.metrics"] = fake_otel
+
+    try:
+        import rbacx.metrics.otel as otel_mod
+
+        importlib.reload(otel_mod)
+        m = otel_mod.OpenTelemetryMetrics()
+        # Force _batch_hist to None
+        m._batch_hist = None
+        # Must not raise — silently no-ops (transition 117→exit)
+        m.observe("rbacx_batch_size", 5.0)
+    finally:
+        sys.modules.pop("opentelemetry", None)
+        sys.modules.pop("opentelemetry.metrics", None)
+        importlib.reload(otel_mod)
diff --git a/tests/unit/core/test_engine_cache_key_property.py b/tests/unit/core/test_engine_cache_key_property.py
index 86c7bff..7c164d7 100644
--- a/tests/unit/core/test_engine_cache_key_property.py
+++ b/tests/unit/core/test_engine_cache_key_property.py
@@ -1,7 +1,7 @@
 import pytest
 
 hypothesis = pytest.importorskip("hypothesis")
-from hypothesis import given
+from hypothesis import HealthCheck, given, settings
 from hypothesis import strategies as st
 
 from rbacx.core.engine import Guard
@@ -34,6 +34,7 @@ def make_env(attrs_map):
 
 
 @given(_env_strategy())
+@settings(suppress_health_check=[HealthCheck.too_slow], deadline=None)
 def test_normalization_deterministic_and_stable(env):
     g = Guard({"rules": []})
     s1 = _normalize(g, env)
diff --git a/tests/unit/engine/test_engine_batch_timeout.py b/tests/unit/engine/test_engine_batch_timeout.py
new file mode 100644
index 0000000..add33b0
--- /dev/null
+++ b/tests/unit/engine/test_engine_batch_timeout.py
@@ -0,0 +1,193 @@
+"""Unit tests for evaluate_batch_async/sync timeout parameter."""
+
+import asyncio
+from unittest.mock import patch
+
+import pytest
+
+from rbacx import Action, Context, Guard, Resource, Subject
+
+_POLICY = {
+    "algorithm": "deny-overrides",
+    "rules": [
+        {"id": "r1", "effect": "permit", "actions": ["read"], "resource": {"type": "doc"}},
+    ],
+}
+_S = Subject(id="u1")
+_R = Resource(type="doc", id="1")
+_CTX = Context()
+_REQ = [(_S, Action("read"), _R, _CTX)]
+
+
+# ---------------------------------------------------------------------------
+# timeout=None — default, no deadline
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_async_no_timeout_succeeds():
+    """evaluate_batch_async without timeout completes normally."""
+    guard = Guard(_POLICY)
+    results = await guard.evaluate_batch_async(_REQ, timeout=None)
+    assert len(results) == 1 and results[0].allowed is True
+
+
+def test_batch_sync_no_timeout_succeeds():
+    """evaluate_batch_sync without timeout completes normally."""
+    guard = Guard(_POLICY)
+    results = guard.evaluate_batch_sync(_REQ, timeout=None)
+    assert len(results) == 1 and results[0].allowed is True
+
+
+# ---------------------------------------------------------------------------
+# timeout — normal completion within deadline
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_async_with_generous_timeout_succeeds():
+    """evaluate_batch_async with ample timeout completes normally."""
+    guard = Guard(_POLICY)
+    results = await guard.evaluate_batch_async(_REQ, timeout=10.0)
+    assert len(results) == 1 and results[0].allowed is True
+
+
+def test_batch_sync_with_generous_timeout_succeeds():
+    guard = Guard(_POLICY)
+    results = guard.evaluate_batch_sync(_REQ, timeout=10.0)
+    assert len(results) == 1 and results[0].allowed is True
+
+
+# ---------------------------------------------------------------------------
+# timeout — exceeded raises asyncio.TimeoutError
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_async_timeout_exceeded_raises():
+    """evaluate_batch_async raises TimeoutError when batch exceeds deadline."""
+    guard = Guard(_POLICY)
+
+    async def slow_evaluate(*args, **kwargs):
+        await asyncio.sleep(10)  # simulate slow ReBAC provider
+
+    with patch.object(guard, "_evaluate_core_async", side_effect=slow_evaluate):
+        with pytest.raises(asyncio.TimeoutError):
+            await guard.evaluate_batch_async(_REQ, timeout=0.01)
+
+
+# ---------------------------------------------------------------------------
+# empty batch — always returns [] regardless of timeout
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_async_empty_returns_empty():
+    guard = Guard(_POLICY)
+    assert await guard.evaluate_batch_async([], timeout=0.001) == []
+
+
+def test_batch_sync_empty_returns_empty():
+    guard = Guard(_POLICY)
+    assert guard.evaluate_batch_sync([], timeout=0.001) == []
+
+
+# ---------------------------------------------------------------------------
+# multiple requests — all complete with timeout
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_async_multiple_requests_with_timeout():
+    guard = Guard(_POLICY)
+    reqs = [(_S, Action("read"), _R, _CTX) for _ in range(5)]
+    results = await guard.evaluate_batch_async(reqs, timeout=5.0)
+    assert len(results) == 5
+    assert all(d.allowed for d in results)
+
+
+# ---------------------------------------------------------------------------
+# batch_size metric emitted
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_async_emits_batch_size_metric():
+    """evaluate_batch_async calls metrics.observe('rbacx_batch_size', N)."""
+    observed: list[tuple[str, float]] = []
+
+    class _MockMetrics:
+        def inc(self, name, labels=None):
+            pass
+
+        def observe(self, name, value, labels=None):
+            observed.append((name, value))
+
+    guard = Guard(_POLICY, metrics=_MockMetrics())
+    reqs = [(_S, Action("read"), _R, _CTX) for _ in range(3)]
+    await guard.evaluate_batch_async(reqs)
+
+    batch_obs = [(n, v) for n, v in observed if n == "rbacx_batch_size"]
+    assert len(batch_obs) == 1
+    assert batch_obs[0][1] == 3.0
+
+
+@pytest.mark.asyncio
+async def test_batch_async_empty_does_not_emit_metric():
+    """evaluate_batch_async with empty input returns early — no metric emitted."""
+    observed: list[tuple[str, float]] = []
+
+    class _MockMetrics:
+        def inc(self, name, labels=None):
+            pass
+
+        def observe(self, name, value, labels=None):
+            observed.append((name, value))
+
+    guard = Guard(_POLICY, metrics=_MockMetrics())
+    await guard.evaluate_batch_async([])
+
+    batch_obs = [(n, v) for n, v in observed if n == "rbacx_batch_size"]
+    assert batch_obs == []
+
+
+# ---------------------------------------------------------------------------
+# Coverage: async metrics.observe + exception in observe
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_async_async_metrics_observe():
+    """When metrics.observe is a coroutine, it is awaited correctly."""
+    observed: list[tuple[str, float]] = []
+
+    class _AsyncMetrics:
+        def inc(self, name, labels=None):
+            pass
+
+        async def observe(self, name, value, labels=None):
+            observed.append((name, value))
+
+    guard = Guard(_POLICY, metrics=_AsyncMetrics())
+    reqs = [(_S, Action("read"), _R, _CTX), (_S, Action("read"), _R, _CTX)]
+    await guard.evaluate_batch_async(reqs)
+
+    batch_obs = [(n, v) for n, v in observed if n == "rbacx_batch_size"]
+    assert batch_obs == [("rbacx_batch_size", 2.0)]
+
+
+@pytest.mark.asyncio
+async def test_batch_async_metrics_observe_exception_swallowed():
+    """Exception in metrics.observe does not propagate — batch result returned."""
+
+    class _RaisingMetrics:
+        def inc(self, name, labels=None):
+            pass
+
+        def observe(self, name, value, labels=None):
+            raise RuntimeError("metrics backend down")
+
+    guard = Guard(_POLICY, metrics=_RaisingMetrics())
+    # Must not raise — exception is swallowed and logged
+    results = await guard.evaluate_batch_async(_REQ)
+    assert len(results) == 1 and results[0].allowed is True
diff --git a/tests/unit/rebac/test_openfga_checker.py b/tests/unit/rebac/test_openfga_checker.py
index 73abd0a..7e69536 100644
--- a/tests/unit/rebac/test_openfga_checker.py
+++ b/tests/unit/rebac/test_openfga_checker.py
@@ -108,3 +108,15 @@ async def test_async_http_error_branch_returns_falses():
     cli = ofga.OpenFGAChecker(cfg, async_client=ofga.httpx.AsyncClient())  # type: ignore[attr-defined]
     out = await cli.batch_check([("u:1", "r", "o:1"), ("u:2", "r", "o:2"), ("u:3", "r", "o:3")])
     assert out == [False, False, False]
+
+
+def test_sync_batch_check_empty_returns_empty(monkeypatch):
+    """sync batch_check([]) returns [] immediately without HTTP call (line 147)."""
+    import sys
+
+    sys.modules["httpx"] = make_httpx(ok=True)
+    ofga = importlib.reload(importlib.import_module("rbacx.rebac.openfga"))
+    cfg = ofga.OpenFGAConfig(api_url="http://api", store_id="s")
+    cli = ofga.OpenFGAChecker(cfg, client=ofga.httpx.Client())  # type: ignore[attr-defined]
+    result = cli.batch_check([])
+    assert result == []
diff --git a/tests/unit/rebac/test_spicedb_batch_native.py b/tests/unit/rebac/test_spicedb_batch_native.py
new file mode 100644
index 0000000..8032ad9
--- /dev/null
+++ b/tests/unit/rebac/test_spicedb_batch_native.py
@@ -0,0 +1,206 @@
+"""Unit tests for SpiceDB native BulkCheckPermissions batch_check."""
+
+import importlib
+import importlib.util
+from unittest.mock import MagicMock
+
+import pytest
+
+for _mod in ("authzed", "grpc", "google.protobuf"):
+    if importlib.util.find_spec(_mod) is None:
+        pytest.skip(
+            f"optional dependency '{_mod}' not installed; skipping SpiceDB tests",
+            allow_module_level=True,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_pair(permissionship):
+    """Build a minimal BulkCheckPermissionPair-like object."""
+    pair = MagicMock()
+    pair.item.permissionship = permissionship
+    return pair
+
+
+def _has_permission():
+    from authzed.api.v1 import CheckPermissionResponse
+
+    return CheckPermissionResponse.PERMISSIONSHIP_HAS_PERMISSION
+
+
+def _no_permission():
+    from authzed.api.v1 import CheckPermissionResponse
+
+    return CheckPermissionResponse.PERMISSIONSHIP_NO_PERMISSION
+
+
+# ---------------------------------------------------------------------------
+# Empty triples — early return
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_check_empty_returns_empty():
+    sp = importlib.import_module("rbacx.rebac.spicedb")
+    cfg = sp.SpiceDBConfig(endpoint="e", token="t", insecure=False)
+    checker = sp.SpiceDBChecker(cfg, async_mode=True)
+    # batch_check([]) returns [] immediately (not a coroutine) even in async mode
+    result = checker.batch_check([])
+    assert result == []
+
+
+def test_batch_check_sync_empty_returns_empty():
+    sp = importlib.import_module("rbacx.rebac.spicedb")
+    cfg = sp.SpiceDBConfig(endpoint="localhost:50051", token=None, insecure=True)
+    checker = sp.SpiceDBChecker(cfg)
+    result = checker.batch_check([])
+    assert result == []
+
+
+# ---------------------------------------------------------------------------
+# Async mode — uses BulkCheckPermissions
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_check_async_native_bulk(monkeypatch):
+    """async batch_check calls BulkCheckPermissions once (not N CheckPermissions)."""
+    sp = importlib.import_module("rbacx.rebac.spicedb")
+    cfg = sp.SpiceDBConfig(endpoint="e", token="t", insecure=False)
+    checker = sp.SpiceDBChecker(cfg, async_mode=True)
+
+    bulk_calls: list = []
+
+    async def fake_bulk(req, timeout=None):
+        bulk_calls.append(req)
+        resp = MagicMock()
+        resp.pairs = [_make_pair(_has_permission()), _make_pair(_no_permission())]
+        return resp
+
+    # Attach BulkCheckPermissions to the client (may not exist on older authzed)
+    checker._aclient.BulkCheckPermissions = fake_bulk
+
+    triples = [("user:1", "viewer", "doc:1"), ("user:2", "viewer", "doc:2")]
+    result = await checker.batch_check(triples)
+
+    assert len(bulk_calls) == 1  # single gRPC call
+    assert result == [True, False]
+
+
+@pytest.mark.asyncio
+async def test_batch_check_async_rpc_error_returns_falses(monkeypatch):
+    """RPC error on BulkCheckPermissions returns [False] * N."""
+    sp = importlib.import_module("rbacx.rebac.spicedb")
+    import grpc
+
+    cfg = sp.SpiceDBConfig(endpoint="e", token="t", insecure=False)
+    checker = sp.SpiceDBChecker(cfg, async_mode=True)
+
+    async def boom(req, timeout=None):
+        raise grpc.RpcError("network failure")
+
+    checker._aclient.BulkCheckPermissions = boom
+
+    triples = [("u:1", "r", "o:1"), ("u:2", "r", "o:2"), ("u:3", "r", "o:3")]
+    result = await checker.batch_check(triples)
+    assert result == [False, False, False]
+
+
+# ---------------------------------------------------------------------------
+# Sync mode — sequential fallback (no bulk endpoint on sync client)
+# ---------------------------------------------------------------------------
+
+
+def test_batch_check_sync_sequential_fallback(monkeypatch):
+    """sync batch_check falls back to sequential CheckPermission calls."""
+    sp = importlib.import_module("rbacx.rebac.spicedb")
+    from authzed.api.v1 import CheckPermissionResponse
+
+    cfg = sp.SpiceDBConfig(endpoint="localhost:50051", token=None, insecure=True)
+    checker = sp.SpiceDBChecker(cfg)
+
+    call_count = [0]
+
+    def fake_check(req, timeout=None):
+        call_count[0] += 1
+        r = CheckPermissionResponse()
+        r.permissionship = CheckPermissionResponse.PERMISSIONSHIP_HAS_PERMISSION
+        return r
+
+    monkeypatch.setattr(checker._client, "CheckPermission", fake_check, raising=True)
+
+    triples = [("u:1", "r", "o:1"), ("u:2", "r", "o:2")]
+    result = checker.batch_check(triples)
+
+    assert call_count[0] == 2  # two separate calls
+    assert result == [True, True]
+
+
+# ---------------------------------------------------------------------------
+# Coverage: zed_token, prefer_fully_consistent, except Exception in BulkCheck
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_batch_check_async_with_zed_token(monkeypatch):
+    """zed_token is forwarded to BulkCheckPermissions consistency field (line 197)."""
+    sp = importlib.import_module("rbacx.rebac.spicedb")
+    cfg = sp.SpiceDBConfig(endpoint="e", token="t", insecure=False)
+    checker = sp.SpiceDBChecker(cfg, async_mode=True)
+
+    captured_req = []
+
+    async def fake_bulk(req, timeout=None):
+        captured_req.append(req)
+        resp = MagicMock()
+        resp.pairs = [_make_pair(_has_permission())]
+        return resp
+
+    checker._aclient.BulkCheckPermissions = fake_bulk
+
+    await checker.batch_check([("user:1", "viewer", "doc:1")], zed_token="Z123")
+    assert len(captured_req) == 1
+    # Consistency must be set (not None)
+    assert captured_req[0].consistency is not None
+
+
+@pytest.mark.asyncio
+async def test_batch_check_async_prefer_fully_consistent(monkeypatch):
+    """prefer_fully_consistent=True sets consistency on the bulk request (line 201)."""
+    sp = importlib.import_module("rbacx.rebac.spicedb")
+    cfg = sp.SpiceDBConfig(endpoint="e", token="t", insecure=False, prefer_fully_consistent=True)
+    checker = sp.SpiceDBChecker(cfg, async_mode=True)
+
+    captured_req = []
+
+    async def fake_bulk(req, timeout=None):
+        captured_req.append(req)
+        resp = MagicMock()
+        resp.pairs = [_make_pair(_no_permission())]
+        return resp
+
+    checker._aclient.BulkCheckPermissions = fake_bulk
+
+    await checker.batch_check([("user:1", "viewer", "doc:1")])
+    assert captured_req[0].consistency is not None
+
+
+@pytest.mark.asyncio
+async def test_batch_check_async_unexpected_exception_returns_falses(monkeypatch):
+    """Generic Exception in BulkCheckPermissions returns [False]*N (lines 258-262)."""
+    sp = importlib.import_module("rbacx.rebac.spicedb")
+    cfg = sp.SpiceDBConfig(endpoint="e", token="t", insecure=False)
+    checker = sp.SpiceDBChecker(cfg, async_mode=True)
+
+    async def boom(req, timeout=None):
+        raise ValueError("unexpected error")
+
+    checker._aclient.BulkCheckPermissions = boom
+
+    triples = [("u:1", "r", "o:1"), ("u:2", "r", "o:2")]
+    result = await checker.batch_check(triples)
+    assert result == [False, False]