feat(backend): bound L1 backfill by the server's remaining freshness (LAB-557)

docs/configuration.md

@@ -192,6 +192,7 @@ Rules and behavior:
 - A failed background recompute is silent: the entry keeps serving stale until its hard eviction bound, after which the next call takes the ordinary synchronous miss path.
 - The background recompute runs with a **snapshot of the caller's `contextvars`** (contextvar-based tenant extraction works), but outside the request otherwise — don't rely on other request-scoped resources (open sessions, connections) inside functions that enable SWR.
 - Stale values are never written to the L1 in-memory cache, and stale reads still count as cache **hits** for metered-misses billing.
+- On the CachekitIO backend, every read (SWR-configured or not) also carries the server's remaining freshness (`X-CacheKit-Fresh-For`, [protocol spec](https://github.com/cachekit-io/protocol/blob/main/spec/saas-api.md#remaining-freshness)): an L2 hit backfilled into L1 lives at most `min(ttl, remaining)` locally, so a value read near the end of its server-side freshness window is never served fresh from L1 past the server's bound. Pre-signal servers omit the header and behavior is unchanged.
 
 ### File Backend Environment Variables
 

src/cachekit/backends/cachekitio/backend.py

@@ -44,9 +44,12 @@
 # Stale-while-revalidate (LAB-381, spec/saas-api.md#stale-while-revalidate).
 # STALE_TTL_HEADER rides PUTs to open a stale-grace window past the fresh TTL;
 # FRESHNESS_HEADER labels every GET/HEAD 200 as fresh|stale. Pre-SWR servers
-# ignore the former and never emit the latter.
+# ignore the former and never emit the latter. FRESH_FOR_HEADER carries the
+# remaining freshness in whole seconds on GET 200s (LAB-557,
+# spec/saas-api.md#remaining-freshness); pre-signal servers omit it.
 STALE_TTL_HEADER = "X-CacheKit-Stale-TTL"
 FRESHNESS_HEADER = "X-CacheKit-Freshness"
+FRESH_FOR_HEADER = "X-CacheKit-Fresh-For"
 
 
 def _inject_metrics_headers(stats: _FunctionStats | None) -> dict[str, str]:
@@ -385,19 +388,42 @@ def _is_stale(response: httpx.Response) -> bool:
         value = response.headers.get(FRESHNESS_HEADER)
         return value is not None and value != "fresh"
 
