feat(SDK): Surface upcoming scheduled changes in the environment document

Add an opt-in `include_scheduled` query param to the SDK environment-document
endpoint. When set, each feature state additionally carries a
`scheduled_change` field (`live_from`/`enabled`/`feature_state_value`)
describing its next not-yet-live version, if any — sourced from both a future
FeatureState/EnvironmentFeatureVersion already in the DB and a committed but
not-yet-materialised VersionChangeSet diff, with an earliest-wins rule when
both exist.

This lets SDKs running in local-evaluation mode resolve a scheduled change's
maturity from an already-cached document, without waiting for a new poll
after `live_from` passes — and keeps a fleet of instances in sync at the same
moment, since every instance resolves maturity from its own clock rather than
from when it happened to last poll.

Fully backward compatible: the field is omitted entirely when not requested,
so existing SDKs and cached documents are byte-for-byte unchanged.

Test names follow the repo's test_{subject}__{condition}__{expected}
convention with Given/When/Then comments, as flat functions per the
flagsmith-lint-tests pre-commit hook. The SDK API reference page for this
endpoint is regenerated from the updated OpenAPI spec.

Author: cloorc

api/environments/models.py

@@ -534,14 +534,41 @@ def get_segments_from_cache(self) -> typing.List[Segment]:
     def get_environment_document(
         cls,
         api_key: str,
+        *,
+        include_scheduled: bool = False,
     ) -> dict[str, typing.Any]:
+        """
+        :param include_scheduled: opt-in. When True, feature states in the
+            returned document carry an additional `scheduled_change` field
+            describing their next not-yet-live version, if any. Defaults to
+            False, in which case the response shape is unchanged.
+        """
+        # PERSISTENT mode relies on proactive rewrite — entries never expire,
+        # and only write_environment_documents and the delete/rename hooks
+        # keep the base `api_key` entry fresh. None of
+        # them maintain the `:scheduled` key variant, and no write-triggered
+        # refresh ever could: a schedule maturing (its live_from passing)
+        # changes the correct scheduled document without any DB write. So
+        # opted-in requests bypass the cache entirely under PERSISTENT mode.
+        if (
+            include_scheduled
+            and settings.CACHE_ENVIRONMENT_DOCUMENT_MODE
+            == EnvironmentDocumentCacheMode.PERSISTENT
+        ):
+            return cls._get_environment_document_from_db(
+                api_key, include_scheduled=True
+            )
         if (
             settings.CACHE_ENVIRONMENT_DOCUMENT_SECONDS > 0
             or settings.CACHE_ENVIRONMENT_DOCUMENT_MODE
             == EnvironmentDocumentCacheMode.PERSISTENT
         ):
-            return cls._get_environment_document_from_cache(api_key)
-        return cls._get_environment_document_from_db(api_key)
+            return cls._get_environment_document_from_cache(
+                api_key, include_scheduled=include_scheduled
+            )
+        return cls._get_environment_document_from_db(
+            api_key, include_scheduled=include_scheduled
+        )
 
     def get_create_log_message(self, history_instance) -> typing.Optional[str]:  # type: ignore[no-untyped-def]
         return ENVIRONMENT_CREATED_MESSAGE % self.name  # type: ignore[no-any-return]
@@ -559,11 +586,21 @@ def get_hide_disabled_flags(self) -> bool:
     def _get_environment_document_from_cache(
         cls,
         api_key: str,
+        *,
+        include_scheduled: bool = False,
     ) -> dict[str, typing.Any]:
-        environment_document = environment_document_cache.get(api_key)
+        # The cache key must vary with `include_scheduled` — the response
+        # shape differs between the two, so a single shared key
+        # would let whichever request populates the cache first silently
+        # dictate what every other request gets back, regardless of what it
+        # actually asked for.
+        cache_key = f"{api_key}:scheduled" if include_scheduled else api_key
+        environment_document = environment_document_cache.get(cache_key)
         if not (cache_hit := environment_document is not None):
-            environment_document = cls._get_environment_document_from_db(api_key)
-            environment_document_cache.set(api_key, environment_document)
+            environment_document = cls._get_environment_document_from_db(
+                api_key, include_scheduled=include_scheduled
+            )
+            environment_document_cache.set(cache_key, environment_document)
 
         flagsmith_environment_document_cache_queries_total.labels(
             result=CACHE_HIT if cache_hit else CACHE_MISS,
@@ -575,6 +612,8 @@ def _get_environment_document_from_cache(
     def _get_environment_document_from_db(
         cls,
         api_key: str,
+        *,
+        include_scheduled: bool = False,
     ) -> dict[str, typing.Any]:
         manager = using_database_replica(cls.objects)
         environment = manager.filter_for_document_builder(
@@ -611,7 +650,9 @@ def _get_environment_document_from_db(
                 ),
             ],
         ).get()
-        return map_environment_to_sdk_document(environment)
+        return map_environment_to_sdk_document(
+            environment, include_scheduled=include_scheduled
+        )
 
     def _get_environment(self):  # type: ignore[no-untyped-def]
         return self

api/environments/sdk/views.py

@@ -3,7 +3,7 @@
 
 from django.utils.decorators import method_decorator
 from django.views.decorators.http import condition
-from drf_spectacular.utils import extend_schema
+from drf_spectacular.utils import OpenApiParameter, extend_schema
 from flagsmith_schemas.api import V1EnvironmentDocumentResponse
 from rest_framework.request import Request
 from rest_framework.response import Response
@@ -17,7 +17,26 @@
 from environments.permissions.permissions import EnvironmentKeyPermissions
 
 
+def _is_include_scheduled_requested(request: Request) -> bool:
+    # Opt-in query param for `scheduled_change` data.
+    return request.GET.get("include_scheduled", "").lower() in ("1", "true", "yes")
+
+
 def get_last_modified(request: Request) -> datetime | None:
+    # A schedule maturing (`live_from` passing) doesn't touch
+    # `environment.updated_at` — nothing is saved when time simply
+    # elapses. Requests that opt into `scheduled_change` data therefore can't
+    # rely on this field to decide freshness, and worse, a client that has a
+    # cached body from *before* it started opting in could otherwise get a
+    # 304 short-circuit here and keep reusing that stale, scheduled-change-free
+    # body forever. Returning None tells Django's `condition()` there is no
+    # known last-modified value, so it always falls through to `get()` instead.
+    # Trade-off: this disables the 304 short-circuit for every
+    # `include_scheduled=True` request (each one always executes `get()` in
+    # full), in exchange for guaranteed freshness rather than a cheap-but-stale
+    # response.
+    if _is_include_scheduled_requested(request):
+        return None
     updated_at: Optional[datetime] = request.environment.updated_at
     return updated_at
 
@@ -31,6 +50,22 @@ def get_authenticators(self):  # type: ignore[no-untyped-def]
         return [EnvironmentKeyAuthentication(required_key_prefix="ser.")]
 
     @extend_schema(
+        parameters=[
+            OpenApiParameter(
+                name="include_scheduled",
+                type=bool,
+                required=False,
+                description=(
+                    "Opt-in. When true, each feature state in "
+                    "the response carries an additional `scheduled_change` "
+                    "field describing its next not-yet-live version, if any. "
+                    "Defaults to false, in which case the response shape is "
+                    "unchanged. Not every scheduled-change shape is surfaced "
+                    "— see the 'Known limitations' note on "
+                    "`map_environment_to_engine` in `util/mappers/engine.py`."
+                ),
+            ),
+        ],
         responses={200: V1EnvironmentDocumentResponse},
         operation_id="sdk_v1_environment_document",
     )
@@ -42,6 +77,7 @@ def get(self, request: Request) -> Response:
         """
         environment_document = Environment.get_environment_document(
             request.environment.api_key,
+            include_scheduled=_is_include_scheduled_requested(request),
         )
         updated_at = self.request.environment.updated_at
         return Response(