feat(snapshots): add a lifetime scope, because a TTL straddles two moments

snapshot_cache_ttl shipped in 0.1.5 taking only seconds, and seconds are
the wrong shape for the case it was added for. A run that has pinned one
point in time does not want that answer to change underneath it, and a
TTL expiring mid-run lets the next resolution return a different
snapshot. Nothing raises, and the data is real on both sides of the
expiry; it is simply from two different times. That is the same failure
this area keeps producing, an answer that looks entirely plausible.

"lifetime" resolves once and holds for the life of the client, which
matches the invariant a sync actually relies on: it pinned one snapshot,
and snapshots are immutable. It still yields to an upload through the
same client, because that is an event that makes the answer wrong rather
than merely old, so lifetime means "until something makes it wrong".

Seconds stay right for repeated independent reads, and the docs now say
which shape suits which caller instead of leaving it to be discovered.

Suggested by the Nautobot integration, whose own snapshot cache is
lifetime-scoped for exactly this reason and who pointed out that a TTL
quietly breaks the invariant while looking like a performance knob.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019Kbxab7ihjPntGAmnsZtqG

Author: captainpacket

CHANGELOG.md

@@ -3,6 +3,23 @@
 All notable changes to this project are documented here. The format follows
 [Keep a Changelog](https://keepachangelog.com/) and the project uses SemVer.
 
+## [Unreleased]
+
+### Added
+
+- `snapshot_cache_ttl="lifetime"`, which resolves a snapshot once and keeps that
+  answer for the life of the client. A number of seconds was the wrong shape for
+  the case the setting was added for: a run that has pinned one point in time
+  does not want the answer to change underneath it, and a TTL expiring mid-run
+  lets the next resolution return a different snapshot, so the run straddles two
+  moments with nothing raising and the data real on both sides.
+
+  Seconds remain right for repeated independent reads, and the documentation now
+  says which shape suits which caller rather than leaving it to be discovered.
+  `"lifetime"` still yields to an upload through the same client, because that is
+  an event that makes the answer wrong rather than merely old. Suggested by the
+  Nautobot integration, whose own cache is lifetime-scoped for this reason.
+
 ## [0.1.5] - 2026-09-09
 
 ### Changed

docs/configuration.md

@@ -145,21 +145,40 @@ that itself. Nothing failed and no test noticed. It showed up only as request
 volume against a live account.
 
 If you know an invariant the SDK cannot assume, tell it, with
-`snapshot_cache_ttl`. It is off by default and takes a number of seconds:
+`snapshot_cache_ttl`. It is off by default and takes two shapes, and which one
+you want follows from what your run is doing.
+
+**A run that holds one point in time wants `"lifetime"`.** It resolves once and
+keeps that answer for as long as the client lives:
+
+```python
+# This sync pins one snapshot for its whole run, and snapshots are immutable,
+# so the answer must not change underneath it.
+client = ForwardClient.from_env(snapshot_cache_ttl="lifetime")
+```
+
+**Repeated independent reads want a number of seconds.** A dashboard, a
+long-lived service answering unrelated questions, anything where "latest" should
+eventually mean something newer:
 
 ```python
-# This sync pins one point in time for its whole run, so resolving "latest"
-# once is right here even though it would not be in general.
-client = ForwardClient.from_env(snapshot_cache_ttl=600)
+client = ForwardClient.from_env(snapshot_cache_ttl=60)
 ```
 
+Do not reach for the number when you meant the first thing. A TTL that expires
+mid-run lets the next resolution return a different snapshot, so the run
+straddles two moments. Nothing raises, and the data is real on both sides of the
+expiry; it is simply from two different times. That is the same failure this
+whole area keeps producing, an answer that looks entirely plausible.
+
 That covers `latest_processed` and `latest_collected_id`, keyed by the exact
 question asked, so two different tag scopes stay two different answers. A
 network with no processed snapshot is cached as an answer too, since re-asking
 would spend the budget the setting exists to save.
 
 Uploading a snapshot through the same client clears it, because that is the
-event that makes a cached answer wrong. For anything the SDK cannot see, such
+event that makes a cached answer wrong. `"lifetime"` means "until something
+makes it wrong", not "forever", so it yields to an upload too. For anything the SDK cannot see, such
 as an upload from another process or a snapshot that finished while you were
 running, call `client.snapshots.clear_cache()`.
 

src/forward_sdk/_async/client.py

@@ -76,7 +76,7 @@ def __init__(
         snapshot_id: str | None = None,
         user_agent: str | None = None,
         cache_ttl: float = 60.0,
-        snapshot_cache_ttl: float = 0.0,
+        snapshot_cache_ttl: float | Literal["lifetime"] = 0.0,
         proxy: str | None = None,
         trust_env: bool = True,
         transport: httpx.AsyncBaseTransport | None = None,

src/forward_sdk/_async/services/snapshots.py

@@ -52,6 +52,10 @@
 """
 
 
+#: Reuse a resolved snapshot for as long as the client lives.
+LIFETIME = "lifetime"
+
+
 class _Miss:
     """Sentinel for "not cached", distinct from a cached ``None``."""
 
@@ -86,32 +90,41 @@ def clear_cache(self) -> None:
         """
         self._resolved.clear()
 
+    def _caching(self) -> bool:
+        ttl = self._config.snapshot_cache_ttl
+        return bool(ttl == LIFETIME or ttl > 0)
+
     def _cached(self, key: tuple[Any, ...]) -> Any:
         """A previously resolved answer to exactly this question, or a miss.
 
         Returns ``_MISS`` rather than ``None`` because ``None`` is a real
         answer: a network with no processed snapshot resolves to nothing, and
         re-asking every time would spend the budget the cache exists to save.
 
+        ``"lifetime"`` never expires. That is the point of it: a run holding one
+        point in time must not have the answer change underneath it, and a TTL
+        expiring mid-run would let the next resolution return a different
+        snapshot, straddling two moments with nothing raising.
+
         A race between two threads costs a duplicate resolution, never a wrong
         answer, since both compute the same thing from the same server state.
         """
         ttl = self._config.snapshot_cache_ttl
-        if ttl <= 0:
+        if not self._caching():
             return _MISS
         entry = self._resolved.get(key)
         if entry is None:
             self._transport.counters.increment("cache_misses")
             return _MISS
         fetched, value = entry
-        if (time.monotonic() - fetched) >= ttl:
+        if ttl != LIFETIME and (time.monotonic() - fetched) >= ttl:
             self._transport.counters.increment("cache_misses")
             return _MISS
         self._transport.counters.increment("cache_hits")
         return value
 
     def _remember(self, key: tuple[Any, ...], value: Any) -> Any:
-        if self._config.snapshot_cache_ttl > 0:
+        if self._caching():
             self._resolved[key] = (time.monotonic(), value)
         return value
 

src/forward_sdk/_sync/client.py

@@ -79,7 +79,7 @@ def __init__(
         snapshot_id: str | None = None,
         user_agent: str | None = None,
         cache_ttl: float = 60.0,
-        snapshot_cache_ttl: float = 0.0,
+        snapshot_cache_ttl: float | Literal["lifetime"] = 0.0,
         proxy: str | None = None,
         trust_env: bool = True,
         transport: httpx.BaseTransport | None = None,

src/forward_sdk/_sync/services/snapshots.py

@@ -54,6 +54,10 @@
 """
 
 
+#: Reuse a resolved snapshot for as long as the client lives.
+LIFETIME = "lifetime"
+
+
 class _Miss:
     """Sentinel for "not cached", distinct from a cached ``None``."""
 
@@ -88,32 +92,41 @@ def clear_cache(self) -> None:
         """
         self._resolved.clear()
 
+    def _caching(self) -> bool:
+        ttl = self._config.snapshot_cache_ttl
+        return bool(ttl == LIFETIME or ttl > 0)
+
     def _cached(self, key: tuple[Any, ...]) -> Any:
         """A previously resolved answer to exactly this question, or a miss.
 
         Returns ``_MISS`` rather than ``None`` because ``None`` is a real
         answer: a network with no processed snapshot resolves to nothing, and
         re-asking every time would spend the budget the cache exists to save.
 
+        ``"lifetime"`` never expires. That is the point of it: a run holding one
+        point in time must not have the answer change underneath it, and a TTL
+        expiring mid-run would let the next resolution return a different
+        snapshot, straddling two moments with nothing raising.
+
         A race between two threads costs a duplicate resolution, never a wrong
         answer, since both compute the same thing from the same server state.
         """
         ttl = self._config.snapshot_cache_ttl
-        if ttl <= 0:
+        if not self._caching():
             return _MISS
         entry = self._resolved.get(key)
         if entry is None:
             self._transport.counters.increment("cache_misses")
             return _MISS
         fetched, value = entry
-        if (time.monotonic() - fetched) >= ttl:
+        if ttl != LIFETIME and (time.monotonic() - fetched) >= ttl:
             self._transport.counters.increment("cache_misses")
             return _MISS
         self._transport.counters.increment("cache_hits")
         return value
 
     def _remember(self, key: tuple[Any, ...], value: Any) -> Any:
-        if self._config.snapshot_cache_ttl > 0:
+        if self._caching():
             self._resolved[key] = (time.monotonic(), value)
         return value
 

src/forward_sdk/config.py

@@ -99,12 +99,13 @@ class ClientConfig:
             Forward uses the network's latest processed snapshot.
         user_agent: Extra token identifying the calling application.
         cache_ttl: Seconds to reuse cached reads, or ``0`` to disable.
-        snapshot_cache_ttl: Seconds to reuse a resolved snapshot, or ``0``, the
-            default, to resolve every time. Off by default because only the
-            caller knows how long "latest" should stay true, and a stale
-            snapshot fails silently: it returns real data from the wrong moment.
-            Set it when a run pins one point in time, which is the usual case
-            for a sync.
+        snapshot_cache_ttl: How long to reuse a resolved snapshot. ``0``, the
+            default, resolves every time. ``"lifetime"`` resolves once and
+            keeps that answer for as long as the client lives, which is what a
+            run holding one point in time wants. A number of seconds suits
+            repeated independent reads, and is the wrong shape for a pinned
+            run: expiring mid-run lets the next resolution return a different
+            snapshot, so the run straddles two moments with nothing raising.
     """
 
     base_url: str
@@ -128,7 +129,7 @@ class ClientConfig:
     snapshot_id: str | None = None
     user_agent: str | None = None
     cache_ttl: float = 60.0
-    snapshot_cache_ttl: float = 0.0
+    snapshot_cache_ttl: float | Literal["lifetime"] = 0.0
     proxy: str | None = None
     trust_env: bool = True
 
@@ -248,7 +249,9 @@ def config_from_env(environ: dict[str, str] | None = None, **overrides: Any) ->
         settings["retries"] = int(retries)
     snapshot_ttl = _env("SNAPSHOT_CACHE_TTL", environ)
     if snapshot_ttl:
-        settings["snapshot_cache_ttl"] = float(snapshot_ttl)
+        settings["snapshot_cache_ttl"] = (
+            "lifetime" if snapshot_ttl.lower() == "lifetime" else float(snapshot_ttl)
+        )
 
     settings.update(overrides)
     return settings
@@ -267,7 +270,7 @@ def build_config(
     snapshot_id: str | None = None,
     user_agent: str | None = None,
     cache_ttl: float = 60.0,
-    snapshot_cache_ttl: float = 0.0,
+    snapshot_cache_ttl: float | Literal["lifetime"] = 0.0,
     proxy: str | None = None,
     trust_env: bool = True,
     stream_read_timeout: float = DEFAULT_STREAM_READ_TIMEOUT,
@@ -297,7 +300,9 @@ def build_config(
         snapshot_id=snapshot_id,
         user_agent=user_agent,
         cache_ttl=max(0.0, cache_ttl),
-        snapshot_cache_ttl=max(0.0, snapshot_cache_ttl),
+        snapshot_cache_ttl=(
+            "lifetime" if snapshot_cache_ttl == "lifetime" else max(0.0, float(snapshot_cache_ttl))
+        ),
         proxy=proxy,
         trust_env=trust_env,
     )

tests/_async/test_core_services.py

@@ -386,6 +386,49 @@ async def test_uploading_invalidates_it(self, recorder: Recorder, tmp_path: Path
         assert before is not None and before.id == "9"
         assert after is not None and after.id == "10"
 
+    async def test_lifetime_never_expires(
+        self, recorder: Recorder, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A run holding one point in time must not have it change underneath.
+
+        A TTL is the wrong shape for that: expiring mid-run lets the next
+        resolution return a newer snapshot, so the run straddles two moments
+        with nothing raising and the data real on both sides. "lifetime" is the
+        shape that matches the invariant a sync actually relies on, which is
+        that it pinned one snapshot and snapshots are immutable.
+        """
+        now = [1000.0]
+        monkeypatch.setattr("forward_sdk._async.services.snapshots.time.monotonic", lambda: now[0])
+        recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("9")]}))
+        async with make_client(recorder, snapshot_cache_ttl="lifetime") as client:
+            first = await client.snapshots.latest_processed()
+            now[0] += 86_400
+            second = await client.snapshots.latest_processed()
+
+        assert first is not None and second is not None
+        assert first.id == second.id == "9"
+        assert recorder.count("GET", SNAPSHOTS) == 1
+
+    async def test_lifetime_still_yields_to_an_upload(
+        self, recorder: Recorder, tmp_path: Path
+    ) -> None:
+        """Never expiring is not the same as never being wrong.
+
+        The client that uploaded a snapshot knows the answer changed, so
+        "lifetime" means "until something makes it wrong", not "forever".
+        """
+        archive = tmp_path / "snap.zip"
+        archive.write_bytes(b"zip")
+        recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("9")]}))
+        recorder.add("POST", SNAPSHOTS, json_response(snapshot("10")))
+        recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("10")]}))
+        async with make_client(recorder, snapshot_cache_ttl="lifetime") as client:
+            await client.snapshots.latest_processed()
+            await client.snapshots.upload([str(archive)])
+            after = await client.snapshots.latest_processed()
+
+        assert after is not None and after.id == "10"
+
     async def test_clear_cache_forces_a_fresh_resolution(self, recorder: Recorder) -> None:
         recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("9")]}))
         recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("11")]}))