-    def get_with_freshness(self, key: str) -> tuple[bytes, bool] | None:
-        """Retrieve value plus its SWR freshness (sync).
+    @staticmethod
+    def _fresh_for(response: httpx.Response) -> int | None:
+        """Parse X-CacheKit-Fresh-For (LAB-557, spec/saas-api.md#remaining-freshness).
+
+        Absent = pre-signal server → None (legacy behavior: no bound).
+        Unparseable or negative = drift → 0 (do not extend local service — the
+        conservative action, mirroring the unrecognized-freshness → stale rule).
+        """
+        value = response.headers.get(FRESH_FOR_HEADER)
+        if value is None:
+            return None
+        try:
+            parsed = int(value)
+        except ValueError:
+            # Drift signal, not a crash: a server/proxy emitting garbage here
+            # disables L1 backfill for affected reads — log so a fleet-wide
+            # latency regression is diagnosable (expert-panel finding).
+            _logger.debug(f"Unparseable {FRESH_FOR_HEADER} header {value!r}; treating as 0 (no L1 backfill)")
+            return 0
+        return parsed if parsed >= 0 else 0
+
+    def get_with_freshness(self, key: str) -> tuple[bytes, bool, int | None] | None:
+        """Retrieve value plus its SWR freshness and remaining-freshness bound (sync).
 
         Returns:
-            ``(value, is_stale)`` on a hit — ``is_stale`` is True only for an
-            entry in its stale-grace window (LAB-381) — or None on a miss.
+            ``(value, is_stale, fresh_for)`` on a hit — ``is_stale`` is True only
+            for an entry in its stale-grace window (LAB-381); ``fresh_for`` is the
+            server's remaining freshness in seconds, or None from a pre-signal
+            server (LAB-557) — or None on a miss.
 
         Raises:
             BackendError: If operation fails (network, auth, etc.)
         """
         try:
             response = self._request_sync("GET", self._encode_key(key))
-            return response.content, self._is_stale(response)
+            return response.content, self._is_stale(response), self._fresh_for(response)
         except BackendError as exc:
             if exc.original_exception and isinstance(exc.original_exception, httpx.HTTPStatusError):
                 if exc.original_exception.response.status_code == 404:

src/cachekit/cache_handler.py

@@ -243,19 +243,29 @@ def supports_streaming_write(backend: BaseBackend) -> TypeGuard[BufferWritableBa
 class SWRCapableBackend(Protocol):
     """Backend with server-signaled stale-while-revalidate reads (LAB-381).
 
-    Reads report whether the entry is in its stale-grace window; writes accept
-    the window length. Currently only CachekitIOBackend (the SaaS signals
-    freshness on read — see protocol spec/saas-api.md#stale-while-revalidate).
+    Reads report whether the entry is in its stale-grace window plus the
+    server's remaining freshness in seconds (LAB-557; None from a pre-signal
+    server); writes accept the window length. Currently only CachekitIOBackend
+    (the SaaS signals freshness on read — see protocol
+    spec/saas-api.md#stale-while-revalidate and #remaining-freshness).
     """
 
-    def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool]]: ...
+    def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]: ...
 
     def set(self, key: str, value: bytes, ttl: Optional[int] = None, stale_ttl: Optional[int] = None) -> None: ...
 
 
 def supports_swr(backend: BaseBackend) -> TypeGuard[SWRCapableBackend]:
-    """Type guard: backend supports server-signaled SWR stale-grace reads (LAB-381)."""
-    return hasattr(backend, "get_with_freshness")
+    """Type guard: backend supports server-signaled SWR stale-grace reads (LAB-381).
+
+    Checked on the backend's CLASS, not the instance: protocol methods live on
+    classes, while instance-level hasattr reads dynamic-attribute objects
+    (unittest.mock.Mock, __getattr__ proxies) as SWR-capable and silently
+    reroutes their reads through the freshness path — reachable without any SWR
+    config since the LAB-557 gate widening (the freshness read now runs for
+    every capable backend, not only when stale_ttl is set).
+    """
+    return callable(getattr(type(backend), "get_with_freshness", None))
 
 
 # Import caching for serializer modules
@@ -1385,10 +1395,13 @@ def get_cached_value(self, cache_key: str, refresh_ttl: Optional[int] = None) ->
             get_logger().warning(f"Backend operation failed for get on {cache_key}: {e}")
             return None
 
-    def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tuple[bool, Any], bool]]:
-        """SWR variant of :meth:`get_cached_value` (LAB-381): also reports staleness.
+    def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tuple[bool, Any], bool, Optional[int]]]:
+        """SWR variant of :meth:`get_cached_value` (LAB-381/LAB-557): also reports
+        staleness and the server's remaining freshness in seconds.
 
-        Returns ``((True, value), is_stale)`` on a hit, None on miss/error. The mmap
+        Returns ``((True, value), is_stale, fresh_for)`` on a hit, None on
+        miss/error. fresh_for is None when no signal exists (pre-signal server,
+        non-SWR backend) — the caller applies legacy L1 TTL behavior. The mmap
         fast path is skipped — SWR is CachekitIO-only, which is not buffer-readable.
         Error semantics mirror get_cached_value: the LAB-108 policy point raises
         DecryptionAuthenticationError when fail-closed (poisoned entry retained as
@@ -1401,10 +1414,16 @@ def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tupl
             hit = self._cache_handler.get_with_freshness(cache_key)
             if hit is None:
                 return None
-            cached_data, is_stale = hit
+            # Length-tolerant unpack: a third-party SWR backend built against the
+            # released 2-tuple (bytes, is_stale) signature must degrade to
+            # fresh_for=None (legacy L1 behavior), not have a strict 3-unpack
+            # ValueError swallowed below into a permanent every-hit-is-a-miss
+            # cache bypass (expert-panel finding, LAB-557).
+            cached_data, is_stale, *_rest = hit
+            fresh_for = _rest[0] if _rest else None
             get_logger().cache_hit(cache_key, "Backend(stale)" if is_stale else "Backend")
             deserialized = self.serialization_handler.deserialize_data(cached_data, cache_key)
-            return ((True, deserialized), is_stale)
+            return ((True, deserialized), is_stale, fresh_for)
         except KeyringConfigurationError:
             # LOCAL keyring config fault (bad tenant_id, bad keyring entry index) —
             # never a legitimate miss, and not tamper. Re-raised past the broad
@@ -1422,13 +1441,18 @@ def get_cached_value_with_freshness(self, cache_key: str) -> Optional[tuple[tupl
             get_logger().warning(f"Backend operation failed for get on {redact_cache_key(cache_key)}: {e}")
             return None
 
-    async def get_cached_value_with_freshness_async(self, cache_key: str) -> Optional[tuple[tuple[bool, Any, bytes], bool]]:
-        """Async SWR variant (LAB-381): staleness + the raw envelope for L1 backfill.
-
-        Returns ``((True, value, raw_bytes), is_stale)`` on a hit — the 3-tuple
-        matches :meth:`get_cached_value_async` (LAB-111 routing) so the async
-        decorator backfills L1 without re-serializing. None on miss/error; the
-        LAB-108 fail-closed policy propagates DecryptionAuthenticationError.
+    async def get_cached_value_with_freshness_async(
+        self, cache_key: str
+    ) -> Optional[tuple[tuple[bool, Any, bytes], bool, Optional[int]]]:
+        """Async SWR variant (LAB-381/LAB-557): staleness + remaining freshness +
+        the raw envelope for L1 backfill.
+
+        Returns ``((True, value, raw_bytes), is_stale, fresh_for)`` on a hit —
+        the inner 3-tuple matches :meth:`get_cached_value_async` (LAB-111
+        routing) so the async decorator backfills L1 without re-serializing;
+        fresh_for (seconds, None = no signal) bounds that backfill to the
+        server's remaining freshness. None on miss/error; the LAB-108
+        fail-closed policy propagates DecryptionAuthenticationError.
         """
         try:
             if self._cache_handler is None:
