Add OIDC back-channel logout with session indexes

Migrate dependency management from pip to uv
- Add a uv workspace at the repository root covering all packages/kaya-* members - Replace pip-compile requirements files with a committed uv.lock - Remove redundant per-package dev extras - Build and publish artifacts with uv build and uv publish in CI - Update the README with uv sync/run/build commands
2026-10-28 07:50:14 +00:00 · 2026-10-21 01:02:08 +00:00 · 2026-10-14 11:52:03 +00:00
43 changed files with 3662 additions and 448 deletions
@@ -5,7 +5,7 @@ on:
      - 'release/*'
 jobs:
  build_pip_package:
-    name: "Build Pip package"
+    name: "Build and publish packages"
    runs-on: hostinger
    timeout-minutes: 30
    container:
@@ -18,12 +18,14 @@ jobs:
          fetch-tags: true
      - uses: actions/setup-python@v6
        with:
-          python-version: '3.14.6' 
-          cache: 'pip'
+          python-version-file: '.python-version'
+      - name: Install uv
+        uses: astral-sh/setup-uv@v10
+        with:
+          version: '0.13.4'
+          enable-cache: true
      - name: Install dependencies
-        run: |
-          python -m venv .venv
-          .venv/bin/pip install -r requirements-dev.txt
+        run: uv sync --locked
      - name: Check kaya-core
        run: |
          .venv/bin/python -m mypy -p kaya.core
@@ -66,81 +68,81 @@ jobs:
          .venv/bin/python -m unittest discover -s packages/kaya-otel/tests
      - name: Publish kaya-core artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-core