tests/_sync/test_core_services.py

@@ -385,6 +385,47 @@ def test_uploading_invalidates_it(self, recorder: Recorder, tmp_path: Path) -> N
         assert before is not None and before.id == "9"
         assert after is not None and after.id == "10"
 
+    def test_lifetime_never_expires(
+        self, recorder: Recorder, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A run holding one point in time must not have it change underneath.
+
+        A TTL is the wrong shape for that: expiring mid-run lets the next
+        resolution return a newer snapshot, so the run straddles two moments
+        with nothing raising and the data real on both sides. "lifetime" is the
+        shape that matches the invariant a sync actually relies on, which is
+        that it pinned one snapshot and snapshots are immutable.
+        """
+        now = [1000.0]
+        monkeypatch.setattr("forward_sdk._sync.services.snapshots.time.monotonic", lambda: now[0])
+        recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("9")]}))
+        with make_client(recorder, snapshot_cache_ttl="lifetime") as client:
+            first = client.snapshots.latest_processed()
+            now[0] += 86_400
+            second = client.snapshots.latest_processed()
+
+        assert first is not None and second is not None
+        assert first.id == second.id == "9"
+        assert recorder.count("GET", SNAPSHOTS) == 1
+
+    def test_lifetime_still_yields_to_an_upload(self, recorder: Recorder, tmp_path: Path) -> None:
+        """Never expiring is not the same as never being wrong.
+
+        The client that uploaded a snapshot knows the answer changed, so
+        "lifetime" means "until something makes it wrong", not "forever".
+        """
+        archive = tmp_path / "snap.zip"
+        archive.write_bytes(b"zip")
+        recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("9")]}))
+        recorder.add("POST", SNAPSHOTS, json_response(snapshot("10")))
+        recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("10")]}))
+        with make_client(recorder, snapshot_cache_ttl="lifetime") as client:
+            client.snapshots.latest_processed()
+            client.snapshots.upload([str(archive)])
+            after = client.snapshots.latest_processed()
+
+        assert after is not None and after.id == "10"
+
     def test_clear_cache_forces_a_fresh_resolution(self, recorder: Recorder) -> None:
         recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("9")]}))
         recorder.add("GET", SNAPSHOTS, json_response({"snapshots": [snapshot("11")]}))

tests/unit/test_config.py

@@ -156,3 +156,24 @@ def test_user_agent_identifies_the_sdk_and_the_application() -> None:
     assert agent.startswith("forward-sdk/")
     assert "python-httpx/" in agent
     assert agent.endswith("my-app/2.0")
+
+
+class TestSnapshotCacheTtl:
+    def test_off_by_default(self) -> None:
+        assert build_config("https://forward.test").snapshot_cache_ttl == 0.0
+
+    def test_lifetime_survives_config_building(self) -> None:
+        config = build_config("https://forward.test", snapshot_cache_ttl="lifetime")
+        assert config.snapshot_cache_ttl == "lifetime"
+
+    def test_a_negative_ttl_is_off_rather_than_an_error(self) -> None:
+        assert build_config("https://forward.test", snapshot_cache_ttl=-5).snapshot_cache_ttl == 0.0
+
+    def test_read_from_the_environment(self) -> None:
+        environ = {"FORWARD_URL": "https://forward.test", "FORWARD_SNAPSHOT_CACHE_TTL": "120"}
+        assert config_from_env(environ)["snapshot_cache_ttl"] == 120.0
+
+    def test_lifetime_read_from_the_environment(self) -> None:
+        """Spelled as a word, so a deployment can pin a run without a magic number."""
+        environ = {"FORWARD_URL": "https://forward.test", "FORWARD_SNAPSHOT_CACHE_TTL": "lifetime"}
+        assert config_from_env(environ)["snapshot_cache_ttl"] == "lifetime"