@@ -1437,10 +1461,13 @@ async def get_cached_value_with_freshness_async(self, cache_key: str) -> Optiona
             hit = await self._cache_handler.get_with_freshness_async(cache_key)
             if hit is None:
                 return None
-            cached_data, is_stale = hit
+            # Length-tolerant unpack — same 2-tuple compatibility contract as the
+            # sync variant above.
+            cached_data, is_stale, *_rest = hit
+            fresh_for = _rest[0] if _rest else None
             get_logger().cache_hit(cache_key, "Backend(stale)" if is_stale else "Backend")
             deserialized = self.serialization_handler.deserialize_data(cached_data, cache_key)
-            return ((True, deserialized, cached_data), is_stale)
+            return ((True, deserialized, cached_data), is_stale, fresh_for)
         except KeyringConfigurationError:
             # LOCAL keyring config fault (bad tenant_id, bad keyring entry index) —
             # never a legitimate miss, and not tamper. Re-raised past the broad
@@ -1793,11 +1820,13 @@ async def delete_async(self, key: str) -> bool:
         """Delete key from cache asynchronously."""
         ...
 
-    def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool]]:
-        """Get value plus SWR staleness (LAB-381); (bytes, is_stale) or None."""
+    def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]:
+        """Get value plus SWR staleness and remaining freshness (LAB-381/LAB-557);
+        (bytes, is_stale, fresh_for) or None. fresh_for is None from a pre-signal
+        server or a non-SWR backend."""
         ...
 
-    async def get_with_freshness_async(self, key: str) -> Optional[tuple[bytes, bool]]:
+    async def get_with_freshness_async(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]:
         """Async variant of get_with_freshness."""
         ...
 
@@ -1964,16 +1993,17 @@ def get_buffer(self, key: str) -> Optional[BufferHandle]:
             get_logger().error(f"Unexpected error mmapping key {key}: {e}")
             return None
 
-    def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool]]:
-        """Get value plus SWR staleness from an SWR-capable backend (LAB-381).
+    def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]:
+        """Get value plus SWR staleness and remaining freshness (LAB-381/LAB-557).
 
-        Returns ``(bytes, is_stale)`` on a hit, or None on miss/error (same
-        degradation contract as :meth:`get` — an error reads as a miss and the
-        caller takes the synchronous recompute path).
+        Returns ``(bytes, is_stale, fresh_for)`` on a hit, or None on miss/error
+        (same degradation contract as :meth:`get` — an error reads as a miss and
+        the caller takes the synchronous recompute path). Non-SWR backends read
+        as ``(bytes, False, None)`` — no freshness signal, legacy L1 behavior.
         """
         if not supports_swr(self.backend):
             value = self.get(key)
-            return (value, False) if value is not None else None
+            return (value, False, None) if value is not None else None
         try:
             return self._with_backpressure_and_timeout(self.backend.get_with_freshness, key)
         except BackendError as e:
@@ -1983,11 +2013,11 @@ def get_with_freshness(self, key: str) -> Optional[tuple[bytes, bool]]:
             get_logger().error(f"Unexpected error getting key {redact_cache_key(key)}: {e}")
             return None
 
-    async def get_with_freshness_async(self, key: str) -> Optional[tuple[bytes, bool]]:
+    async def get_with_freshness_async(self, key: str) -> Optional[tuple[bytes, bool, Optional[int]]]:
         """Async variant of :meth:`get_with_freshness` (sync backend call in the thread pool)."""
         if not supports_swr(self.backend):
             value = await self.get_async(key)
-            return (value, False) if value is not None else None
+            return (value, False, None) if value is not None else None
         try:
             return await self._with_backpressure_and_timeout_async(self.backend.get_with_freshness, key)
         except BackendError as e: