feat(flags): feature-flag system with sdk.flags and operation gating

Platform instances enable features per environment, exposed at
GET /ide-service/flags (categories internal/preview/features, each with
available+enabled lists). This adds:

- sdk.flags (FlagsResource): get()/refresh()/is_available()/is_enabled()/
  require(), category-scoped or union lookups, categories_with() for
  collision insight. Snapshot models are frozen pydantic classes
- Gating infrastructure: APIOperation gains required_flag:
  FlagRequirement | None (mirrors the input_model pattern); enforced as
  the first step of execute_operation, so a gated call raises BEFORE any
  request when the flag is not enabled. _require_flag() escape hatch on
  ResourceBase/BoundModel for non-operation paths (streams, shell
  features). No existing endpoint is gated
- FlagsStore engine in new top-level feature_flags.py (cycle-free wrt
  core<->http_client), hosted per-connection on APIHttpClient; lazy
  fetch under an asyncio.Lock (one fetch across concurrent gated calls),
  cached for client lifetime, refresh() re-fetches. Fail closed on
  legacy instances: flags-endpoint 404 caches an empty snapshot and
  gated calls raise with a legacy-platform hint; transient errors
  propagate uncached
- New exceptions: FeatureFlagError base (flag/category attrs) with
  FeatureNotAvailableError (wrong instance/platform version) and
  FeatureNotEnabledError (available but off), exported top-level along
  with FlagCategory/FlagsSnapshot/CategoryFlags
- Tests: snapshot model matrix, sdk.flags caching/refresh/404/transient
  behavior, operation gating incl. pre-request fail-fast proof
  (gated route never called), single flag fetch across concurrent gated
  calls, _require_flag on ResourceBase and bound/detached BoundModel;
  exception matrix extended; live integration test
- Docs: CONTRIBUTING 'Gating an Endpoint Behind a Feature Flag' recipe
  (explicit category required in SDK declarations, async-generator
  guidance, test pattern), README 'Feature flags' section

Author: Datata1

CONTRIBUTING.md

@@ -129,6 +129,56 @@ versions.
 
 ---
 
+## Gating an Endpoint Behind a Feature Flag
+
+Platform instances enable features via flags (`GET /ide-service/flags`,
+three categories: `internal`, `preview`, `features`). If an SDK feature only
+works when a flag is enabled, declare it on the operation — **always with an
+explicit category**:
+
+```python
+from ...core.operations import APIOperation
+from ...feature_flags import FlagCategory, FlagRequirement
+
+_LIST_VPN_CONFIGS_OP = APIOperation(
+    method="GET",
+    endpoint_template="/vpn/configs",
+    response_model=ResourceList[VpnConfig],
+    required_flag=FlagRequirement("vpn", FlagCategory.INTERNAL),
+)
+```
+
+Nothing else is needed: `execute_operation` checks the flag **before any
+request** and raises `FeatureNotAvailableError` (flag missing on this
+instance) or `FeatureNotEnabledError` (available but off). The flags
+snapshot is fetched once per client and cached; `await sdk.flags.refresh()`
+re-fetches. Instances without the flags endpoint fail closed.
+
+For code paths that don't go through an `APIOperation` (SSE streams,
+shell-command features), check explicitly:
+
+```python
+await self._require_flag(FlagRequirement("workspace-ssh", FlagCategory.PREVIEW))
+```
+
+For async generators, perform this check in a plain `async def` factory
+*before* returning the generator, so the call fails fast instead of at the
+first iteration.
+
+**Testing a gated endpoint:** register a respx route for
+`/ide-service/flags` alongside the endpoint route, and assert the endpoint
+route was **not** called when the flag is disabled:
+
+```python
+mock_api.get("/ide-service/flags").respond(200, json={"preview": {"available": ["x"], "enabled": []}})
+route = mock_api.get("/gated").respond(200, json={})
+with pytest.raises(FeatureNotEnabledError):
+    await resource.gated_call()
+assert route.called is False
+```
+
+---
+
 ## Testing Guidelines
 
 We maintain two types of tests: **unit tests** and **integration tests**. When contributing, please ensure appropriate test coverage for your changes.

README.md

@@ -82,6 +82,22 @@ add it to `retry_methods` explicitly:
 RetryConfig(max_retries=3, retry_methods=frozenset({"GET", "POST"}))
 ```
 
+### Feature flags
+
+Platform instances enable different features. Inspect them via `sdk.flags`:
+
+```python
+snapshot = await sdk.flags.get()          # fetched once, then cached
+if await sdk.flags.is_enabled("workspace-ssh"):
+    ...
+
+await sdk.flags.refresh()                 # re-fetch after platform changes
+```
+
+SDK features that depend on a platform flag raise `FeatureNotEnabledError`
+(or `FeatureNotAvailableError` if the instance doesn't have the feature at
+all) *before* sending any request when the flag is off.
+
 ## Quick Start
 
 ```python

src/codesphere/__init__.py

@@ -29,12 +29,16 @@
     AuthorizationError,
     CodesphereError,
     ConflictError,
+    FeatureFlagError,
+    FeatureNotAvailableError,
+    FeatureNotEnabledError,
     NetworkError,
     NotFoundError,
     RateLimitError,
     TimeoutError,
     ValidationError,
 )
+from .resources.flags import CategoryFlags, FlagCategory, FlagsSnapshot
 from .resources.metadata import Characteristic, Datacenter, Image, WsPlan
 from .resources.team import (
     CustomDomainConfig,
@@ -72,6 +76,12 @@
     "DomainRouting",
     "DomainVerificationStatus",
     "EnvVar",
+    "FeatureFlagError",
+    "FeatureNotAvailableError",
+    "FeatureNotEnabledError",
+    "FlagCategory",
+    "FlagsSnapshot",
+    "CategoryFlags",
     "Image",
     "NetworkError",
     "NotFoundError",

src/codesphere/client.py

@@ -6,6 +6,7 @@
 from .config import RetryConfig, Settings
 from .exceptions import AuthenticationError
 from .http_client import APIHttpClient
+from .resources.flags import FlagsResource
 from .resources.metadata import MetadataResource
 from .resources.team import TeamsResource
 from .resources.workspace import WorkspacesResource
@@ -24,6 +25,7 @@ class CodesphereSDK:
     teams: TeamsResource
     workspaces: WorkspacesResource
     metadata: MetadataResource
+    flags: FlagsResource
 
     def __init__(
         self,
@@ -60,6 +62,7 @@ def __init__(
         self.teams = TeamsResource(self._http_client)
         self.workspaces = WorkspacesResource(self._http_client)
         self.metadata = MetadataResource(self._http_client)
+        self.flags = FlagsResource(self._http_client)
 
     @staticmethod
     def _resolve_token(token: str | SecretStr | None, settings: Settings) -> SecretStr:

src/codesphere/core/base.py

@@ -5,6 +5,7 @@
 from pydantic import BaseModel, ConfigDict, PrivateAttr, RootModel
 from pydantic.alias_generators import to_camel
 
+from ..feature_flags import FlagRequirement
 from ..http_client import APIHttpClient
 from .handler import RequestData, execute_operation
 from .operations import APIOperation
@@ -31,6 +32,10 @@ async def _execute(
             self._http_client, op, data=data, params=params, path_params=path_params
         )
 
+    async def _require_flag(self, requirement: FlagRequirement) -> None:
+        """Fail fast unless the platform flag is enabled (non-operation paths)."""
+        await self._http_client.flags_store.require(requirement)
+
 
 class TeamScopedResource(ResourceBase):
     """Base for managers operating within one team."""
@@ -137,6 +142,10 @@ async def _execute(
             self._client(), op, data=data, params=params, path_params=path_params
         )
 
+    async def _require_flag(self, requirement: FlagRequirement) -> None:
+        """Fail fast unless the platform flag is enabled (non-operation paths)."""
+        await self._client().flags_store.require(requirement)
+
 
 class ResourceList(RootModel[list[ModelT]], Generic[ModelT]):
     root: list[ModelT]

src/codesphere/core/handler.py

@@ -29,8 +29,13 @@ async def execute_operation(
     Path parameters are passed explicitly and formatted into
     ``op.endpoint_template``. When ``op.input_model`` is set, ``data`` is
     validated against it before the request is sent. A non-``None`` ``data``
-    is always sent as the JSON body, even when empty.
+    is always sent as the JSON body, even when empty. When the operation
+    declares ``required_flag``, the flag is checked first — nothing is
+    sent if the connected instance does not have it enabled.
     """
+    if op.required_flag is not None:
+        await client.flags_store.require(op.required_flag)
+
     endpoint = _format_endpoint(op.endpoint_template, path_params or {})
 
     request_kwargs: dict[str, Any] = {}

src/codesphere/core/operations.py

@@ -3,6 +3,8 @@
 
 from pydantic import BaseModel
 
+from ..feature_flags import FlagRequirement
+
 ResponseT = TypeVar("ResponseT")
 EntryT = TypeVar("EntryT", bound=BaseModel)
 
@@ -15,18 +17,25 @@ class APIOperation(Generic[ResponseT]):
     type of ``_execute``: use a pydantic model class (``ResourceList[Team]``
     included) or ``types.NoneType`` for endpoints without a response body.
     ``input_model``, when set, validates the ``data`` payload before the
-    request is sent.
+    request is sent. ``required_flag``, when set, gates the operation on a
+    platform feature flag — the call fails before any request when the
+    flag is not enabled on the connected instance.
     """
 
     method: str
     endpoint_template: str
     response_model: type[ResponseT]
     input_model: type[BaseModel] | None = None
+    required_flag: FlagRequirement | None = None
 
 
 @dataclass(frozen=True, slots=True)
 class StreamOperation(Generic[EntryT]):
-    """Declarative description of an SSE streaming endpoint."""
+    """Declarative description of an SSE streaming endpoint.
+
+    Streaming endpoints are not flag-gated declaratively yet; gate their
+    entry points explicitly with ``await self._require_flag(...)``.
+    """
 
     endpoint_template: str
     entry_model: type[EntryT]

src/codesphere/exceptions.py

@@ -160,6 +160,83 @@ def __init__(self, message: str | None = None):
         super().__init__(message)
 
 
+def _flag_label(flag: str, category: str | None) -> str:
+    if category:
+        return f"'{flag}' (category '{category}')"
+    return f"'{flag}'"
+
+
+class FeatureFlagError(CodesphereError):
+    """Base for feature-flag gating errors.
+
+    Raised before any request is sent when an SDK feature requires a
+    platform feature flag that the connected instance does not satisfy.
+    """
+
+    def __init__(
+        self,
+        message: str | None = None,
+        *,
+        flag: str = "",
+        category: str | None = None,
+    ):
+        self.flag = flag
+        self.category = category
+        if message is None:
+            message = (
+                f"Feature flag {_flag_label(flag, category)} blocks this SDK feature."
+            )
+        super().__init__(message)
+
+
+class FeatureNotAvailableError(FeatureFlagError):
+    """The required flag is not available on the connected instance.
+
+    The feature does not exist on this platform instance (wrong
+    environment, or the platform predates the feature).
+    """
+
+    def __init__(
+        self,
+        message: str | None = None,
+        *,
+        flag: str = "",
+        category: str | None = None,
+        legacy_platform: bool = False,
+    ):
+        if message is None:
+            message = (
+                f"Feature flag {_flag_label(flag, category)} is not available "
+                "on this Codesphere instance."
+            )
+            if legacy_platform:
+                message += (
+                    " The instance did not expose the flags endpoint at all; "
+                    "it may predate feature flags."
+                )
+            message += " Use sdk.flags.get() to inspect available flags."
+        super().__init__(message, flag=flag, category=category)
+
+
+class FeatureNotEnabledError(FeatureFlagError):
+    """The required flag is available but not enabled on the connected instance."""
+
+    def __init__(
+        self,
+        message: str | None = None,
+        *,
+        flag: str = "",
+        category: str | None = None,
+    ):
+        if message is None:
+            message = (
+                f"Feature flag {_flag_label(flag, category)} is not enabled on "
+                "this Codesphere instance. Ask an administrator to enable it, "
+                "or use sdk.flags.get() to inspect the current flags."
+            )
+        super().__init__(message, flag=flag, category=category)
+
+
 def raise_for_status(response: httpx.Response) -> None:
     """Convert HTTP errors to appropriate SDK exceptions.