-          .venv/bin/twine upload --repository gitea packages/kaya-core/dist/*.whl packages/kaya-core/dist/*.tar.gz
+          uv build --package kaya-core --out-dir packages/kaya-core/dist --no-sources
+          uv publish packages/kaya-core/dist/*.whl packages/kaya-core/dist/*.tar.gz
      - name: Publish kaya-rsgi artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-rsgi
-          .venv/bin/twine upload --repository gitea packages/kaya-rsgi/dist/*.whl packages/kaya-rsgi/dist/*.tar.gz
+          uv build --package kaya-rsgi --out-dir packages/kaya-rsgi/dist --no-sources
+          uv publish packages/kaya-rsgi/dist/*.whl packages/kaya-rsgi/dist/*.tar.gz
      - name: Publish kaya-session artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-session
-          .venv/bin/twine upload --repository gitea packages/kaya-session/dist/*.whl packages/kaya-session/dist/*.tar.gz
+          uv build --package kaya-session --out-dir packages/kaya-session/dist --no-sources
+          uv publish packages/kaya-session/dist/*.whl packages/kaya-session/dist/*.tar.gz
      - name: Publish kaya-session-memcache artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-session-memcache
-          .venv/bin/twine upload --repository gitea packages/kaya-session-memcache/dist/*.whl packages/kaya-session-memcache/dist/*.tar.gz
+          uv build --package kaya-session-memcache --out-dir packages/kaya-session-memcache/dist --no-sources
+          uv publish packages/kaya-session-memcache/dist/*.whl packages/kaya-session-memcache/dist/*.tar.gz
      - name: Publish kaya-session-redis artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-session-redis
-          .venv/bin/twine upload --repository gitea packages/kaya-session-redis/dist/*.whl packages/kaya-session-redis/dist/*.tar.gz
+          uv build --package kaya-session-redis --out-dir packages/kaya-session-redis/dist --no-sources
+          uv publish packages/kaya-session-redis/dist/*.whl packages/kaya-session-redis/dist/*.tar.gz
      - name: Publish kaya-oidc artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-oidc
-          .venv/bin/twine upload --repository gitea packages/kaya-oidc/dist/*.whl packages/kaya-oidc/dist/*.tar.gz
+          uv build --package kaya-oidc --out-dir packages/kaya-oidc/dist --no-sources
+          uv publish packages/kaya-oidc/dist/*.whl packages/kaya-oidc/dist/*.tar.gz
      - name: Publish kaya-openapi artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-openapi
-          .venv/bin/twine upload --repository gitea packages/kaya-openapi/dist/*.whl packages/kaya-openapi/dist/*.tar.gz
+          uv build --package kaya-openapi --out-dir packages/kaya-openapi/dist --no-sources
+          uv publish packages/kaya-openapi/dist/*.whl packages/kaya-openapi/dist/*.tar.gz
      - name: Publish kaya-cors artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-cors
-          .venv/bin/twine upload --repository gitea packages/kaya-cors/dist/*.whl packages/kaya-cors/dist/*.tar.gz
+          uv build --package kaya-cors --out-dir packages/kaya-cors/dist --no-sources
+          uv publish packages/kaya-cors/dist/*.whl packages/kaya-cors/dist/*.tar.gz
      - name: Publish kaya-forwarded artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-forwarded
-          .venv/bin/twine upload --repository gitea packages/kaya-forwarded/dist/*.whl packages/kaya-forwarded/dist/*.tar.gz
+          uv build --package kaya-forwarded --out-dir packages/kaya-forwarded/dist --no-sources
+          uv publish packages/kaya-forwarded/dist/*.whl packages/kaya-forwarded/dist/*.tar.gz
      - name: Publish kaya-otel artifacts
        env:
-          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
-          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
-          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+          UV_PUBLISH_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          UV_PUBLISH_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          UV_PUBLISH_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
        run: |
-          .venv/bin/pyproject-build packages/kaya-otel
-          .venv/bin/twine upload --repository gitea packages/kaya-otel/dist/*.whl packages/kaya-otel/dist/*.tar.gz
+          uv build --package kaya-otel --out-dir packages/kaya-otel/dist --no-sources
+          uv publish packages/kaya-otel/dist/*.whl packages/kaya-otel/dist/*.tar.gz
@@ -0,0 +1 @@
+3.14.6
@@ -17,69 +17,65 @@ This repository is a monorepo for the Kaya framework. The code is split into ind
 - **kaya-forwarded** — trusted-proxy `Forwarded`/`X-Forwarded-*` client address resolution (`packages/kaya-forwarded/`)
 - **kaya-otel** — OpenTelemetry tracing and metrics (`packages/kaya-otel/`)

-Additional `kaya-*` packages can be added as new directories under `packages/`.
+Additional `kaya-*` packages can be added as new directories under `packages/`. When adding a new package, also add it to the root `pyproject.toml` dependencies and `[tool.uv.sources]` so `uv sync` installs it.

 ## Build & run locally

-Install dev dependencies:
+This project uses [uv](https://docs.astral.sh/uv/) for dependency management. From the repository root:

 ```bash
-pip install --index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --extra-index-url https://pypi.org/simple -r requirements-dev.txt
+uv sync
 ```

-Install the packages in development mode:
-
-```bash
-pip install -e packages/kaya-core -e packages/kaya-rsgi -e packages/kaya-session -e packages/kaya-session-redis -e packages/kaya-session-memcache -e packages/kaya-oidc -e packages/kaya-openapi -e packages/kaya-cors -e packages/kaya-forwarded -e packages/kaya-otel
-```
+This creates `.venv`, installs all workspace packages in editable mode, and installs the dev dependency group from `uv.lock`. Use `uv lock` after changing dependencies, and commit the updated `uv.lock`.

 Run the example:

 ```bash
-python example/hello.py
+uv run python example/hello.py
 ```

 ## Tests

 ```bash
-python -m unittest discover -s packages/kaya-core/tests
-python -m unittest discover -s packages/kaya-rsgi/tests
-python -m unittest discover -s packages/kaya-session/tests
-python -m unittest discover -s packages/kaya-session-redis/tests
-python -m unittest discover -s packages/kaya-session-memcache/tests
-python -m unittest discover -s packages/kaya-oidc/tests
-python -m unittest discover -s packages/kaya-openapi/tests
-python -m unittest discover -s packages/kaya-cors/tests
-python -m unittest discover -s packages/kaya-forwarded/tests
-python -m unittest discover -s packages/kaya-otel/tests
+uv run python -m unittest discover -s packages/kaya-core/tests
+uv run python -m unittest discover -s packages/kaya-rsgi/tests
+uv run python -m unittest discover -s packages/kaya-session/tests
+uv run python -m unittest discover -s packages/kaya-session-redis/tests
+uv run python -m unittest discover -s packages/kaya-session-memcache/tests
+uv run python -m unittest discover -s packages/kaya-oidc/tests
+uv run python -m unittest discover -s packages/kaya-openapi/tests
+uv run python -m unittest discover -s packages/kaya-cors/tests
+uv run python -m unittest discover -s packages/kaya-forwarded/tests
+uv run python -m unittest discover -s packages/kaya-otel/tests
 ```

 ## Static analysis

 ```bash
-mypy -p kaya.core
-mypy -p kaya.rsgi
-mypy -p kaya.session
-mypy -p kaya.session.redis
-mypy -p kaya.session.memcache
-mypy -p kaya.oidc
-mypy -p kaya.openapi
-mypy -p kaya.cors
-mypy -p kaya.forwarded
-mypy -p kaya.otel
+uv run mypy -p kaya.core
+uv run mypy -p kaya.rsgi
+uv run mypy -p kaya.session
+uv run mypy -p kaya.session.redis
+uv run mypy -p kaya.session.memcache
+uv run mypy -p kaya.oidc
+uv run mypy -p kaya.openapi
+uv run mypy -p kaya.cors
+uv run mypy -p kaya.forwarded
+uv run mypy -p kaya.otel
 ```

 ## Building packages

 ```bash
-python -m build packages/kaya-core
-python -m build packages/kaya-rsgi
-python -m build packages/kaya-session
-python -m build packages/kaya-session-redis
-python -m build packages/kaya-session-memcache
-python -m build packages/kaya-oidc
-python -m build packages/kaya-openapi
-python -m build packages/kaya-cors
-python -m build packages/kaya-forwarded
-python -m build packages/kaya-otel
+uv build --package kaya-core --out-dir packages/kaya-core/dist
+uv build --package kaya-rsgi --out-dir packages/kaya-rsgi/dist
+uv build --package kaya-session --out-dir packages/kaya-session/dist
+uv build --package kaya-session-redis --out-dir packages/kaya-session-redis/dist
+uv build --package kaya-session-memcache --out-dir packages/kaya-session-memcache/dist
+uv build --package kaya-oidc --out-dir packages/kaya-oidc/dist
+uv build --package kaya-openapi --out-dir packages/kaya-openapi/dist
+uv build --package kaya-cors --out-dir packages/kaya-cors/dist
+uv build --package kaya-forwarded --out-dir packages/kaya-forwarded/dist
+uv build --package kaya-otel --out-dir packages/kaya-otel/dist
 ```
@@ -26,11 +26,6 @@ dependencies = [
    "typing-extensions",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine", "httpx", "httpx-ws"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -25,11 +25,6 @@ dependencies = [
    "kaya-core",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine", "httpx", "httpx-ws", "kaya-rsgi"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -25,11 +25,6 @@ dependencies = [
    "kaya-core",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine", "httpx", "httpx-ws", "kaya-rsgi"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -128,6 +128,50 @@ app = KayaApp(mixins=[session, google, github])
 > default provider is `default`). Existing logged-in users are signed out once
 > and simply re-authenticate.

+## Back-channel logout
+
+Set `backchannel_logout_path` to accept OIDC Back-Channel Logout tokens from
+the provider:
+
+```python
+OIDCConfig(
+    issuer=os.environ['OIDC_ISSUER'],
+    client_id=os.environ['OIDC_CLIENT_ID'],
+    client_secret=os.environ['OIDC_CLIENT_SECRET'],
+    redirect_uri='https://example.com/auth/callback',
+    backchannel_logout_path='/auth/backchannel-logout',
+)
+```
+
+Register the corresponding absolute URL (for example
+`https://example.com/auth/backchannel-logout`) as the client's back-channel
+logout URI with the provider. This registration happens at the provider; it is
+not part of OIDC discovery. With `OIDCProviders`, enabling the option derives
+`{base_path}/{provider}/backchannel-logout` for that provider.
+
+The endpoint accepts form-encoded `POST` requests containing `logout_token`.
+It validates the JWT signature against the provider JWKS, the `logout+jwt`
+type, issuer, audience, `iat`, `jti`, and the back-channel logout event, and
+rejects tokens containing `nonce`. A token with `sid` logs out sessions for
+that provider session; without `sid`, `sub` logs out all sessions for that
+subject. Under `multi-identity`, only the named provider's identity is removed
+from a shared session. Valid requests return `200` with
+`Cache-Control: no-store`, including when the session is already gone. Invalid
+requests return `400`.
+
+Back-channel logout requires a session store with index support. All built-in
+stores (`InMemorySessionStore`, `RedisSessionStore`, and
+`MemcacheSessionStore`) support it. `OIDCMixin` raises immediately if the
+configured store does not. Sessions created before upgrading are indexed on
+their next request.
+
+Logout tokens must be asymmetrically signed; unsigned tokens, HMAC signatures,
+and encrypted logout tokens (JWE) are not supported. `jti` is required, but no
+replay cache is maintained: repeating a valid logout request is idempotent.
+`backchannel_logout_leeway` controls the allowed clock skew (60 seconds by
+default). For a templated Microsoft Entra issuer, the logout token must carry
+`tid`, just as for ID token validation.
+
 ## Pushed authorization requests (PAR)

 `OIDCConfig.par` controls RFC 9126 Pushed Authorization Requests:
@@ -172,6 +216,7 @@ authentication methods (`private_key_jwt`, mTLS) is not supported.
 - Optional userinfo endpoint fetch
 - Refresh token support
 - RP-initiated logout (when provider advertises `end_session_endpoint`)
+- Back-channel logout (`sid` and `sub` session invalidation)
 - Pushed Authorization Requests (RFC 9126) via `OIDCConfig.par`
 - Multiple providers in one app via `OIDCProviders` (or several `OIDCMixin` instances)
 - `single-identity` (default) or `multi-identity` session strategy
@@ -28,11 +28,6 @@ dependencies = [
    "PyJWT[crypto]",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine", "httpx", "httpx-ws"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -10,6 +10,8 @@ from ._config import PARPolicy, OIDCConfig
 from ._utils import generate_nonce, generate_pkce, generate_state

 _TENANT_ID_PLACEHOLDER = re.compile(r'\{tenantid\}', re.IGNORECASE)
+_BACKCHANNEL_LOGOUT_EVENT = 'http://schemas.openid.net/event/backchannel-logout'
+_LOGOUT_TOKEN_TYPES = frozenset({'logout+jwt', 'application/logout+jwt'})


 class OIDCClient:
@@ -189,6 +191,60 @@ class OIDCClient:
            raise ValueError('ID token nonce mismatch')
        return payload

+    async def validate_logout_token(self, logout_token: str) -> Mapping[str, Any]:
+        """Validate an OIDC Back-Channel Logout token and return its claims."""
+        header = jwt.get_unverified_header(logout_token)
+        typ = header.get('typ')
+        if not isinstance(typ, str) or typ.lower() not in _LOGOUT_TOKEN_TYPES:
+            raise ValueError('Logout token typ must be logout+jwt')
+
+        alg = header.get('alg')
+        if not isinstance(alg, str):
+            raise ValueError('Logout token header is missing alg')
+        if alg == 'none' or alg.startswith('HS'):
+            raise ValueError('Logout tokens must use an asymmetric signature')
+        if alg not in self._config.allowed_id_token_algorithms:
+            raise ValueError(f'Logout token algorithm {alg} is not allowed')
+
+        metadata_issuer = await self._get_metadata_issuer()
+        jwks = await self._ensure_jwks()
+        signing_key = self._get_signing_key(jwks, header.get('kid'), alg)
+        payload = jwt.decode(
+            logout_token,
+            signing_key.key,
+            algorithms=[alg],
+            audience=self._config.client_id,
+            leeway=self._config.backchannel_logout_leeway,
+            options={
+                'require': ['iss', 'aud', 'iat', 'jti', 'events'],
+                'verify_exp': True,
+                'verify_iat': True,
+                'verify_nbf': True,
+            },
+        )
+        if not isinstance(payload, dict):
+            raise ValueError('Logout token payload is not a JSON object')
+        self._validate_issuer(payload, metadata_issuer)
+
+        if not isinstance(payload.get('jti'), str):
+            raise ValueError('Logout token jti must be a string')
+        events = payload.get('events')
+        if (not isinstance(events, Mapping)
+                or not isinstance(events.get(_BACKCHANNEL_LOGOUT_EVENT), Mapping)):
+            raise ValueError('Logout token is missing the back-channel logout event')
+        if 'nonce' in payload:
+            raise ValueError('Logout token must not contain a nonce')
+
+        sub = payload.get('sub')
+        sid = payload.get('sid')
+        if sub is not None and not isinstance(sub, str):
+            raise ValueError('Logout token sub must be a string')
+        if sid is not None and not isinstance(sid, str):
+            raise ValueError('Logout token sid must be a string')
+        if sub is None and sid is None:
+            raise ValueError('Logout token must contain sub or sid')
+        return payload
+
    async def _decode_id_token(self, id_token: str) -> Mapping[str, Any]:
        header = jwt.get_unverified_header(id_token)
        alg = header.get('alg')
@@ -26,6 +26,12 @@ class OIDCConfig:
    A provider that sets ``require_pushed_authorization_requests`` is always
    treated as requiring PAR.

+    ``backchannel_logout_path`` enables OIDC Back-Channel Logout when set to
+    an application path. Register the corresponding absolute URL with the
+    provider. It requires a session store with index support.
+    ``backchannel_logout_leeway`` is the clock-skew allowance, in seconds,
+    used when validating logout tokens.
+
    ``client_secret`` is optional; omit it for public clients.
    ``fetch_userinfo`` controls whether the userinfo endpoint is queried after
    token exchange.
@@ -55,6 +61,8 @@ class OIDCConfig:
    http_client: Optional[httpx.AsyncClient] = None
    provider: str = 'default'
    par: PARPolicy = 'never'
+    backchannel_logout_path: Optional[str] = None
+    backchannel_logout_leeway: int = 60

    def __post_init__(self) -> None:
        if not self.provider or _PROVIDER_PATTERN.match(self.provider) is None:
@@ -68,6 +76,11 @@ class OIDCConfig:
                f'par must be one of {", ".join(map(repr, _PAR_POLICIES))} '
                f'(got {self.par!r})'
            )
+        if (self.backchannel_logout_path is not None
+                and not self.backchannel_logout_path.startswith('/')):
+            raise ValueError('backchannel_logout_path must start with "/"')
+        if self.backchannel_logout_leeway < 0:
+            raise ValueError('backchannel_logout_leeway must be non-negative')
        scopes_list = list(self.scopes)
        if 'openid' not in scopes_list:
            self.scopes = ('openid', *scopes_list)
@@ -15,8 +15,9 @@ type IdentityStrategy = Literal['single-identity', 'multi-identity']

 _SESSION_NAMESPACE = 'oidc'
 _USER_SUFFIX = 'user'
-_IDENTITY_FIELDS = frozenset({'user', 'id_token', 'access_token', 'refresh_token'})
+_IDENTITY_FIELDS = frozenset({'user', 'id_token', 'access_token', 'refresh_token', 'sid'})
 _IDENTITY_STRATEGIES: tuple[IdentityStrategy, ...] = ('single-identity', 'multi-identity')
+_BACKCHANNEL_BODY_LIMIT = 64 * 1024


 class OIDCUser(Mapping[str, Any]):
@@ -64,6 +65,10 @@ class OIDCMixin(KayaMixin):
    - ``callback_path`` (default ``/auth/callback``): handles the provider callback.
    - ``logout_path`` (default ``/auth/logout``): logs the user out.

+    When ``backchannel_logout_path`` is configured, a POST route accepting
+    OIDC Back-Channel Logout tokens is registered as well. This requires an
+    index-capable session store.
+
    Session state is namespaced by ``OIDCConfig.provider`` as
    ``oidc:{provider}:...`` so several providers can share one session. Use one
    ``OIDCMixin`` per provider (with distinct route paths), or
@@ -104,6 +109,11 @@ class OIDCMixin(KayaMixin):
        self._client = OIDCClient(config)
        self._strategy = strategy
        self._prefix = f'{_SESSION_NAMESPACE}:{config.provider}:'
+        if (config.backchannel_logout_path is not None
+                and not session.store.supports_indexes):
+            raise ValueError(
+                'backchannel_logout_path requires a session store with index support'
+            )

    @property
    def provider(self) -> str:
@@ -116,11 +126,73 @@ class OIDCMixin(KayaMixin):
    def _key(self, name: str) -> str:
        return f'{self._prefix}{name}'

+    def _index_name(self, claim: str, provider: Optional[str] = None) -> str:
+        return f'{_SESSION_NAMESPACE}:{provider or self.provider}:{claim}'
+
+    def _apply_identity_indexes(
+        self,
+        session: Session,
+        sub: str,
+        sid: Optional[str],
+    ) -> None:
+        session.add_index(self._index_name('sub'), sub)
+        if sid is not None:
+            session.add_index(self._index_name('sid'), sid)
+        session[self._key('indexed')] = True
+
+    def _remove_identity_indexes(
+        self,
+        session: Session,
+        provider: Optional[str] = None,
+    ) -> None:
+        prefix = f'{_SESSION_NAMESPACE}:{provider or self.provider}:'
+        user = session.get(f'{prefix}user')
+        if isinstance(user, Mapping):
+            sub = user.get('sub')
+            if isinstance(sub, str):
+                session.remove_index(self._index_name('sub', provider), sub)
+        sid = session.get(f'{prefix}sid')
+        if isinstance(sid, str):
+            session.remove_index(self._index_name('sid', provider), sid)
+
+    def _backfill_identity_indexes(self, session: Session) -> None:
+        """Index an authenticated session created before index support."""
+        user = session.get(self._key('user'))
+        if not isinstance(user, Mapping) or self._key('indexed') in session:
+            return
+        sub = user.get('sub')
+        if not isinstance(sub, str):
+            return
+
+        sid = session.get(self._key('sid'))
+        if not isinstance(sid, str):
+            id_token = session.get(self._key('id_token'))
+            if isinstance(id_token, str):
+                try:
+                    claims = jwt.decode(
+                        id_token,
+                        options={'verify_signature': False},
+                    )
+                    candidate = claims.get('sid')
+                    if isinstance(candidate, str):
+                        sid = candidate
+                        session[self._key('sid')] = candidate
+                except jwt.PyJWTError:
+                    pass
+        self._apply_identity_indexes(
+            session,
+            sub,
+            sid if isinstance(sid, str) else None,
+        )
+
    @property
    def dependencies(self) -> Sequence[KayaMixin]:
        return [self._session]

    def apply(self, app: KayaApp) -> None:
+        if self._config.backchannel_logout_path is not None:
+            app.add_before_request_hook(self._before_request)
+
        @app.GET(self._config.login_path)
        async def login(ctx: HttpContext) -> None:
            auth_url, state, nonce, code_verifier = await self._client.build_authorization_url(
@@ -168,6 +240,12 @@ class OIDCMixin(KayaMixin):
                    return

                claims = await self._client.validate_id_token(id_token, nonce)
+                sub = claims.get('sub')
+                if not isinstance(sub, str):
+                    raise ValueError('ID token is missing a valid sub claim')
+                sid = claims.get('sid')
+                if sid is not None and not isinstance(sid, str):
+                    raise ValueError('ID token sid must be a string')

                user: dict[str, Any] = dict(claims)
                access_token = token_response.get('access_token')
@@ -180,8 +258,13 @@ class OIDCMixin(KayaMixin):
                # Keep this provider's nonce so refreshed ID tokens can be
                # checked for nonce continuity (OIDC Core 12.2).

+                self._remove_identity_indexes(session)
                session[self._key('user')] = user
                session[self._key('id_token')] = id_token
+                session.pop(self._key('sid'), None)
+                if sid is not None:
+                    session[self._key('sid')] = sid
+                self._apply_identity_indexes(session, sub, sid)
                if isinstance(access_token, str):
                    session[self._key('access_token')] = access_token
                if 'refresh_token' in token_response:
@@ -204,6 +287,27 @@ class OIDCMixin(KayaMixin):
            location = logout_url if logout_url is not None else self._config.post_logout_redirect
            await ctx.send_empty(302, {'Location': location})

+        backchannel_path = self._config.backchannel_logout_path
+        if backchannel_path is not None:
+            @app.POST(backchannel_path)
+            async def backchannel_logout(ctx: HttpContext) -> None:
+                # This endpoint is server-to-server. Never persist a browser
+                # session that happens to accompany the request.
+                ctx.session = Session()
+                try:
+                    logout_token = await self._read_logout_token(ctx)
+                    claims = await self._client.validate_logout_token(logout_token)
+                except (ValueError, jwt.PyJWTError):
+                    await ctx.send_str(
+                        400,
+                        'Invalid logout token',
+                        {'Cache-Control': 'no-store'},
+                    )
+                    return
+
+                await self._invalidate_by_logout_token(claims)
+                await ctx.send_empty(200, {'Cache-Control': 'no-store'})
+
    def _logout_session(self, session: Session) -> None:
        """Drop this provider's session state.

@@ -211,6 +315,7 @@ class OIDCMixin(KayaMixin):
        logged in, so logging out of one provider does not sign the user out of
        the others.
        """
+        self._remove_identity_indexes(session)
        for key in [key for key in session if key.startswith(self._prefix)]:
            del session[key]
        if not any(_is_user_key(key) for key in session):
@@ -224,10 +329,75 @@ class OIDCMixin(KayaMixin):
        can still complete. Under this strategy the last callback to finish
        wins.
        """
+        other_providers = {
+            _provider_of_user_key(key)
+            for key in session
+            if _is_user_key(key) and not key.startswith(self._prefix)
+        }
+        for provider in other_providers:
+            self._remove_identity_indexes(session, provider)
        for key in [key for key in session
                    if _is_identity_key(key) and not key.startswith(self._prefix)]:
            del session[key]

+    async def _before_request(self, ctx: HttpContext) -> Optional[HttpContext]:
+        session = ctx.session
+        if isinstance(session, Session):
+            self._backfill_identity_indexes(session)
+        return None
+
+    async def _read_logout_token(self, ctx: HttpContext) -> str:
+        content_type_values = ctx.headers.get('content-type')
+        if isinstance(content_type_values, str):
+            content_type = content_type_values
+        elif content_type_values:
+            content_type = content_type_values[0]
+        else:
+            content_type = ''
+        if content_type.split(';', 1)[0].strip().lower() != 'application/x-www-form-urlencoded':
+            raise ValueError('Back-channel logout requires a form-encoded request')
+
+        body = bytearray()
+        async for chunk in ctx.request_body:
+            body.extend(chunk)
+            if len(body) > _BACKCHANNEL_BODY_LIMIT:
+                raise ValueError('Back-channel logout request is too large')
+        form = parse_qs(body.decode('utf-8'), keep_blank_values=True)
+        logout_token = self._first_value(form.get('logout_token'))
+        if logout_token is None:
+            raise ValueError('Missing logout_token')
+        return logout_token
+
+    async def _invalidate_by_logout_token(self, claims: Mapping[str, Any]) -> None:
+        store = self._session.store
+        sub = claims.get('sub')
+        sid = claims.get('sid')
+        if isinstance(sid, str):
+            session_ids = await store.find_sessions(self._index_name('sid'), sid)
+        elif isinstance(sub, str):
+            session_ids = await store.find_sessions(self._index_name('sub'), sub)
+        else:
+            return
+
+        for session_id in session_ids:
+            session = await store.load(session_id)
+            if session is None:
+                continue
+            if isinstance(sid, str) and isinstance(sub, str):
+                user = session.get(self._key('user'))
+                if not isinstance(user, Mapping) or user.get('sub') != sub:
+                    continue
+
+            # A session can hold several provider identities. Remove only this
+            # provider's identity and delete the whole session when it was the
+            # last one, mirroring the local logout behavior.
+            self._logout_session(session)
+            if session.invalidated:
+                await store.delete(session_id)
+            else:
+                await store.save(session_id, session, self._session.max_age)
+                session.commit_indexes()
+
    @staticmethod
    def _session_of(ctx: HttpContext) -> Session:
        session = ctx.session
@@ -293,6 +463,15 @@ class OIDCMixin(KayaMixin):
                if isinstance(original_sub, str) and claims.get('sub') != original_sub:
                    raise ValueError('ID token subject mismatch')

+                refreshed_sid = claims.get('sid')
+                stored_sid = session.get(self._key('sid'))
+                if (isinstance(refreshed_sid, str)
+                        and refreshed_sid != stored_sid):
+                    if isinstance(stored_sid, str):
+                        session.remove_index(self._index_name('sid'), stored_sid)
+                    session[self._key('sid')] = refreshed_sid
+                    session.add_index(self._index_name('sid'), refreshed_sid)
+
            if isinstance(access_token, str):
                session[self._key('access_token')] = access_token
            if isinstance(id_token, str):
@@ -20,7 +20,9 @@ class OIDCProviders(KayaMixin):

    Each config's ``provider`` becomes a route segment under ``base_path``:
    ``{base_path}/{provider}/login``, ``{base_path}/{provider}/callback`` and
-    ``{base_path}/{provider}/logout``. All providers share the supplied
+    ``{base_path}/{provider}/logout``. Providers that enable back-channel
+    logout also get ``{base_path}/{provider}/backchannel-logout``. All
+    providers share the supplied
    :class:`~kaya.session.SessionMixin` and namespace their session state by
    provider.

@@ -68,12 +70,17 @@ class OIDCProviders(KayaMixin):
            name = config.provider
            if name in self._providers:
                raise ValueError(f'Duplicate OIDC provider {name!r}')
+            backchannel_path = (
+                f'{self._base_path}/{name}/backchannel-logout'
+                if config.backchannel_logout_path is not None else None
+            )
            self._providers[name] = OIDCMixin(
                replace(
                    config,
                    login_path=f'{self._base_path}/{name}/login',
                    callback_path=f'{self._base_path}/{name}/callback',
                    logout_path=f'{self._base_path}/{name}/logout',
+                    backchannel_logout_path=backchannel_path,
                ),
                session=session,
                strategy=strategy,
@@ -0,0 +1,427 @@
+import unittest
+from time import time
+from typing import Any, Optional
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+import jwt
+from cryptography.hazmat.primitives.asymmetric import rsa
+from pwo import async_test
+from test_oidc import MockOIDCProvider, MockTransport
+
+from kaya.core import HttpContext, KayaApp
+from kaya.oidc import OIDCConfig, OIDCMixin, OIDCProviders
+from kaya.session import InMemorySessionStore, SessionMixin
+
+_EVENT = 'http://schemas.openid.net/event/backchannel-logout'
+
+
+class BackchannelProvider(MockOIDCProvider):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.sid: Optional[str] = None
+
+    def issue_id_token(self, *args: Any, **kwargs: Any) -> str:
+        token = super().issue_id_token(*args, **kwargs)
+        if self.sid is None:
+            return token
+        payload = jwt.decode(token, options={'verify_signature': False})
+        payload['sid'] = self.sid
+        return jwt.encode(
+            payload,
+            self.private_key,
+            algorithm='RS256',
+            headers={'kid': self.kid},
+        )
+
+
+class BackchannelLogoutTest(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.provider = BackchannelProvider()
+        self.http_client = httpx.AsyncClient(transport=MockTransport(self.provider))
+        self.store = InMemorySessionStore()
+        self.session = SessionMixin(self.store)
+        self.oidc = OIDCMixin(
+            OIDCConfig(
+                issuer=self.provider.issuer,
+                client_id='client',
+                client_secret='secret',
+                redirect_uri='http://localhost:8000/auth/callback',
+                backchannel_logout_path='/auth/backchannel-logout',
+                http_client=self.http_client,
+            ),
+            session=self.session,
+        )
+        self.app = KayaApp(mixins=[self.session, self.oidc])
+        self.transport = httpx.ASGITransport(app=self.app)
+
+        @self.app.GET('/profile')
+        @self.oidc.require_auth
+        async def profile(ctx: HttpContext) -> None:
+            user = self.oidc.get_user(ctx)
+            assert user is not None
+            await ctx.send_str(200, user.sub)
+
+        @self.app.GET('/refresh')
+        @self.oidc.require_auth
+        async def refresh(ctx: HttpContext) -> None:
+            token = await self.oidc.refresh_access_token(ctx.session)
+            await ctx.send_str(200, token or 'no-token')
+
+    def _client(self) -> httpx.AsyncClient:
+        return httpx.AsyncClient(transport=self.transport, base_url='http://127.0.0.1:80')
+
+    async def _login(self, client: httpx.AsyncClient, sid: str) -> None:
+        self.provider.sid = sid
+        response = await client.get('/auth/login', follow_redirects=False)
+        self.assertEqual(302, response.status_code)
+        query = parse_qs(urlparse(response.headers['Location']).query)
+        self.provider.nonce = query['nonce'][0]
+        response = await client.get(
+            '/auth/callback',
+            params={'code': 'mock-code', 'state': query['state'][0]},
+            follow_redirects=False,
+        )
+        self.assertEqual(302, response.status_code)
+
+    def _claims(self, **overrides: Any) -> dict[str, Any]:
+        claims: dict[str, Any] = {
+            'iss': self.provider.issuer,
+            'aud': 'client',
+            'iat': time(),
+            'jti': 'logout-token-id',
+            'events': {_EVENT: {}},
+            'sub': self.provider.sub,
+        }
+        claims.update(overrides)
+        return claims
+
+    def _token(
+        self,
+        claims: dict[str, Any],
+        *,
+        typ: str = 'logout+jwt',
+        key: Any = None,
+        algorithm: str = 'RS256',
+    ) -> str:
+        return jwt.encode(
+            claims,
+            self.provider.private_key if key is None else key,
+            algorithm=algorithm,
+            headers={'kid': self.provider.kid, 'typ': typ},
+        )
+
+    async def _post_logout(self, token: str) -> httpx.Response:
+        async with self._client() as client:
+            return await client.post(
+                '/auth/backchannel-logout',
+                data={'logout_token': token},
+            )
+
+    async def _profile_status(self, client: httpx.AsyncClient) -> int:
+        response = await client.get('/profile', follow_redirects=False)
+        return response.status_code
+
+    @async_test
+    async def test_logout_by_sid_only_invalidates_matching_session(self) -> None:
+        async with self._client() as first, self._client() as second:
+            await self._login(first, 'sid-a')
+            await self._login(second, 'sid-b')
+
+            response = await self._post_logout(self._token(self._claims(sid='sid-a')))
+            self.assertEqual(200, response.status_code)
+            self.assertEqual('no-store', response.headers['Cache-Control'])
+            self.assertNotIn('Set-Cookie', response.headers)
+            self.assertEqual(302, await self._profile_status(first))
+            self.assertEqual(200, await self._profile_status(second))
+            self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a'))
+            self.assertEqual(
+                [second.cookies['session_id']],
+                await self.store.find_sessions('oidc:default:sid', 'sid-b'),
+            )
+
+    @async_test
+    async def test_logout_by_sub_invalidates_all_user_sessions(self) -> None:
+        async with self._client() as first, self._client() as second, self._client() as other:
+            await self._login(first, 'sid-a')
+            await self._login(second, 'sid-b')
+            self.provider.sub = 'bob'
+            await self._login(other, 'sid-c')
+
+            response = await self._post_logout(self._token(self._claims(sub='user123')))
+            self.assertEqual(200, response.status_code)
+            self.assertEqual(302, await self._profile_status(first))
+            self.assertEqual(302, await self._profile_status(second))
+            self.assertEqual(200, await self._profile_status(other))
+
+    @async_test
+    async def test_sid_logout_verifies_sub_when_both_are_present(self) -> None:
+        async with self._client() as client:
+            await self._login(client, 'sid-a')
+            response = await self._post_logout(
+                self._token(self._claims(sub='someone-else', sid='sid-a'))
+            )
+            self.assertEqual(200, response.status_code)
+            self.assertEqual(200, await self._profile_status(client))
+
+    @async_test
+    async def test_unknown_session_is_idempotent(self) -> None:
+        response = await self._post_logout(self._token(self._claims(sid='unknown')))
+        self.assertEqual(200, response.status_code)
+
+    @async_test
+    async def test_invalid_logout_tokens_are_rejected(self) -> None:
+        other_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        missing_iat = self._claims(sid='sid-a')
+        del missing_iat['iat']
+        missing_jti = self._claims(sid='sid-a')
+        del missing_jti['jti']
+        missing_events = self._claims(sid='sid-a')
+        del missing_events['events']
+        missing_subject = self._claims()
+        del missing_subject['sub']
+
+        invalid_tokens = [
+            self._token(self._claims(sid='sid-a'), typ='JWT'),
+            jwt.encode(
+                self._claims(sid='sid-a'),
+                None,
+                algorithm='none',
+                headers={'typ': 'logout+jwt'},
+            ),
+            self._token(self._claims(sid='sid-a'), key=other_key),
+            self._token(self._claims(iss='https://evil.local', sid='sid-a')),
+            self._token(self._claims(aud='other-client', sid='sid-a')),
+            self._token(missing_iat),
+            self._token(missing_jti),
+            self._token(missing_events),
+            self._token(self._claims(events={'wrong-event': {}}, sid='sid-a')),
+            self._token(self._claims(events={_EVENT: None}, sid='sid-a')),
+            self._token(self._claims(nonce='forbidden', sid='sid-a')),
+            self._token(missing_subject),
+            self._token(self._claims(exp=time() - 3600, sid='sid-a')),
+            self._token(self._claims(sid='sid-a'), key='x' * 32, algorithm='HS256'),
+        ]
+
+        async with self._client() as client:
+            await self._login(client, 'sid-a')
+            for token in invalid_tokens:
+                response = await self._post_logout(token)
+                self.assertEqual(400, response.status_code)
+            self.assertEqual(200, await self._profile_status(client))
+
+    @async_test
+    async def test_invalid_requests_are_rejected(self) -> None:
+        async with self._client() as client:
+            response = await client.post('/auth/backchannel-logout', data={})
+            self.assertEqual(400, response.status_code)
+            response = await client.post(
+                '/auth/backchannel-logout',
+                content='logout_token=abc',
+                headers={'Content-Type': 'text/plain'},
+            )
+            self.assertEqual(400, response.status_code)
+            response = await client.post(
+                '/auth/backchannel-logout',
+                content='x' * (64 * 1024 + 1),
+                headers={'Content-Type': 'application/x-www-form-urlencoded'},
+            )
+            self.assertEqual(400, response.status_code)
+
+    @async_test
+    async def test_local_logout_removes_indexes(self) -> None:
+        async with self._client() as client:
+            await self._login(client, 'sid-a')
+            self.assertEqual(
+                [client.cookies['session_id']],
+                await self.store.find_sessions('oidc:default:sub', 'user123'),
+            )
+            await client.get('/auth/logout', follow_redirects=False)
+            self.assertEqual([], await self.store.find_sessions('oidc:default:sub', 'user123'))
+            self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a'))
+
+    @async_test
+    async def test_relogin_replaces_old_sid_index(self) -> None:
+        async with self._client() as client:
+            await self._login(client, 'sid-a')
+            await self._login(client, 'sid-b')
+            self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a'))
+            self.assertEqual(
+                [client.cookies['session_id']],
+                await self.store.find_sessions('oidc:default:sid', 'sid-b'),
+            )
+
+    @async_test
+    async def test_refresh_updates_changed_sid_index(self) -> None:
+        async with self._client() as client:
+            await self._login(client, 'sid-a')
+            self.provider.sid = 'sid-b'
+            response = await client.get('/refresh')
+            self.assertEqual(200, response.status_code)
+            self.assertNotEqual('no-token', response.text)
+            self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a'))
+            self.assertEqual(
+                [client.cookies['session_id']],
+                await self.store.find_sessions('oidc:default:sid', 'sid-b'),
+            )
+
+    @async_test
+    async def test_existing_session_is_backfilled(self) -> None:
+        async with self._client() as client:
+            await self._login(client, 'sid-a')
+            session_id = client.cookies['session_id']
+            self.store._data[session_id].pop('oidc:default:indexed')
+            self.store._session_indexes.pop(session_id, None)
+            for session_ids in self.store._index_sessions.values():
+                session_ids.discard(session_id)
+            self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a'))
+
+            self.assertEqual(200, await self._profile_status(client))
+            self.assertEqual(
+                [session_id],
+                await self.store.find_sessions('oidc:default:sid', 'sid-a'),
+            )
+
+    @async_test
+    async def test_index_capable_store_is_required(self) -> None:
+        class LegacyStore(InMemorySessionStore):
+            supports_indexes = False
+
+        with self.assertRaises(ValueError):
+            OIDCMixin(
+                OIDCConfig(
+                    issuer=self.provider.issuer,
+                    client_id='client',
+                    redirect_uri='http://localhost:8000/auth/callback',
+                    backchannel_logout_path='/auth/backchannel-logout',
+                ),
+                session=SessionMixin(LegacyStore()),
+            )
+
+
+class MultiProviderBackchannelLogoutTest(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.google = BackchannelProvider('google', sub='shared-sub', email='g@example.com')
+        self.github = BackchannelProvider('github', sub='shared-sub', email='h@example.com')
+        self.store = InMemorySessionStore()
+        self.session = SessionMixin(self.store)
+        self.oidc = OIDCProviders(
+            [
+                self._config(self.google, 'google'),
+                self._config(self.github, 'github'),
+            ],
+            session=self.session,
+            strategy='multi-identity',
+        )
+        self.app = KayaApp(mixins=[self.session, self.oidc])
+        self.transport = httpx.ASGITransport(app=self.app)
+
+        @self.app.GET('/google-profile')
+        @self.oidc.require_auth('google')
+        async def google_profile(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'google')
+
+        @self.app.GET('/github-profile')
+        @self.oidc.require_auth('github')
+        async def github_profile(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'github')
+
+    @staticmethod
+    def _config(provider: BackchannelProvider, name: str) -> OIDCConfig:
+        return OIDCConfig(
+            issuer=provider.issuer,
+            client_id='client',
+            client_secret='secret',
+            redirect_uri=f'http://localhost:8000/auth/{name}/callback',
+            provider=name,
+            backchannel_logout_path=f'/auth/{name}/backchannel-logout',
+            http_client=httpx.AsyncClient(transport=MockTransport(provider)),
+        )
+
+    async def _login(
+        self,
+        client: httpx.AsyncClient,
+        provider: BackchannelProvider,
+        name: str,
+        sid: str,
+    ) -> None:
+        provider.sid = sid
+        response = await client.get(f'/auth/{name}/login', follow_redirects=False)
+        query = parse_qs(urlparse(response.headers['Location']).query)
+        provider.nonce = query['nonce'][0]
+        await client.get(
+            f'/auth/{name}/callback',
+            params={'code': 'mock-code', 'state': query['state'][0]},
+            follow_redirects=False,
+        )
+
+    @async_test
+    async def test_single_identity_eviction_removes_old_provider_indexes(self) -> None:
+        session = SessionMixin(self.store)
+        oidc = OIDCProviders(
+            [
+                self._config(self.google, 'google'),
+                self._config(self.github, 'github'),
+            ],
+            session=session,
+            strategy='single-identity',
+        )
+        app = KayaApp(mixins=[session, oidc])
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport,
+            base_url='http://127.0.0.1:80',
+        ) as client:
+            await self._login(client, self.google, 'google', 'google-sid')
+            self.assertEqual(
+                [client.cookies['session_id']],
+                await self.store.find_sessions('oidc:google:sid', 'google-sid'),
+            )
+            await self._login(client, self.github, 'github', 'github-sid')
+            self.assertEqual(
+                [],
+                await self.store.find_sessions('oidc:google:sid', 'google-sid'),
+            )
+            self.assertEqual(
+                [client.cookies['session_id']],
+                await self.store.find_sessions('oidc:github:sid', 'github-sid'),
+            )
+
+    @async_test
+    async def test_provider_logout_is_isolated(self) -> None:
+        async with httpx.AsyncClient(
+            transport=self.transport,
+            base_url='http://127.0.0.1:80',
+        ) as client:
+            await self._login(client, self.google, 'google', 'google-sid')
+            await self._login(client, self.github, 'github', 'github-sid')
+            token = jwt.encode(
+                {
+                    'iss': self.google.issuer,
+                    'aud': 'client',
+                    'iat': time(),
+                    'jti': 'google-logout',
+                    'events': {_EVENT: {}},
+                    'sub': 'shared-sub',
+                    'sid': 'google-sid',
+                },
+                self.google.private_key,
+                algorithm='RS256',
+                headers={'kid': self.google.kid, 'typ': 'logout+jwt'},
+            )
+            response = await client.post(
+                '/auth/google/backchannel-logout',
+                data={'logout_token': token},
+            )
+            self.assertEqual(200, response.status_code)
+
+            response = await client.get('/google-profile', follow_redirects=False)
+            self.assertEqual(302, response.status_code)
+            response = await client.get('/github-profile', follow_redirects=False)
+            self.assertEqual(200, response.status_code)
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -69,6 +69,48 @@ object, so any valid OpenAPI 3.1 construct can be used:
 `parameters` entries whose `name` and `in` match an auto-generated path
 parameter override it; all others are appended.

+## Reusable schemas (`components`)
+
+Schemas can be defined once and referenced from operation fragments: pass a
+`components` mapping to the mixin and use `schema_ref` in `@operation`.
+
+```python
+from kaya.openapi import OpenAPIMixin, operation, schema_ref
+
+USER = {
+    'type': 'object',
+    'required': ['id'],
+    'properties': {'id': {'type': 'integer'}, 'name': {'type': 'string'}},
+}
+USER_PAGE = {
+    'type': 'object',
+    'properties': {'results': {'type': 'array', 'items': schema_ref('User')}},
+}
+
+app = KayaApp(mixins=[OpenAPIMixin(
+    title='My API',
+    version='1.0.0',
+    components={'schemas': {'User': USER, 'UserPage': USER_PAGE}},
+)])
+
+
+@app.GET('/users')
+@operation(summary='List users',
+           responses={200: {
+               'description': 'A page of users',
+               'content': {'application/json': {'schema': schema_ref('UserPage')}},
+           }})
+async def list_users(ctx: HttpContext) -> None:
+    ...
+```
+
+`components` is merged verbatim (as a copy) into the generated document under
+`components`, so Swagger UI lists the named schemas. Every
+`#/components/<section>/<name>` reference — in operations or between
+components — is validated when the document is generated: a missing section or
+entry raises `ValueError`. Reflected references (`./other.json#/...`, absolute
+URLs) are left alone.
+
 ## Configuration

 ```python
@@ -81,6 +123,7 @@ OpenAPIMixin(
    servers=[{'url': 'https://api.example.com'}],
    openapi_version='3.1.0',
    include_self=False,            # include spec/docs endpoints in the document
+    components=None,               # reusable OpenAPI components (e.g. schemas)
 )
 ```

@@ -25,11 +25,6 @@ dependencies = [
    "kaya-core",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine", "httpx", "httpx-ws"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -1,10 +1,11 @@
 from ._metadata import operation
 from ._mixin import OpenAPIMixin
-from ._spec import generate_spec
+from ._spec import generate_spec, schema_ref


 __all__ = [
    'OpenAPIMixin',
    'generate_spec',
    'operation',
+    'schema_ref',
 ]
@@ -25,7 +25,9 @@ def operation(summary: Optional[str] = None,
    :class:`~kaya.openapi.OpenAPIMixin` when generating the specification.

    All schema fragments are plain dicts inserted verbatim into the generated
-    OpenAPI document, so any valid OpenAPI 3.1 construct can be used.
+    OpenAPI document, so any valid OpenAPI 3.1 construct can be used. ``$ref``
+    fragments may point at the reusable schemas registered through the
+    ``components`` argument of :class:`~kaya.openapi.OpenAPIMixin`.

    Example::

@@ -55,6 +55,11 @@ class OpenAPIMixin(KayaMixin):
    bodies and response schemas to individual handlers; the handler docstring
    is used as the operation description when no explicit one is given.

+    Reusable schemas are supplied through ``components`` and referenced from
+    operation fragments with :func:`~kaya.openapi.schema_ref`; every
+    ``#/components/...`` reference is validated when the document is
+    generated.
+
    Example::

        openapi = OpenAPIMixin(title='My API', version='1.0.0')
@@ -74,7 +79,8 @@ class OpenAPIMixin(KayaMixin):
                 docs_path: str = '/docs',
                 servers: Optional[Sequence[Mapping[str, Any]]] = None,
                 openapi_version: str = '3.1.0',
-                 include_self: bool = False) -> None:
+                 include_self: bool = False,
+                 components: Optional[Mapping[str, Any]] = None) -> None:
        self._title = title
        self._version = version
        self._description = description
@@ -83,6 +89,7 @@ class OpenAPIMixin(KayaMixin):
        self._servers = servers
        self._openapi_version = openapi_version
        self._include_self = include_self
+        self._components = components

    def apply(self, app: KayaApp) -> None:
        @app.GET(self._spec_path)
@@ -107,4 +114,5 @@ class OpenAPIMixin(KayaMixin):
                             description=self._description,
                             servers=self._servers,
                             exclude_paths=exclude_paths,
-                             openapi_version=self._openapi_version)
+                             openapi_version=self._openapi_version,
+                             components=self._components)
@@ -1,7 +1,7 @@
 from copy import deepcopy
 from dataclasses import dataclass
 from inspect import getdoc
-from typing import AbstractSet, Any, Mapping, Optional, Sequence
+from typing import AbstractSet, Any, Dict, Iterator, Mapping, Optional, Sequence

 from kaya.core import HttpMethod, KayaApp
 from kaya.core._path_handler import PathHandler
@@ -22,6 +22,17 @@ STANDARD_METHODS: Sequence[HttpMethod] = (
 )


+def schema_ref(name: str) -> Dict[str, str]:
+    """Return a reference to the component schema ``name``.
+
+    The reference resolves to ``#/components/schemas/<name>`` in the
+    generated document; the schema itself is supplied through the
+    ``components`` argument of :func:`generate_spec` (or
+    :class:`~kaya.openapi.OpenAPIMixin`).
+    """
+    return {'$ref': f'#/components/schemas/{name}'}
+
+
@dataclass
 class _Route:
    raw_path: str
@@ -122,13 +133,67 @@ def _build_operation(params: Sequence[Mapping[str, Any]],
    return operation


+def _unescape_pointer_token(token: str) -> str:
+    """Decode one JSON Pointer reference token (``~1`` → ``/``, ``~0`` → ``~``)."""
+    return token.replace('~1', '/').replace('~0', '~')
+
+
+def _iter_refs(node: Any) -> Iterator[str]:
+    """Yield every ``$ref`` string found anywhere under ``node``."""
+    if isinstance(node, Mapping):
+        ref = node.get('$ref')
+        if isinstance(ref, str):
+            yield ref
+        for value in node.values():
+            yield from _iter_refs(value)
+    elif isinstance(node, (list, tuple)):
+        for value in node:
+            yield from _iter_refs(value)
+
+
+def _validate_component_refs(spec: Mapping[str, Any]) -> None:
+    """Fail fast on references to missing ``components`` entries.
+
+    Every ``#/components/<section>/<name>`` reference found in the operations
+    or in the components themselves must resolve against the document;
+    references in any other form (relative files, absolute URLs, other
+    pointers) are ignored. Raises :class:`ValueError` naming the offending
+    reference and the operation it appears in.
+    """
+    components = spec.get('components')
+    known: Mapping[str, Any] = components if isinstance(components, Mapping) else {}
+
+    def check(node: Any, location: str) -> None:
+        for ref in _iter_refs(node):
+            if not ref.startswith('#/components/'):
+                continue
+            parts = [_unescape_pointer_token(part) for part in ref[len('#/components/'):].split('/')]
+            if len(parts) < 2:
+                raise ValueError(f'invalid component reference {ref!r} in {location}')
+            section, name = parts[0], parts[1]
+            bucket = known.get(section)
+            if not isinstance(bucket, Mapping) or name not in bucket:
+                raise ValueError(f'unknown component reference {ref!r} in {location}')
+
+    paths = spec.get('paths')
+    if isinstance(paths, Mapping):
+        for path, item in paths.items():
+            if not isinstance(item, Mapping):
+                continue
+            for method, operation in item.items():
+                check(operation, f'{method.upper()} {path}')
+    if isinstance(components, Mapping):
+        check(components, 'components')
+
+
 def generate_spec(app: KayaApp,
                  title: str,
                  version: str,
                  description: str = '',
                  servers: Optional[Sequence[Mapping[str, Any]]] = None,
                  exclude_paths: AbstractSet[str] = frozenset(),
-                  openapi_version: str = '3.1.0') -> Mapping[str, Any]:
+                  openapi_version: str = '3.1.0',
+                  components: Optional[Mapping[str, Any]] = None) -> Mapping[str, Any]:
    """Generate an OpenAPI specification document from a :class:`KayaApp`.

    The application's routing tree is walked and every route is converted to
@@ -147,6 +212,9 @@ def generate_spec(app: KayaApp,
    :param servers: list of OpenAPI server objects
    :param exclude_paths: Kaya paths (e.g. ``/openapi.json``) to omit
    :param openapi_version: OpenAPI version to declare
+    :param components: OpenAPI ``components`` object merged verbatim into the
+        document; references to ``#/components/<section>/<name>`` must resolve
+        against it (e.g. :func:`schema_ref` for reusable schemas)
    :return: the OpenAPI document as a JSON-serializable mapping
    """
    routes: list[_Route] = []
@@ -174,4 +242,7 @@ def generate_spec(app: KayaApp,
    }
    if servers:
        spec['servers'] = [dict(server) for server in servers]
+    if components:
+        spec['components'] = deepcopy(dict(components))
+    _validate_component_refs(spec)
    return spec
@@ -6,7 +6,7 @@ import httpx
 from pwo import async_test

 from kaya.core import HttpContext, HttpMethod, KayaApp, WebSocket
-from kaya.openapi import OpenAPIMixin, operation
+from kaya.openapi import OpenAPIMixin, generate_spec, operation, schema_ref


 class OpenAPITest(unittest.TestCase):
@@ -163,5 +163,127 @@ class OpenAPITest(unittest.TestCase):
            self.assertIn('/spec.json', response.text)


+class ComponentsTest(unittest.TestCase):
+    """Reusable schemas through ``components`` and ``schema_ref``."""
+
+    async def _spec_of(self, app: KayaApp) -> Mapping[str, Any]:
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            response = await client.get('/openapi.json')
+            self.assertEqual(200, response.status_code)
+            return json.loads(response.text)
+
+    def test_schema_ref_helper(self) -> None:
+        self.assertEqual({'$ref': '#/components/schemas/User'}, schema_ref('User'))
+
+    @async_test
+    async def test_components_merged_and_refs_preserved(self) -> None:
+        user = {'type': 'object', 'properties': {'id': {'type': 'integer'}}}
+        page = {'type': 'object',
+                'properties': {'results': {'type': 'array', 'items': schema_ref('User')}}}
+        app = KayaApp(mixins=[OpenAPIMixin(
+            title='Components API',
+            version='1.0.0',
+            components={'schemas': {'User': user, 'UserPage': page}},
+        )])
+
+        @app.GET('/users')
+        @operation(responses={200: {
+            'description': 'A page of users',
+            'content': {'application/json': {'schema': schema_ref('UserPage')}},
+        }})
+        async def list_users(ctx: HttpContext) -> None:
+            await ctx.send_str(200, '[]')
+
+        spec = await self._spec_of(app)
+        self.assertEqual({'schemas': {'User': user, 'UserPage': page}}, spec['components'])
+        schema = spec['paths']['/users']['get']['responses']['200']['content']['application/json']['schema']
+        self.assertEqual(schema_ref('UserPage'), schema)
+
+    @async_test
+    async def test_request_body_ref_preserved(self) -> None:
+        app = KayaApp(mixins=[OpenAPIMixin(
+            title='Components API',
+            version='1.0.0',
+            components={'schemas': {'User': {'type': 'object'}}},
+        )])
+
+        @app.POST('/users')
+        @operation(request_body={'required': True,
+                                 'content': {'application/json': {'schema': schema_ref('User')}}},
+                   responses={201: {'description': 'Created'}})
+        async def create_user(ctx: HttpContext) -> None:
+            await ctx.send_str(201, '{}')
+
+        spec = await self._spec_of(app)
+        schema = spec['paths']['/users']['post']['requestBody']['content']['application/json']['schema']
+        self.assertEqual(schema_ref('User'), schema)
+
+    @async_test
+    async def test_default_spec_has_no_components(self) -> None:
+        app = KayaApp(mixins=[OpenAPIMixin(title='Plain API', version='1.0.0')])
+
+        @app.GET('/plain')
+        async def plain(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'ok')
+
+        spec = await self._spec_of(app)
+        self.assertNotIn('components', spec)
+
+    def test_unknown_schema_ref_raises(self) -> None:
+        app = KayaApp()
+
+        @app.GET('/broken')
+        @operation(responses={200: {
+            'description': 'x',
+            'content': {'application/json': {'schema': schema_ref('Missing')}},
+        }})
+        async def broken(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'x')
+
+        with self.assertRaises(ValueError) as caught:
+            generate_spec(app, title='Broken', version='1.0.0',
+                          components={'schemas': {'User': {}}})
+        message = str(caught.exception)
+        self.assertIn('#/components/schemas/Missing', message)
+        self.assertIn('GET /broken', message)
+
+    def test_component_to_component_ref_validates(self) -> None:
+        app = KayaApp()
+        spec = generate_spec(app, title='Nested', version='1.0.0', components={'schemas': {
+            'User': {'type': 'object'},
+            'Page': {'type': 'object',
+                     'properties': {'results': {'type': 'array', 'items': schema_ref('User')}}},
+        }})
+        self.assertIn('components', spec)
+
+    def test_external_refs_are_ignored(self) -> None:
+        app = KayaApp()
+
+        @app.GET('/external')
+        @operation(responses={200: {
+            'description': 'x',
+            'content': {'application/json': {'schema': {'$ref': './other.json#/Thing'}}},
+        }})
+        async def external(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'x')
+
+        spec = generate_spec(app, title='External', version='1.0.0')
+        self.assertNotIn('components', spec)
+
+    def test_generic_component_section_validated(self) -> None:
+        app = KayaApp()
+
+        @app.GET('/p')
+        @operation(parameters=[{'name': 'q', 'in': 'query',
+                                'schema': {'$ref': '#/components/parameters/Missing'}}])
+        async def p(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'x')
+
+        with self.assertRaises(ValueError) as caught:
+            generate_spec(app, title='P', version='1.0.0', components={'parameters': {}})
+        self.assertIn('#/components/parameters/Missing', str(caught.exception))
+
+
 if __name__ == '__main__':
    unittest.main()
@@ -27,11 +27,6 @@ dependencies = [
    "opentelemetry-exporter-otlp-proto-http",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine", "httpx", "httpx-ws", "kaya-rsgi"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -27,11 +27,6 @@ dependencies = [
    "pwo",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -32,11 +32,25 @@ and slides on each access when the session mixin passes a `max_age`. TTLs
 larger than 30 days are automatically converted to absolute Unix timestamps,
 as required by the memcached protocol.

+## Session indexes
+
+`MemcacheSessionStore` supports `Session.add_index()`,
+`SessionStore.find_sessions()`, and `delete_by_index()`. Reverse memberships
+are stored as sets and updated with memcached CAS operations. Bucket keys use
+a SHA-256 digest of the index name and value, and stale members are pruned
+during lookups.
+
+Memcached's item-size limit (normally 1 MB) applies to each index bucket. Use
+the Redis store when one index value can have a very large number of active
+sessions.
+
 ## Serialization

 Session data is serialized with `pickle` by default, so arbitrary Python
-objects can be stored. A different serializer can be plugged in via the
-`dumps`/`loads` arguments:
+objects can be stored. The serialized value is an envelope containing the
+session dictionary and index metadata; legacy plain-dict values are still
+accepted. A different serializer can be plugged in via the `dumps`/`loads`
+arguments:

 ```python
 import json
@@ -26,11 +26,6 @@ dependencies = [
    "aiomcache>=0.8",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -1,12 +1,15 @@
+import hashlib
 import pickle
 from time import time
-from typing import Any, Callable, Optional
+from typing import Any, Callable, Collection, Optional, Sequence

 import aiomcache

-from kaya.session import Session, SessionStore
+from kaya.session import IndexKey, Session, SessionStore

 _THIRTY_DAYS = 30 * 24 * 60 * 60
+_INDEX_GRACE_PERIOD = 60
+_CAS_RETRIES = 5


 def _exptime(max_age: int, clock: Callable[[], float]) -> int:
@@ -36,8 +39,14 @@ class MemcacheSessionStore(SessionStore):
    Python objects can be stored. Custom serializers can be plugged in via
    the ``dumps``/``loads`` arguments. Only connect this store to a trusted
    memcached server, as pickle deserialization of untrusted data is unsafe.
+
+    Reverse index buckets are pickled sets updated with memcached CAS
+    operations. Memcached's item-size limit applies to each bucket; Redis is a
+    better fit when one index value can have a very large number of sessions.
    """

+    supports_indexes = True
+
    def __init__(
        self,
        client: aiomcache.Client,
@@ -55,6 +64,69 @@ class MemcacheSessionStore(SessionStore):
    def _key(self, session_id: str) -> bytes:
        return f"{self._prefix}{session_id}".encode()

+    def _index_key(self, name: str, value: str) -> bytes:
+        digest = hashlib.sha256(f"{name}\0{value}".encode()).hexdigest()
+        return f"{self._prefix}idx:{digest}".encode()
+
+    def _index_exptime(self, max_age: Optional[int]) -> int:
+        if max_age is None:
+            return 0
+        return _exptime(max_age + _INDEX_GRACE_PERIOD, self._clock)
+
+    @staticmethod
+    def _decode_bucket(payload: bytes) -> tuple[set[str], Optional[int]]:
+        members, max_age = pickle.loads(payload)
+        return set(members), max_age
+
+    async def _update_bucket(
+        self,
+        index_key: bytes,
+        *,
+        add: Collection[str] = (),
+        remove: Collection[str] = (),
+        max_age: Optional[int] = None,
+    ) -> None:
+        for _ in range(_CAS_RETRIES):
+            payload, cas_token = await self._client.gets(index_key)
+            if cas_token is None:
+                if not add:
+                    return
+                members = set(add)
+                created = await self._client.add(
+                    index_key,
+                    pickle.dumps((members, max_age)),
+                    exptime=self._index_exptime(max_age),
+                )
+                if created:
+                    return
+                continue
+
+            assert payload is not None
+            members, stored_max_age = self._decode_bucket(payload)
+            updated = (members | set(add)) - set(remove)
+            if updated == members:
+                if not add:
+                    return
+                if await self._client.touch(
+                    index_key,
+                    self._index_exptime(max_age),
+                ):
+                    return
+                # The bucket may have expired between gets() and touch().
+                # Retry so it can be recreated instead of losing the index.
+                continue
+
+            effective_max_age = max_age if add else stored_max_age
+            if await self._client.cas(
+                index_key,
+                pickle.dumps((updated, effective_max_age)),
+                cas_token,
+                exptime=self._index_exptime(effective_max_age),
+            ):
+                return
+
+        raise RuntimeError('Could not update session index after repeated CAS conflicts')
+
    async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]:
        key = self._key(session_id)
        payload = await self._client.get(key)
@@ -65,12 +137,63 @@ class MemcacheSessionStore(SessionStore):
            await self._client.touch(key, _exptime(max_age, self._clock))

        data = self._loads(payload)
-        return Session(session_id, data)
+        return Session.from_payload(session_id, data)

    async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None:
-        payload = self._dumps(dict(session))
+        # Index first: if the session write subsequently fails, a stale index
+        # entry is harmless and will be pruned, while a live unindexed session
+        # would be invisible to index-based invalidation.
+        for name, value in session.indexes:
+            await self._update_bucket(
+                self._index_key(name, value),
+                add={session_id},
+                max_age=max_age,
+            )
+        for name, value in session.removed_indexes:
+            await self._update_bucket(
+                self._index_key(name, value),
+                remove={session_id},
+            )
+
+        payload = self._dumps(session.to_payload())
        exptime = _exptime(max_age, self._clock) if max_age is not None else 0
        await self._client.set(self._key(session_id), payload, exptime=exptime)

    async def delete(self, session_id: str) -> None:
-        await self._client.delete(self._key(session_id))
+        key = self._key(session_id)
+        payload = await self._client.get(key)
+        indexes: set[IndexKey] = set()
+        if payload is not None:
+            try:
+                indexes = set(Session.from_payload(
+                    session_id,
+                    self._loads(payload),
+                ).indexes)
+            except Exception:
+                # A corrupt payload must not prevent session deletion.
+                pass
+
+        await self._client.delete(key)
+        for name, value in indexes:
+            await self._update_bucket(
+                self._index_key(name, value),
+                remove={session_id},
+            )
+
+    async def find_sessions(self, name: str, value: str) -> Sequence[str]:
+        index_key = self._index_key(name, value)
+        payload = await self._client.get(index_key)
+        if payload is None:
+            return []
+
+        members, _ = self._decode_bucket(payload)
+        live: list[str] = []
+        stale: list[str] = []
+        for session_id in members:
+            if await self._client.get(self._key(session_id)) is None:
+                stale.append(session_id)
+            else:
+                live.append(session_id)
+        if stale:
+            await self._update_bucket(index_key, remove=stale)
+        return live
@@ -0,0 +1,122 @@
+import pickle
+import unittest
+
+from pwo import async_test
+from test_store import FakeClock, StubMemcacheClient
+
+from kaya.session import Session
+from kaya.session.memcache import MemcacheSessionStore
+
+
+class MemcacheSessionIndexTest(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.clock = FakeClock()
+        self.client = StubMemcacheClient(self.clock)
+        self.store = MemcacheSessionStore(self.client, clock=self.clock)  # type: ignore[arg-type]
+
+    @async_test
+    async def test_save_load_and_find(self) -> None:
+        self.assertTrue(self.store.supports_indexes)
+        await self.store.save('abc', Session('abc', {'user': 'alice'}, indexes={('sub', 'alice')}))
+        loaded = await self.store.load('abc')
+        assert loaded is not None
+        self.assertEqual({'user': 'alice'}, dict(loaded))
+        self.assertEqual(frozenset({('sub', 'alice')}), loaded.indexes)
+        self.assertEqual(['abc'], await self.store.find_sessions('sub', 'alice'))
+
+    @async_test
+    async def test_rotation_preserves_index(self) -> None:
+        await self.store.save('old', Session('old', indexes={('sub', 'alice')}))
+        loaded = await self.store.load('old')
+        assert loaded is not None
+        loaded.set_id('new')
+        await self.store.delete('old')
+        await self.store.save('new', loaded)
+        self.assertEqual(['new'], await self.store.find_sessions('sub', 'alice'))
+
+    @async_test
+    async def test_delete_cleans_reverse_index(self) -> None:
+        await self.store.save('abc', Session('abc', indexes={('sub', 'alice')}))
+        await self.store.delete('abc')
+        self.assertEqual([], await self.store.find_sessions('sub', 'alice'))
+
+    @async_test
+    async def test_delete_by_index_deletes_all_sessions(self) -> None:
+        await self.store.save('a', Session('a', indexes={('sub', 'alice')}))
+        await self.store.save('b', Session('b', indexes={('sub', 'alice')}))
+        await self.store.save('c', Session('c', indexes={('sub', 'bob')}))
+        self.assertEqual(2, await self.store.delete_by_index('sub', 'alice'))
+        self.assertIsNone(await self.store.load('a'))
+        self.assertIsNone(await self.store.load('b'))
+        self.assertIsNotNone(await self.store.load('c'))
+
+    @async_test
+    async def test_stale_member_is_pruned(self) -> None:
+        await self.store.save('abc', Session('abc', indexes={('sub', 'alice')}))
+        await self.client.delete(b'kaya:session:abc')
+        self.assertEqual([], await self.store.find_sessions('sub', 'alice'))
+        index_key = self.store._index_key('sub', 'alice')
+        payload = await self.client.get(index_key)
+        assert payload is not None
+        members, _ = pickle.loads(payload)
+        self.assertEqual(set(), members)
+
+    @async_test
+    async def test_index_ttl_tracks_session_max_age(self) -> None:
+        await self.store.save(
+            'abc',
+            Session('abc', indexes={('sub', 'alice')}),
+            max_age=60,
+        )
+        index_key = self.store._index_key('sub', 'alice')
+        self.assertIn(index_key, self.client._expires)
+        self.clock.advance(61)
+        await self.store.save(
+            'abc',
+            Session('abc', indexes={('sub', 'alice')}),
+            max_age=60,
+        )
+        self.assertIsNotNone(await self.client.get(index_key))
+
+    @async_test
+    async def test_legacy_payload_is_upgraded_on_save(self) -> None:
+        await self.client.set(b'kaya:session:abc', pickle.dumps({'foo': 'bar'}))
+        loaded = await self.store.load('abc')
+        assert loaded is not None
+        self.assertEqual({'foo': 'bar'}, dict(loaded))
+        self.assertEqual(frozenset(), loaded.indexes)
+
+        loaded.add_index('sub', 'alice')
+        await self.store.save('abc', loaded)
+        raw = await self.client.get(b'kaya:session:abc')
+        assert raw is not None
+        payload = pickle.loads(raw)
+        self.assertEqual(1, payload['__kaya_session_v1__'])
+        self.assertEqual({'foo': 'bar'}, payload['data'])
+        self.assertEqual(['abc'], await self.store.find_sessions('sub', 'alice'))
+
+    @async_test
+    async def test_cas_conflict_is_retried(self) -> None:
+        await self.store.save('existing', Session('existing', indexes={('sub', 'alice')}))
+        original_cas = self.client.cas
+        attempts = 0
+
+        async def fail_once(key: bytes, value: bytes, cas_token: int, exptime: int = 0) -> bool:
+            nonlocal attempts
+            attempts += 1
+            if attempts == 1:
+                return False
+            return await original_cas(key, value, cas_token, exptime)
+
+        self.client.cas = fail_once  # type: ignore[method-assign]
+        await self.store.save('abc', Session('abc', indexes={('sub', 'alice')}))
+        self.assertEqual(
+            {'existing', 'abc'},
+            set(await self.store.find_sessions('sub', 'alice')),
+        )
+        self.assertGreaterEqual(attempts, 2)
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -28,6 +28,8 @@ class StubMemcacheClient:
        self._clock = clock
        self._data: dict[bytes, bytes] = {}
        self._expires: dict[bytes, float] = {}
+        self._cas: dict[bytes, int] = {}
+        self._next_cas = 1
        self.set_exptimes: list[int] = []
        self.touch_exptimes: list[int] = []

@@ -42,19 +44,46 @@ class StubMemcacheClient:
            return None
        return self._data.get(key)

-    async def set(self, key: bytes, value: bytes, exptime: int = 0) -> bool:
-        self.set_exptimes.append(exptime)
+    def _store(self, key: bytes, value: bytes, exptime: int) -> None:
        self._data[key] = value
+        self._cas[key] = self._next_cas
+        self._next_cas += 1
        if exptime > 0:
            self._expires[key] = self._clock() + exptime
        else:
            self._expires.pop(key, None)
+
+    async def set(self, key: bytes, value: bytes, exptime: int = 0) -> bool:
+        self.set_exptimes.append(exptime)
+        self._store(key, value, exptime)
+        return True
+
+    async def add(self, key: bytes, value: bytes, exptime: int = 0) -> bool:
+        if self._expired(key):
+            self._data.pop(key, None)
+            self._expires.pop(key, None)
+        if key in self._data:
+            return False
+        self._store(key, value, exptime)
+        return True
+
+    async def gets(self, key: bytes) -> tuple[Optional[bytes], Optional[int]]:
+        value = await self.get(key)
+        if value is None:
+            return None, None
+        return value, self._cas[key]
+
+    async def cas(self, key: bytes, value: bytes, cas_token: int, exptime: int = 0) -> bool:
+        if self._expired(key) or self._cas.get(key) != cas_token:
+            return False
+        self._store(key, value, exptime)
        return True

    async def delete(self, key: bytes) -> bool:
        existed = key in self._data
        self._data.pop(key, None)
        self._expires.pop(key, None)
+        self._cas.pop(key, None)
        return existed

    async def touch(self, key: bytes, exptime: int) -> bool:
@@ -169,7 +198,11 @@ class MemcacheSessionStoreTest(unittest.TestCase):
            clock=self.clock,
        )
        await store.save('abc', Session('abc', {'foo': 'bar'}))
-        self.assertEqual(b'{"foo": "bar"}', self.client._data[b'kaya:session:abc'])
+        self.assertEqual({
+            '__kaya_session_v1__': 1,
+            'data': {'foo': 'bar'},
+            'indexes': [],
+        }, json.loads(self.client._data[b'kaya:session:abc']))
        loaded = await store.load('abc')
        self.assertIsNotNone(loaded)
        assert loaded is not None
@@ -30,11 +30,21 @@ Sessions are stored under keys with the prefix `kaya:session:` (configurable
 via the `prefix` argument). Server-side expiry uses Redis key TTLs and slides
 on each access when the session mixin passes a `max_age`.

+## Session indexes
+
+`RedisSessionStore` supports `Session.add_index()`,
+`SessionStore.find_sessions()`, and `delete_by_index()`. Reverse memberships
+are Redis sets updated in the same transactional pipeline as session saves.
+They are removed on deletion, and stale members are pruned during lookups.
+Bucket keys use a SHA-256 digest of the index name and value.
+
 ## Serialization

 Session data is serialized with `pickle` by default, so arbitrary Python
-objects can be stored. A different serializer can be plugged in via the
-`dumps`/`loads` arguments:
+objects can be stored. The serialized value is an envelope containing the
+session dictionary and index metadata; legacy plain-dict values are still
+accepted. A different serializer can be plugged in via the `dumps`/`loads`
+arguments:

 ```python
 import json
@@ -26,11 +26,6 @@ dependencies = [
    "redis>=5.0",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine", "fakeredis"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -1,9 +1,10 @@
+import hashlib
 import pickle
-from typing import Any, Callable, Optional, cast
+from typing import Any, Callable, Optional, Sequence, cast

 from redis.asyncio import Redis

-from kaya.session import Session, SessionStore
+from kaya.session import IndexKey, Session, SessionStore


 class RedisSessionStore(SessionStore):
@@ -20,8 +21,14 @@ class RedisSessionStore(SessionStore):
    Python objects can be stored. Custom serializers can be plugged in via
    the ``dumps``/``loads`` arguments. Only connect this store to a trusted
    Redis server, as pickle deserialization of untrusted data is unsafe.
+
+    Reverse session indexes are stored as Redis sets. They are updated in the
+    same pipeline as session saves, cleaned on deletion, and pruned when stale
+    memberships are encountered during a lookup.
    """

+    supports_indexes = True
+
    def __init__(
        self,
        client: Redis,
@@ -37,6 +44,10 @@ class RedisSessionStore(SessionStore):
    def _key(self, session_id: str) -> str:
        return f"{self._prefix}{session_id}"

+    def _index_key(self, name: str, value: str) -> str:
+        digest = hashlib.sha256(f"{name}\0{value}".encode()).hexdigest()
+        return f"{self._prefix}idx:{digest}"
+
    async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]:
        key = self._key(session_id)
        payload = await self._client.get(key)
@@ -47,11 +58,65 @@ class RedisSessionStore(SessionStore):
            await self._client.expire(key, max_age)

        data = self._loads(cast(bytes, payload))
-        return Session(session_id, data)
+        return Session.from_payload(session_id, data)

    async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None:
-        payload = self._dumps(dict(session))
-        await self._client.set(self._key(session_id), payload, ex=max_age)
+        payload = self._dumps(session.to_payload())
+        pipeline = self._client.pipeline(transaction=True)
+        pipeline.set(self._key(session_id), payload, ex=max_age)
+        for name, value in session.indexes:
+            index_key = self._index_key(name, value)
+            pipeline.sadd(index_key, session_id)
+            if max_age is None:
+                pipeline.persist(index_key)
+            else:
+                pipeline.expire(index_key, max_age + 60)
+        for name, value in session.removed_indexes:
+            pipeline.srem(self._index_key(name, value), session_id)
+        await pipeline.execute()

    async def delete(self, session_id: str) -> None:
-        await self._client.delete(self._key(session_id))
+        key = self._key(session_id)
+        payload = await self._client.get(key)
+        indexes: set[IndexKey] = set()
+        if payload is not None:
+            try:
+                indexes = set(Session.from_payload(
+                    session_id,
+                    self._loads(cast(bytes, payload)),
+                ).indexes)
+            except Exception:
+                # A corrupt payload must not prevent session deletion.
+                pass
+
+        pipeline = self._client.pipeline(transaction=True)
+        pipeline.delete(key)
+        for name, value in indexes:
+            pipeline.srem(self._index_key(name, value), session_id)
+        await pipeline.execute()
+
+    async def find_sessions(self, name: str, value: str) -> Sequence[str]:
+        index_key = self._index_key(name, value)
+        members = await self._client.smembers(index_key)
+        session_ids = [
+            member.decode() if isinstance(member, bytes) else str(member)
+            for member in members
+        ]
+        if not session_ids:
+            return []
+
+        pipeline = self._client.pipeline(transaction=False)
+        for session_id in session_ids:
+            pipeline.exists(self._key(session_id))
+        exists = await pipeline.execute()
+
+        live: list[str] = []
+        stale: list[str] = []
+        for session_id, found in zip(session_ids, exists):
+            if found:
+                live.append(session_id)
+            else:
+                stale.append(session_id)
+        if stale:
+            await self._client.srem(index_key, *stale)
+        return live
@@ -0,0 +1,94 @@
+import pickle
+import unittest
+
+import fakeredis.aioredis
+from pwo import async_test
+
+from kaya.session import Session
+from kaya.session.redis import RedisSessionStore
+
+
+class RedisSessionIndexTest(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.client = fakeredis.aioredis.FakeRedis()
+        self.store = RedisSessionStore(self.client)
+
+    @async_test
+    async def test_save_load_and_find(self) -> None:
+        self.assertTrue(self.store.supports_indexes)
+        await self.store.save('abc', Session('abc', {'user': 'alice'}, indexes={('sub', 'alice')}))
+        loaded = await self.store.load('abc')
+        assert loaded is not None
+        self.assertEqual({'user': 'alice'}, dict(loaded))
+        self.assertEqual(frozenset({('sub', 'alice')}), loaded.indexes)
+        self.assertEqual(['abc'], await self.store.find_sessions('sub', 'alice'))
+
+    @async_test
+    async def test_rotation_preserves_index(self) -> None:
+        await self.store.save('old', Session('old', indexes={('sub', 'alice')}))
+        loaded = await self.store.load('old')
+        assert loaded is not None
+        loaded.set_id('new')
+        await self.store.delete('old')
+        await self.store.save('new', loaded)
+        self.assertEqual(['new'], await self.store.find_sessions('sub', 'alice'))
+
+    @async_test
+    async def test_delete_cleans_reverse_index(self) -> None:
+        await self.store.save('abc', Session('abc', indexes={('sub', 'alice')}))
+        index_key = self.store._index_key('sub', 'alice')
+        self.assertEqual({b'abc'}, await self.client.smembers(index_key))
+        await self.store.delete('abc')
+        self.assertEqual(set(), await self.client.smembers(index_key))
+
+    @async_test
+    async def test_delete_by_index_deletes_all_sessions(self) -> None:
+        await self.store.save('a', Session('a', indexes={('sub', 'alice')}))
+        await self.store.save('b', Session('b', indexes={('sub', 'alice')}))
+        await self.store.save('c', Session('c', indexes={('sub', 'bob')}))
+        self.assertEqual(2, await self.store.delete_by_index('sub', 'alice'))
+        self.assertIsNone(await self.store.load('a'))
+        self.assertIsNone(await self.store.load('b'))
+        self.assertIsNotNone(await self.store.load('c'))
+
+    @async_test
+    async def test_stale_member_is_pruned(self) -> None:
+        await self.store.save('abc', Session('abc', indexes={('sub', 'alice')}))
+        await self.client.delete('kaya:session:abc')
+        self.assertEqual([], await self.store.find_sessions('sub', 'alice'))
+        index_key = self.store._index_key('sub', 'alice')
+        self.assertEqual(set(), await self.client.smembers(index_key))
+
+    @async_test
+    async def test_index_ttl_tracks_session_max_age(self) -> None:
+        await self.store.save(
+            'abc',
+            Session('abc', indexes={('sub', 'alice')}),
+            max_age=60,
+        )
+        index_key = self.store._index_key('sub', 'alice')
+        ttl = await self.client.ttl(index_key)
+        self.assertGreater(ttl, 60)
+        self.assertLessEqual(ttl, 120)
+
+    @async_test
+    async def test_legacy_payload_is_upgraded_on_save(self) -> None:
+        await self.client.set('kaya:session:abc', pickle.dumps({'foo': 'bar'}))
+        loaded = await self.store.load('abc')
+        assert loaded is not None
+        self.assertEqual({'foo': 'bar'}, dict(loaded))
+        self.assertEqual(frozenset(), loaded.indexes)
+
+        loaded.add_index('sub', 'alice')
+        await self.store.save('abc', loaded)
+        raw = await self.client.get('kaya:session:abc')
+        assert raw is not None
+        payload = pickle.loads(raw)
+        self.assertEqual(1, payload['__kaya_session_v1__'])
+        self.assertEqual({'foo': 'bar'}, payload['data'])
+        self.assertEqual(['abc'], await self.store.find_sessions('sub', 'alice'))
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -89,7 +89,13 @@ class RedisSessionStoreTest(unittest.TestCase):
            loads=lambda b: json.loads(b.decode('utf-8')),
        )
        await store.save('abc', Session('abc', {'foo': 'bar'}))
-        self.assertEqual(b'{"foo": "bar"}', await self.client.get('kaya:session:abc'))
+        raw = await self.client.get('kaya:session:abc')
+        assert raw is not None
+        self.assertEqual({
+            '__kaya_session_v1__': 1,
+            'data': {'foo': 'bar'},
+            'indexes': [],
+        }, json.loads(raw))
        loaded = await store.load('abc')
        self.assertIsNotNone(loaded)
        assert loaded is not None
@@ -66,6 +66,37 @@ session and creates a fresh empty one.
 Set `max_age=None` to disable server-side expiry (and the `Max-Age` cookie
 attribute) entirely.

+## Session indexes
+
+Authentication layers can attach reverse-index metadata to a session without
+adding it to the user-visible session dictionary:
+
+```python
+ctx.session.add_index('oidc:default:sub', 'user-123')
+```
+
+Index-capable stores persist these memberships on `save()` and maintain them
+across session ID rotation and deletion:
+
+```python
+session_ids = await store.find_sessions('oidc:default:sub', 'user-123')
+deleted = await store.delete_by_index('oidc:default:sub', 'user-123')
+```
+
+Use `session.remove_index(name, value)` to remove a membership while keeping
+the session. `SessionMixin.store` exposes the backing store for features such
+as back-channel logout.
+
+`InMemorySessionStore`, `RedisSessionStore`, and `MemcacheSessionStore` set
+`supports_indexes = True`. Custom stores can opt in by setting that flag,
+implementing `find_sessions()`, maintaining reverse memberships on `save()`,
+and removing them on `delete()`. `delete_by_index()` has a generic default
+implementation based on `find_sessions()` and `delete()`.
+
+Remote stores serialize an envelope containing both the session dictionary and
+its index metadata. Existing plain-dict payloads remain readable and are
+upgraded on the next save.
+
 ## Features

 - `Session`: dict-like session object with modification tracking
@@ -74,6 +105,7 @@ attribute) entirely.
 - `SessionMixin`: composable Kaya mixin managing session cookies and persistence
 - Session ID regeneration (`session.regenerate_id()`) and invalidation
  (`session.invalidate()`) for authentication layers
+- Reverse session indexes for looking up and invalidating sessions by metadata
 - WebSocket support: the session is exposed as `ws.session` in websocket
  handlers, loaded at connect time and persisted on close

@@ -26,11 +26,6 @@ dependencies = [
    "pwo",
 ]

-[project.optional-dependencies]
-dev = [
-    "build", "mypy", "ipdb", "twine", "httpx", "httpx-ws"
-]
-
 [project.urls]
 "Homepage" = "https://github.com/woggioni/kaya"
 "Bug Tracker" = "https://github.com/woggioni/kaya/issues"
@@ -3,12 +3,13 @@ from pkgutil import extend_path
 __path__ = extend_path(__path__, __name__)

 from ._mixin import SessionMixin
-from ._session import Session
+from ._session import IndexKey, Session
 from ._store import InMemorySessionStore, SessionStore


 __all__ = [
    'InMemorySessionStore',
+    'IndexKey',
    'Session',
    'SessionMixin',
    'SessionStore',
@@ -183,6 +183,16 @@ class SessionMixin(KayaMixin):
        self._secure = secure
        self._samesite = samesite

+    @property
+    def store(self) -> SessionStore:
+        """Return the backing session store."""
+        return self._store
+
+    @property
+    def max_age(self) -> Optional[int]:
+        """Return the idle timeout used when persisting sessions."""
+        return self._max_age
+
    def apply(self, app: KayaApp) -> None:
        app.add_before_request_hook(self._before_request)
        app.add_after_request_hook(self._after_request)
@@ -273,3 +283,4 @@ class SessionMixin(KayaMixin):

        if session.id is not None:
            await self._store.save(session.id, session, self._max_age)
+            session.commit_indexes()
@@ -1,4 +1,10 @@
-from typing import Any, Iterator, Mapping, MutableMapping, Optional
+from typing import Any, Iterable, Iterator, Mapping, MutableMapping, Optional
+
+
+IndexKey = tuple[str, str]
+
+_PAYLOAD_MARKER = '__kaya_session_v1__'
+_PAYLOAD_VERSION = 1


 class Session(MutableMapping[str, Any]):
@@ -9,13 +15,21 @@ class Session(MutableMapping[str, Any]):
    it.
    """

-    def __init__(self, session_id: Optional[str] = None, data: Optional[Mapping[str, Any]] = None) -> None:
+    def __init__(
+        self,
+        session_id: Optional[str] = None,
+        data: Optional[Mapping[str, Any]] = None,
+        indexes: Iterable[IndexKey] = (),
+    ) -> None:
        self._id: Optional[str] = session_id
        self._old_id: Optional[str] = None
        self._data: dict[str, Any] = dict(data) if data else {}
        self._modified: bool = False
        self._regenerate: bool = False
        self._invalidated: bool = False
+        self._persisted_indexes: set[IndexKey] = set(indexes)
+        self._added_indexes: set[IndexKey] = set()
+        self._removed_indexes: set[IndexKey] = set()

    @property
    def id(self) -> Optional[str]:
@@ -60,6 +74,75 @@ class Session(MutableMapping[str, Any]):
    def mark_modified(self) -> None:
        self._modified = True

+    def add_index(self, name: str, value: str) -> None:
+        """Index this session under ``(name, value)`` when it is saved.
+
+        Index-capable stores can use this metadata to find and invalidate
+        sessions without knowing their IDs. Adding an already persisted index
+        is harmless and does not create a duplicate membership.
+        """
+        index = (name, value)
+        self._removed_indexes.discard(index)
+        if index not in self._persisted_indexes:
+            self._added_indexes.add(index)
+        self._modified = True
+
+    def remove_index(self, name: str, value: str) -> None:
+        """Remove an index membership when this session is saved."""
+        index = (name, value)
+        if index in self._added_indexes:
+            self._added_indexes.discard(index)
+        elif index in self._persisted_indexes:
+            self._removed_indexes.add(index)
+        self._modified = True
+
+    @property
+    def indexes(self) -> frozenset[IndexKey]:
+        """Return the index memberships that should be persisted."""
+        return frozenset(
+            (self._persisted_indexes - self._removed_indexes) | self._added_indexes
+        )
+
+    @property
+    def added_indexes(self) -> frozenset[IndexKey]:
+        """Return index memberships added since the last successful save."""
+        return frozenset(self._added_indexes)
+
+    @property
+    def removed_indexes(self) -> frozenset[IndexKey]:
+        """Return index memberships removed since the last successful save."""
+        return frozenset(self._removed_indexes)
+
+    def commit_indexes(self) -> None:
+        """Mark the current index memberships as persisted."""
+        self._persisted_indexes = set(self.indexes)
+        self._added_indexes.clear()
+        self._removed_indexes.clear()
+
+    def to_payload(self) -> dict[str, Any]:
+        """Return the serializable session payload used by remote stores."""
+        return {
+            _PAYLOAD_MARKER: _PAYLOAD_VERSION,
+            'data': dict(self),
+            'indexes': sorted(self.indexes),
+        }
+
+    @classmethod
+    def from_payload(cls, session_id: str, payload: Mapping[str, Any]) -> 'Session':
+        """Build a session from an envelope or a legacy plain-data payload."""
+        data = payload.get('data')
+        raw_indexes = payload.get('indexes')
+        if (payload.get(_PAYLOAD_MARKER) == _PAYLOAD_VERSION
+                and isinstance(data, Mapping)
+                and isinstance(raw_indexes, (list, tuple))):
+            indexes: set[IndexKey] = set()
+            for item in raw_indexes:
+                if (isinstance(item, (list, tuple)) and len(item) == 2
+                        and isinstance(item[0], str) and isinstance(item[1], str)):
+                    indexes.add((item[0], item[1]))
+            return cls(session_id, data, indexes=indexes)
+        return cls(session_id, payload)
+
    def __getitem__(self, key: str) -> Any:
        return self._data[key]

@@ -1,13 +1,21 @@
 from abc import ABC, abstractmethod
 from secrets import token_urlsafe
 from time import monotonic
-from typing import Any, Callable, Optional
+from typing import Any, Callable, ClassVar, Optional, Sequence

-from ._session import Session
+from ._session import IndexKey, Session


 class SessionStore(ABC):
-    """Pluggable backend for session persistence."""
+    """Pluggable backend for session persistence.
+
+    Stores with ``supports_indexes = True`` also maintain reverse indexes for
+    :class:`~kaya.session.Session` index metadata. They must add the session ID
+    to every effective index on ``save()``, apply pending removals, and remove
+    all of a session's index memberships on ``delete()``.
+    """
+
+    supports_indexes: ClassVar[bool] = False

    @abstractmethod
    async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]:
@@ -33,6 +41,23 @@ class SessionStore(ABC):
        """Remove the session from the store."""
        pass

+    async def find_sessions(self, name: str, value: str) -> Sequence[str]:
+        """Return live session IDs indexed under ``(name, value)``.
+
+        Only available on stores whose ``supports_indexes`` flag is true.
+        Implementations should prune stale index memberships encountered here.
+        """
+        raise NotImplementedError(
+            f'{type(self).__name__} does not support session indexes'
+        )
+
+    async def delete_by_index(self, name: str, value: str) -> int:
+        """Delete every live session indexed under ``(name, value)``."""
+        session_ids = await self.find_sessions(name, value)
+        for session_id in session_ids:
+            await self.delete(session_id)
+        return len(session_ids)
+
    def new_session_id(self) -> str:
        """Return a new opaque session identifier.

@@ -51,10 +76,26 @@ class InMemorySessionStore(SessionStore):
    inactivity. Active sessions slide the expiry window on each access.
    """

+    supports_indexes = True
+
    def __init__(self, clock: Callable[[], float] = monotonic) -> None:
        self._clock = clock
        self._data: dict[str, dict[str, Any]] = {}
        self._expires: dict[str, float] = {}
+        self._session_indexes: dict[str, set[IndexKey]] = {}
+        self._index_sessions: dict[IndexKey, set[str]] = {}
+
+    def _drop_session(self, session_id: str) -> None:
+        self._data.pop(session_id, None)
+        self._expires.pop(session_id, None)
+        indexes = self._session_indexes.pop(session_id, set())
+        for index in indexes:
+            session_ids = self._index_sessions.get(index)
+            if session_ids is None:
+                continue
+            session_ids.discard(session_id)
+            if not session_ids:
+                self._index_sessions.pop(index, None)

    async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]:
        data = self._data.get(session_id)
@@ -64,14 +105,17 @@ class InMemorySessionStore(SessionStore):
        expires = self._expires.get(session_id)
        now = self._clock()
        if expires is not None and now > expires:
-            self._data.pop(session_id, None)
-            self._expires.pop(session_id, None)
+            self._drop_session(session_id)
            return None

        if max_age is not None and expires is not None:
            self._expires[session_id] = now + max_age

-        return Session(session_id, data)
+        return Session(
+            session_id,
+            data,
+            indexes=self._session_indexes.get(session_id, ()),
+        )

    async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None:
        self._data[session_id] = dict(session)
@@ -80,6 +124,34 @@ class InMemorySessionStore(SessionStore):
        else:
            self._expires.pop(session_id, None)

+        for index in session.removed_indexes:
+            session_ids = self._index_sessions.get(index)
+            if session_ids is None:
+                continue
+            session_ids.discard(session_id)
+            if not session_ids:
+                self._index_sessions.pop(index, None)
+
+        indexes = set(session.indexes)
+        self._session_indexes[session_id] = indexes
+        for index in indexes:
+            self._index_sessions.setdefault(index, set()).add(session_id)
+
    async def delete(self, session_id: str) -> None:
-        self._data.pop(session_id, None)
-        self._expires.pop(session_id, None)
+        self._drop_session(session_id)
+
+    async def find_sessions(self, name: str, value: str) -> Sequence[str]:
+        session_ids = self._index_sessions.get((name, value))
+        if session_ids is None:
+            return []
+
+        now = self._clock()
+        live: list[str] = []
+        for session_id in list(session_ids):
+            expires = self._expires.get(session_id)
+            if (session_id not in self._data
+                    or (expires is not None and now > expires)):
+                self._drop_session(session_id)
+            else:
+                live.append(session_id)
+        return live
@@ -0,0 +1,169 @@
+import unittest
+
+import httpx
+from pwo import async_test
+
+from kaya.core import HttpContext, KayaApp
+from kaya.session import InMemorySessionStore, Session, SessionMixin
+
+
+class FakeClock:
+    def __init__(self, start: float = 0.0) -> None:
+        self._now = start
+
+    def __call__(self) -> float:
+        return self._now
+
+    def advance(self, seconds: float) -> None:
+        self._now += seconds
+
+
+class SessionIndexUnitTest(unittest.TestCase):
+
+    def test_add_and_remove_index(self) -> None:
+        session = Session()
+        session.add_index('sub', 'alice')
+        self.assertEqual(frozenset({('sub', 'alice')}), session.indexes)
+        self.assertEqual(frozenset({('sub', 'alice')}), session.added_indexes)
+        self.assertTrue(session.modified)
+
+        session.remove_index('sub', 'alice')
+        self.assertEqual(frozenset(), session.indexes)
+        self.assertEqual(frozenset(), session.added_indexes)
+        self.assertEqual(frozenset(), session.removed_indexes)
+
+    def test_remove_persisted_index(self) -> None:
+        session = Session(indexes={('sub', 'alice')})
+        self.assertFalse(session.modified)
+        session.remove_index('sub', 'alice')
+        self.assertEqual(frozenset(), session.indexes)
+        self.assertEqual(frozenset({('sub', 'alice')}), session.removed_indexes)
+
+        session.add_index('sub', 'alice')
+        self.assertEqual(frozenset({('sub', 'alice')}), session.indexes)
+        self.assertEqual(frozenset(), session.removed_indexes)
+
+    def test_payload_round_trip(self) -> None:
+        session = Session('abc', {'user': 'alice'}, indexes={('sub', 'alice')})
+        loaded = Session.from_payload('abc', session.to_payload())
+        self.assertEqual({'user': 'alice'}, dict(loaded))
+        self.assertEqual(frozenset({('sub', 'alice')}), loaded.indexes)
+
+    def test_legacy_payload(self) -> None:
+        loaded = Session.from_payload('abc', {'user': 'alice'})
+        self.assertEqual({'user': 'alice'}, dict(loaded))
+        self.assertEqual(frozenset(), loaded.indexes)
+
+
+class InMemorySessionIndexTest(unittest.TestCase):
+
+    @async_test
+    async def test_save_load_and_find(self) -> None:
+        store = InMemorySessionStore()
+        self.assertTrue(store.supports_indexes)
+        session = Session('abc', {'user': 'alice'})
+        session.add_index('sub', 'alice')
+        await store.save('abc', session)
+        session.commit_indexes()
+
+        loaded = await store.load('abc')
+        assert loaded is not None
+        self.assertEqual(frozenset({('sub', 'alice')}), loaded.indexes)
+        self.assertEqual(['abc'], await store.find_sessions('sub', 'alice'))
+        self.assertEqual([], await store.find_sessions('sub', 'bob'))
+
+    @async_test
+    async def test_delete_cleans_reverse_index(self) -> None:
+        store = InMemorySessionStore()
+        session = Session('abc', indexes={('sub', 'alice')})
+        await store.save('abc', session)
+        await store.delete('abc')
+        self.assertEqual([], await store.find_sessions('sub', 'alice'))
+        self.assertEqual({}, store._session_indexes)
+
+    @async_test
+    async def test_delete_by_index_deletes_all_sessions(self) -> None:
+        store = InMemorySessionStore()
+        await store.save('a', Session('a', indexes={('sub', 'alice')}))
+        await store.save('b', Session('b', indexes={('sub', 'alice')}))
+        await store.save('c', Session('c', indexes={('sub', 'bob')}))
+
+        self.assertEqual(2, await store.delete_by_index('sub', 'alice'))
+        self.assertIsNone(await store.load('a'))
+        self.assertIsNone(await store.load('b'))
+        self.assertIsNotNone(await store.load('c'))
+
+    @async_test
+    async def test_expiry_prunes_index(self) -> None:
+        clock = FakeClock()
+        store = InMemorySessionStore(clock=clock)
+        await store.save(
+            'abc',
+            Session('abc', indexes={('sub', 'alice')}),
+            max_age=60,
+        )
+        clock.advance(61)
+        self.assertEqual([], await store.find_sessions('sub', 'alice'))
+        self.assertEqual({}, store._session_indexes)
+
+    @async_test
+    async def test_removed_index_is_not_readded(self) -> None:
+        store = InMemorySessionStore()
+        session = Session('abc', indexes={('sub', 'alice')})
+        await store.save('abc', session)
+        session.commit_indexes()
+
+        session.remove_index('sub', 'alice')
+        await store.save('abc', session)
+        self.assertEqual([], await store.find_sessions('sub', 'alice'))
+        self.assertEqual(frozenset(), session.indexes)
+
+
+class SessionIndexIntegrationTest(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.store = InMemorySessionStore()
+        self.mixin = SessionMixin(self.store)
+        self.app = KayaApp(mixins=[self.mixin])
+
+        @self.app.GET('/login')
+        async def login(ctx: HttpContext) -> None:
+            ctx.session['user'] = 'alice'
+            ctx.session.add_index('sub', 'alice')
+            await ctx.send_str(200, 'ok')
+
+        @self.app.GET('/rotate')
+        async def rotate(ctx: HttpContext) -> None:
+            ctx.session.regenerate_id()
+            await ctx.send_str(200, 'ok')
+
+        @self.app.GET('/logout')
+        async def logout(ctx: HttpContext) -> None:
+            ctx.session.invalidate()
+            await ctx.send_str(200, 'ok')
+
+    @async_test
+    async def test_rotation_preserves_index(self) -> None:
+        self.assertIs(self.store, self.mixin.store)
+        transport = httpx.ASGITransport(app=self.app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            await client.get('/login')
+            old_id = client.cookies['session_id']
+            self.assertEqual([old_id], await self.store.find_sessions('sub', 'alice'))
+
+            await client.get('/rotate')
+            new_id = client.cookies['session_id']
+            self.assertNotEqual(old_id, new_id)
+            self.assertEqual([new_id], await self.store.find_sessions('sub', 'alice'))
+
+    @async_test
+    async def test_invalidation_cleans_index(self) -> None:
+        transport = httpx.ASGITransport(app=self.app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            await client.get('/login')
+            await client.get('/logout')
+            self.assertEqual([], await self.store.find_sessions('sub', 'alice'))
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -0,0 +1,51 @@
+[project]
+name = "kaya"
+version = "0.0.0"
+description = "Development workspace for the Kaya lightweight ASGI web framework"
+requires-python = ">=3.10"
+dependencies = [
+    "kaya-core",
+    "kaya-rsgi",
+    "kaya-session",
+    "kaya-session-redis",
+    "kaya-session-memcache",
+    "kaya-oidc",
+    "kaya-openapi",
+    "kaya-cors",
+    "kaya-forwarded",
+    "kaya-otel",
+]
+
+[dependency-groups]
+dev = [
+    "fakeredis",
+    "mypy",
+    "ipdb",
+    "httpx",
+    "httpx-ws",
+]
+
+[tool.uv]
+package = false
+required-version = ">=0.13"
+
+[tool.uv.workspace]
+members = ["packages/*"]
+
+[tool.uv.sources]
+kaya-core = { workspace = true }
+kaya-rsgi = { workspace = true }
+kaya-session = { workspace = true }
+kaya-session-redis = { workspace = true }
+kaya-session-memcache = { workspace = true }
+kaya-oidc = { workspace = true }
+kaya-openapi = { workspace = true }
+kaya-cors = { workspace = true }
+kaya-forwarded = { workspace = true }
+kaya-otel = { workspace = true }
+pwo = { index = "gitea" }
+
+[[tool.uv.index]]
+name = "gitea"
+url = "https://gitea.woggioni.net/api/packages/woggioni/pypi/simple"
+explicit = true
@@ -1,17 +0,0 @@
-kaya-core @ file:./packages/kaya-core
-kaya-rsgi @ file:./packages/kaya-rsgi
-kaya-session @ file:./packages/kaya-session
-kaya-session-redis @ file:./packages/kaya-session-redis
-kaya-session-memcache @ file:./packages/kaya-session-memcache
-kaya-oidc @ file:./packages/kaya-oidc
-kaya-openapi @ file:./packages/kaya-openapi
-kaya-cors @ file:./packages/kaya-cors
-kaya-forwarded @ file:./packages/kaya-forwarded
-kaya-otel @ file:./packages/kaya-otel
-build
-fakeredis
-mypy
-ipdb
-twine
-httpx
-httpx-ws
@@ -1,247 +0,0 @@
-#
-# This file is autogenerated by pip-compile with Python 3.14
-# by the following command:
-#
-#    pip-compile --allow-unsafe --extra-index-url=https://pypi.org/simple --index-url=https://gitea.woggioni.net/api/packages/woggioni/pypi/simple --output-file=requirements-dev.txt requirements-dev.in
-#
--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
--extra-index-url https://pypi.org/simple
-
-aiomcache==0.8.2
-    # via kaya-session-memcache
-anyio==4.14.2
-    # via
-    #   httpx
-    #   httpx-ws
-ast-serialize==0.6.0
-    # via mypy
-asttokens==3.0.2
-    # via stack-data
-build==1.5.0
-    # via -r requirements-dev.in
-certifi==2026.6.17
-    # via
-    #   httpcore
-    #   httpx
-    #   requests
-cffi==2.1.0
-    # via cryptography
-charset-normalizer==3.4.9
-    # via requests
-click==8.4.2
-    # via granian
-cryptography==49.0.0
-    # via
-    #   pyjwt
-    #   secretstorage
-decorator==5.3.1
-    # via
-    #   ipdb
-    #   ipython
-docutils==0.23
-    # via readme-renderer
-executing==2.2.1
-    # via stack-data
-fakeredis==2.36.2
-    # via -r requirements-dev.in
-googleapis-common-protos==1.75.3
-    # via opentelemetry-exporter-otlp-proto-http
-granian==2.7.9
-    # via kaya-rsgi
-h11==0.16.0
-    # via
-    #   httpcore
-    #   wsproto
-httpcore==1.0.9
-    # via
-    #   httpx
-    #   httpx-ws
-httpx==0.28.1
-    # via
-    #   -r requirements-dev.in
-    #   httpx-ws
-    #   kaya-oidc
-httpx-ws==0.9.0
-    # via -r requirements-dev.in
-id==1.6.1
-    # via twine
-idna==3.18
-    # via
-    #   anyio
-    #   httpx
-    #   requests
-ipdb==0.13.13
-    # via -r requirements-dev.in
-ipython==9.15.0
-    # via ipdb
-ipython-pygments-lexers==1.1.1
-    # via ipython
-jaraco-classes==3.4.0
-    # via keyring
-jaraco-context==6.1.2
-    # via keyring
-jaraco-functools==4.6.0
-    # via keyring
-jedi==0.20.0
-    # via ipython
-jeepney==0.9.0
-    # via
-    #   keyring
-    #   secretstorage
-file:./packages/kaya-core
-    # via
-    #   -r requirements-dev.in
-    #   kaya-cors
-    #   kaya-forwarded
-    #   kaya-oidc
-    #   kaya-openapi
-    #   kaya-otel
-    #   kaya-rsgi
-    #   kaya-session
-file:./packages/kaya-cors
-    # via -r requirements-dev.in
-file:./packages/kaya-forwarded
-    # via -r requirements-dev.in
-file:./packages/kaya-oidc
-    # via -r requirements-dev.in
-file:./packages/kaya-openapi
-    # via -r requirements-dev.in
-file:./packages/kaya-otel
-    # via -r requirements-dev.in
-file:./packages/kaya-rsgi
-    # via -r requirements-dev.in
-file:./packages/kaya-session
-    # via
-    #   -r requirements-dev.in
-    #   kaya-oidc
-    #   kaya-session-memcache
-    #   kaya-session-redis
-file:./packages/kaya-session-memcache
-    # via -r requirements-dev.in
-file:./packages/kaya-session-redis
-    # via -r requirements-dev.in
-keyring==25.7.0
-    # via twine
-librt==0.13.0
-    # via mypy
-markdown-it-py==4.2.0
-    # via rich
-matplotlib-inline==0.2.2
-    # via ipython
-mdurl==0.1.2
-    # via markdown-it-py
-more-itertools==11.1.0
-    # via
-    #   jaraco-classes
-    #   jaraco-functools
-mypy==2.3.0
-    # via -r requirements-dev.in
-mypy-extensions==1.1.0
-    # via mypy
-nh3==0.3.6
-    # via readme-renderer
-opentelemetry-api==1.44.0
-    # via
-    #   opentelemetry-exporter-otlp-proto-http
-    #   opentelemetry-sdk
-    #   opentelemetry-semantic-conventions
-opentelemetry-exporter-otlp-proto-common==1.44.0
-    # via opentelemetry-exporter-otlp-proto-http
-opentelemetry-exporter-otlp-proto-http==1.44.0
-    # via kaya-otel
-opentelemetry-proto==1.44.0
-    # via
-    #   opentelemetry-exporter-otlp-proto-common
-    #   opentelemetry-exporter-otlp-proto-http
-opentelemetry-sdk==1.44.0
-    # via
-    #   kaya-otel
-    #   opentelemetry-exporter-otlp-proto-http
-opentelemetry-semantic-conventions==0.65b0
-    # via opentelemetry-sdk
-packaging==26.2
-    # via
-    #   build
-    #   twine
-parso==0.8.7
-    # via jedi
-pathspec==1.1.1
-    # via mypy
-pexpect==4.9.0
-    # via ipython
-prompt-toolkit==3.0.52
-    # via ipython
-protobuf==7.36.2
-    # via
-    #   googleapis-common-protos
-    #   opentelemetry-proto
-psutil==7.2.2
-    # via ipython
-ptyprocess==0.7.0
-    # via pexpect
-pure-eval==0.2.3
-    # via stack-data
-pwo==0.1.2
-    # via
-    #   kaya-core
-    #   kaya-rsgi
-    #   kaya-session
-pycparser==3.0
-    # via cffi
-pygments==2.20.0
-    # via
-    #   ipython
-    #   ipython-pygments-lexers
-    #   readme-renderer
-    #   rich
-pyjwt[crypto]==2.13.0
-    # via kaya-oidc
-pyproject-hooks==1.2.0
-    # via build
-readme-renderer==45.0
-    # via twine
-redis==8.0.1
-    # via
-    #   fakeredis
-    #   kaya-session-redis
-requests==2.34.2
-    # via
-    #   opentelemetry-exporter-otlp-proto-http
-    #   requests-toolbelt
-    #   twine
-requests-toolbelt==1.0.0
-    # via twine
-rfc3986==2.0.0
-    # via twine
-rich==15.0.0
-    # via twine
-secretstorage==3.5.0
-    # via keyring
-sortedcontainers==2.4.0
-    # via fakeredis
-stack-data==0.6.3
-    # via ipython
-traitlets==5.15.1
-    # via
-    #   ipython
-    #   matplotlib-inline
-twine==6.2.0
-    # via -r requirements-dev.in
-typing-extensions==4.16.0
-    # via
-    #   kaya-core
-    #   mypy
-    #   opentelemetry-api
-    #   opentelemetry-exporter-otlp-proto-http
-    #   opentelemetry-sdk
-    #   opentelemetry-semantic-conventions
-    #   pwo
-urllib3==2.7.0
-    # via
-    #   id
-    #   requests
-    #   twine
-wcwidth==0.8.2
-    # via prompt-toolkit
-wsproto==1.3.2
-    # via httpx-ws