Snuffy2 · Snuffy2 · Jul 19, 2026 · Jul 19, 2026 · Jul 19, 2026 · Jul 19, 2026
diff --git a/aiopnsense/__init__.py b/aiopnsense/__init__.py
@@ -1,6 +1,7 @@
 """aiopnsense package to manage OPNsense."""
 
 from .client import OPNsenseClient
+from ._typing import CategoryResult, CategoryState
 from .exceptions import (
     OPNsenseBelowMinFirmware,
     OPNsenseConnectionError,
@@ -17,6 +18,8 @@
 )
 
 __all__ = [
+    "CategoryResult",
+    "CategoryState",
     "OPNsenseBelowMinFirmware",
     "OPNsenseClient",
     "OPNsenseConnectionError",

diff --git a/aiopnsense/_typing.py b/aiopnsense/_typing.py
@@ -1,17 +1,82 @@
 """Typing protocol contracts for aiopnsense mixins."""
 
+import asyncio
 from collections.abc import AsyncGenerator, MutableMapping
+from contextvars import ContextVar
+from dataclasses import dataclass
 from datetime import tzinfo
-from typing import Any, Protocol
+from typing import Any, Iterator, Literal, Protocol
-from typing import Any, Iterator, Literal, Protocol
+from collections.abc import AsyncGenerator, Iterator, MutableMapping
+from typing import Any, Literal, Protocol
-from typing import Any, Iterator, Literal, Protocol
+from collections.abc import AsyncGenerator, Iterator, MutableMapping
+from typing import Any, Literal, Protocol
+
+
+type CategoryState = Literal["available", "pending", "missing", "transient", "malformed"]
+EndpointAvailabilityState = Literal["available", "missing", "pending"]
+
+
+@dataclass(frozen=True, slots=True)
+class CategoryResult[T]:
+    """Immutable data and availability result for an optional API category."""
+
+    data: T
+    state: CategoryState
+    authoritative: bool
+
+    def __post_init__(self) -> None:
+        """Reject envelopes whose authority contradicts their state."""
+        if self.authoritative is not (self.state == "available"):
+            raise ValueError("authoritative must be true exactly when state is 'available'")
+
+    @staticmethod
+    def coerce(value: object) -> "CategoryResult[object]":
+        """Normalize legacy internal tuple results during the contract migration."""
+        if isinstance(value, CategoryResult):
+            return value
+        if isinstance(value, tuple) and len(value) == 2 and isinstance(value[0], str):
+            state: CategoryState | str = "transient" if value[0] == "unavailable" else value[0]
+            if state in {"available", "pending", "missing", "transient", "malformed"}:
+                typed_state: CategoryState = state
+                return CategoryResult(value[1], typed_state, typed_state == "available")
+        return CategoryResult({}, "malformed", False)
+
+    def __iter__(self) -> Iterator[object]:
+        """Yield legacy state/data tuple values for internal compatibility."""
+        yield self.state
+        yield self.data
+
+    def __eq__(self, other: object) -> bool:
+        """Compare result objects, with temporary support for legacy tuples."""
+        if isinstance(other, CategoryResult):
+            return (
+                self.data == other.data
+                and self.state == other.state
+                and self.authoritative == other.authoritative
+            )
+        if isinstance(other, tuple) and len(other) == 2:
+            legacy_state = "transient" if other[0] == "unavailable" else other[0]
+            return self.state == legacy_state and self.data == other[1]
+        return NotImplemented
 
 
 class AiopnsenseClientProtocol(Protocol):
     """Structural typing contract used by split aiopnsense mixins."""
 
     _throw_errors: bool
+    _use_snake_case: bool | None
+    _endpoint_availability: dict[tuple[Literal["get", "post"], str], EndpointAvailabilityState]
+    _endpoint_checked_at: dict[tuple[Literal["get", "post"], str], float]
+    _endpoint_locks: dict[tuple[Literal["get", "post"], str], asyncio.Lock]
+    _optional_endpoint_missing_pending_confirmation: set[tuple[Literal["get", "post"], str]]
+    _dhcp_source_states_context: ContextVar[list[CategoryState] | None]
 
     async def _get(self, path: str) -> MutableMapping[str, Any] | list | None: ...
 
+    async def _get_optional(self, path: str) -> CategoryResult[object]: ...
+
+    async def _post_optional(
+        self,
+        path: str,
+        payload: MutableMapping[str, Any] | None = None,
+    ) -> CategoryResult[object]: ...
+
     async def _get_text(self, path: str) -> str | None: ...
 
     async def _post(
@@ -57,3 +122,40 @@ async def _is_get_endpoint_available(self, path: str, force_refresh: bool = Fals
     async def _is_post_endpoint_available(
         self, path: str, force_refresh: bool = False
     ) -> bool | None: ...
+
+    async def _check_optional_get_endpoint(
+        self,
+        path: str,
+        cache_path: str | None = None,
+        *,
+        force_refresh: bool = False,
+    ) -> CategoryResult[object]: ...
+
+    async def _check_optional_post_endpoint(
+        self,
+        path: str,
+        payload: MutableMapping[str, Any] | None = None,
+        cache_path: str | None = None,
+        *,
+        force_refresh: bool = False,
+    ) -> CategoryResult[object]: ...
+
+    async def get_smart_result(self) -> CategoryResult[list[dict[str, Any]]]: ...
+
+    async def get_smart_info_result(
+        self, device: str, info_type: str = "a"
+    ) -> CategoryResult[dict[str, Any]]: ...
+
+    async def get_nut_ups_status_result(self) -> CategoryResult[dict[str, Any]]: ...
+
+    async def get_vnstat_result(self) -> CategoryResult[MutableMapping[str, Any]]: ...
+
+    async def get_unbound_blocklist_result(self) -> CategoryResult[dict[str, Any]]: ...
+
+    async def get_dhcp_leases_result(
+        self, opnsense_tz: tzinfo | None = None
+    ) -> CategoryResult[dict[str, Any]]: ...
+
+    async def get_arp_table_result(
+        self, resolve_hostnames: bool = False
+    ) -> CategoryResult[list[dict[str, Any]]]: ...
diff --git a/aiopnsense/client_base.py b/aiopnsense/client_base.py
@@ -2,8 +2,8 @@
 
 import asyncio
 from collections.abc import MutableMapping
-from datetime import datetime
-from typing import Any
+from contextvars import ContextVar
+from typing import Any, Literal
 from urllib.parse import urlparse
 import warnings
 
@@ -12,8 +12,9 @@
 from .client_endpoint import ClientEndpointMixin
 from .client_queue import ClientQueueMixin
 from .client_transport import ClientTransportMixin
-from .const import DEFAULT_CACHE_TTL_SECONDS
+from .const import DEFAULT_CACHE_TTL_SECONDS, DEFAULT_NEGATIVE_CACHE_TTL_SECONDS
 from .exceptions import OPNsenseInvalidArgument
+from ._typing import CategoryState, EndpointAvailabilityState
 
 _UNSET: object = object()
 
@@ -41,7 +42,8 @@ def __init__(
             password (str): Password for API authentication.
             session (aiohttp.ClientSession): HTTP client session used for API requests.
             opts (MutableMapping[str, Any] | None, optional): Optional client configuration values
-                (e.g. ``opts={"verify_ssl": True}``).
+                such as ``verify_ssl``, ``endpoint_positive_cache_ttl_seconds``,
+                and ``endpoint_negative_cache_ttl_seconds``.
             initial (bool | object): Deprecated alias for ``throw_errors``. When provided,
                 a ``DeprecationWarning`` is emitted. Ignored when ``throw_errors`` is also set.
             throw_errors (bool | object): Whether request and decorator errors should be
@@ -79,9 +81,33 @@ def __init__(
                 self._throw_errors = initial
         self._firmware_version: str | None = None
         self._use_snake_case: bool | None = None
-        self._endpoint_availability: dict[str, bool] = {}
-        self._endpoint_checked_at: dict[str, datetime] = {}
-        self._endpoint_cache_ttl_seconds = DEFAULT_CACHE_TTL_SECONDS
+        self._endpoint_availability: dict[
+            tuple[Literal["get", "post"], str], EndpointAvailabilityState
+        ] = {}
+        self._endpoint_checked_at: dict[tuple[Literal["get", "post"], str], float] = {}
+        self._endpoint_locks: dict[tuple[Literal["get", "post"], str], asyncio.Lock] = {}
+        self._optional_endpoint_missing_pending_confirmation: set[
+            tuple[Literal["get", "post"], str]
+        ] = set()
+        self._dhcp_source_states_context: ContextVar[list[CategoryState] | None] = ContextVar(
+            "dhcp_source_states", default=None
+        )
+        positive_ttl = self._opts.get(
+            "endpoint_positive_cache_ttl_seconds", DEFAULT_CACHE_TTL_SECONDS
+        )
+        negative_ttl = self._opts.get(
+            "endpoint_negative_cache_ttl_seconds", DEFAULT_NEGATIVE_CACHE_TTL_SECONDS
+        )
+        if not isinstance(positive_ttl, int) or isinstance(positive_ttl, bool) or positive_ttl <= 0:
+            raise OPNsenseInvalidArgument(
+                "`endpoint_positive_cache_ttl_seconds` must be a positive integer."
+            )
+        if not isinstance(negative_ttl, int) or isinstance(negative_ttl, bool) or negative_ttl <= 0:
+            raise OPNsenseInvalidArgument(
+                "`endpoint_negative_cache_ttl_seconds` must be a positive integer."
+            )
+        self._endpoint_cache_ttl_seconds = positive_ttl
+        self._endpoint_negative_cache_ttl_seconds = negative_ttl
         self._rest_api_query_count = 0
         self._request_queue: asyncio.Queue = asyncio.Queue()
         self._workers: list[asyncio.Task[Any]] = []