Add OIDC back-channel logout with session indexes

- 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

.gitea/workflows/build.yaml

@@ -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

.python-version

@@ -0,0 +1 @@
+3.14.6

README.md

@@ -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
 ```

packages/kaya-core/pyproject.toml

@@ -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"

packages/kaya-cors/pyproject.toml

@@ -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"

packages/kaya-forwarded/pyproject.toml

@@ -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"

packages/kaya-oidc/README.md

@@ -40,20 +40,219 @@ async def profile(ctx: HttpContext):
 `OIDCMixin` depends on `SessionMixin`; passing only `oidc` to `KayaApp(mixins=...)`
 also works because the app applies mixin dependencies automatically.
 
+## Multiple providers
+
+Use `OIDCProviders` to run several OIDC providers in one application. Each
+config's `provider` becomes a route segment under `base_path`
+(`/auth/{provider}/login`, `/auth/{provider}/callback` and
+`/auth/{provider}/logout`), and all providers share a single `SessionMixin`:
+
+```python
+from kaya.oidc import OIDCConfig, OIDCProviders
+
+session = SessionMixin(InMemorySessionStore())
+oidc = OIDCProviders(
+    [
+        OIDCConfig(
+            issuer='https://accounts.google.com',
+            client_id=os.environ['GOOGLE_CLIENT_ID'],
+            client_secret=os.environ['GOOGLE_CLIENT_SECRET'],
+            redirect_uri='http://localhost:8000/auth/google/callback',
+            provider='google',
+        ),
+        OIDCConfig(
+            issuer='https://github.com/login/oauth',  # or any OIDC issuer
+            client_id=os.environ['GITHUB_CLIENT_ID'],
+            client_secret=os.environ['GITHUB_CLIENT_SECRET'],
+            redirect_uri='http://localhost:8000/auth/github/callback',
+            provider='github',
+        ),
+    ],
+    session=session,
+    base_path='/auth',  # optional
+)
+app = KayaApp(mixins=[session, oidc])
+
+@app.GET('/google-profile')
+@oidc.require_auth('google')
+async def google_profile(ctx: HttpContext):
+    user = oidc.get_user(ctx, 'google')
+    await ctx.send_str(200, f"Hello {user.email or user.sub}")
+
+@app.GET('/who')
+async def who(ctx: HttpContext):
+    users = oidc.get_users(ctx)                     # {provider: OIDCUser}
+    providers = oidc.logged_in_providers(ctx)       # ['google', ...]
+    await ctx.send_str(200, ', '.join(users))
+```
+
+Each provider's session state is namespaced as `oidc:{provider}:...`. Make sure
+each config's `redirect_uri` points at that provider's derived callback path.
+
+### Identity strategy
+
+`strategy` controls how many identities may coexist in the shared session:
+
+- `'single-identity'` (default): completing a login evicts every other
+  provider's identity and tokens, so the session holds exactly one user — last
+  login wins, like Spring Security. After a GitHub login, `get_user(ctx,
+  'google')` returns `None` and `require_auth('google')` redirects to the Google
+  login.
+- `'multi-identity'`: each provider stays logged in independently. Use it for
+  linked accounts, aggregating several provider APIs, or step-up flows. Logging
+  out of one provider keeps the shared session alive while another provider is
+  still logged in.
+
+```python
+oidc = OIDCProviders([google_config, github_config], session=session,
+                     strategy='multi-identity')
+```
+
+Ephemeral flow keys (`state`, `nonce`, `code_verifier`) are never evicted, so
+logins already in flight can still complete; under `'single-identity'` the last
+callback to finish wins. Under either strategy, `get_users(ctx)` only reports
+providers actually present in the session.
+
+The composable alternative is to create one `OIDCMixin` per provider, pointing
+each at distinct `login_path`/`callback_path`/`logout_path` values and sharing
+the same `SessionMixin`. The same `strategy` keyword is accepted and defaults to
+`'single-identity'`:
+
+```python
+google = OIDCMixin(google_config, session=session)
+github = OIDCMixin(github_config, session=session, strategy='multi-identity')
+app = KayaApp(mixins=[session, google, github])
+```
+
+> **Upgrading:** session keys are now prefixed with `oidc:{provider}:` (the
+> 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:
+
+- `'never'` (default): the authorization request travels as query parameters,
+  as before.
+- `'auto'`: push the request whenever the provider advertises a
+  `pushed_authorization_request_endpoint`; otherwise fall back to the plain
+  redirect.
+- `'always'`: always push; login raises a `RuntimeError` if the provider does
+  not support PAR.
+
+```python
+OIDCConfig(..., par='auto')
+```
+
+A provider that advertises `require_pushed_authorization_requests: true` must be
+configured with `par='auto'` or `par='always'`. With `par='never'` the login
+fails fast with a clear error instead of issuing a redirect the provider would
+reject.
+
+The same parameters as the plain authorization request (PKCE challenge,
+`state`, `nonce`, scopes, ...) are POSTed directly to the provider, and the
+browser is redirected with only `client_id` and the single-use `request_uri`.
+Request parameters therefore no longer travel through the user agent. PKCE,
+`state`, and `nonce` are still used and protect against request URI swapping.
+
+Client authentication follows the token endpoint: HTTP Basic when
+`client_secret` is set, or none for public clients. PAR with other client
+authentication methods (`private_key_jwt`, mTLS) is not supported.
+
 ## Features
 
 - Generic OIDC discovery
 - Authorization Code Flow with PKCE (S256)
 - ID token signature validation with JWKS
+- Issuer validation against discovery metadata, including support for
+  Microsoft Entra ID `{tenantid}` issuer templates
+- Refreshed ID token validation on token refresh
 - `state` and `nonce` protection
 - Session fixation defense via `regenerate_id()` after login
 - 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
 - Composable with any other `KayaMixin` (RSGI, MCP, etc.)
 
+## Microsoft Entra ID
+
+Entra ID v2 ID tokens carry an `iss` claim of the form
+`https://login.microsoftonline.com/<tenant-id>/v2.0`, while the discovery
+metadata for the multi-tenant endpoints advertises the templated issuer
+`https://login.microsoftonline.com/{tenantid}/v2.0`. `kaya-oidc` validates the
+`iss` claim against the discovered metadata and substitutes the token's `tid`
+claim into `{tenantid}`, so both tenant-specific and multi-tenant issuers work:
+
+```python
+OIDCConfig(
+    issuer='https://login.microsoftonline.com/<tenant-id-or-domain>/v2.0',
+    client_id=os.environ['OIDC_CLIENT_ID'],
+    client_secret=os.environ['OIDC_CLIENT_SECRET'],
+    redirect_uri='https://example.com/auth/callback',
+)
+```
+
+Use a tenant-specific issuer (the tenant GUID or a verified domain name) to
+restrict logins to that tenant. The `common` and `organizations` aliases are
+multi-tenant: any tenant in that cloud can obtain an ID token that passes
+validation. If you build a multi-tenant application, check the `tid` claim
+against the tenants you trust.
+
 ## Security notes
 
+- The `iss` claim is validated against the `issuer` value of the provider's
+  discovery document, not against `OIDCConfig.issuer` (they can differ, for
+  example for Entra ID multi-tenant endpoints).
+- Refreshed ID tokens are validated before being stored; the `nonce` is also
+  checked when the refreshed token carries one.
 - The `none` signing algorithm is rejected by default.
 - Only algorithms listed in `OIDCConfig.allowed_id_token_algorithms` are accepted.
 - Always use HTTPS for `redirect_uri` in production.

packages/kaya-oidc/pyproject.toml

@@ -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"

packages/kaya-oidc/src/kaya/oidc/__init__.py

@@ -1,11 +1,15 @@
 from ._client import OIDCClient
-from ._config import OIDCConfig
-from ._mixin import OIDCMixin, OIDCUser
+from ._config import OIDCConfig, PARPolicy
+from ._mixin import IdentityStrategy, OIDCMixin, OIDCUser
+from ._providers import OIDCProviders
 
 
 __all__ = [
+    'IdentityStrategy',
     'OIDCClient',
     'OIDCConfig',
     'OIDCMixin',
+    'OIDCProviders',
     'OIDCUser',
+    'PARPolicy',
 ]

packages/kaya-oidc/src/kaya/oidc/_client.py

@@ -1,3 +1,4 @@
+import re
 from typing import Any, Mapping, Optional, Sequence, cast
 from urllib.parse import urlencode
 
@@ -5,9 +6,13 @@ import httpx
 import jwt
 from jwt import PyJWK
 
-from ._config import OIDCConfig
+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:
     """Low-level OIDC client implementing discovery, PKCE, and token validation."""
@@ -42,8 +47,14 @@ class OIDCClient:
         async with httpx.AsyncClient() as client:
             return await client.request(method, url, **kwargs)
 
-    async def build_authorization_url(self) -> tuple[str, str, str, str]:
-        """Return (authorization_url, state, nonce, code_verifier)."""
+    async def build_authorization_url(self, par: PARPolicy = 'never') -> tuple[str, str, str, str]:
+        """Return (authorization_url, state, nonce, code_verifier).
+
+        When ``par`` selects Pushed Authorization Requests (RFC 9126), the
+        authorization request parameters are POSTed to the provider's PAR
+        endpoint first and the returned URL carries only ``client_id`` and
+        ``request_uri``.
+        """
         metadata = await self._ensure_metadata()
         authorization_endpoint = metadata.get('authorization_endpoint')
         if not isinstance(authorization_endpoint, str):
@@ -54,7 +65,6 @@ class OIDCClient:
         code_verifier, code_challenge = generate_pkce()
 
         params = {
-            'client_id': self._config.client_id,
             'response_type': 'code',
             'scope': ' '.join(self._config.scopes),
             'redirect_uri': self._config.redirect_uri,
@@ -63,9 +73,67 @@ class OIDCClient:
             'code_challenge': code_challenge,
             'code_challenge_method': 'S256',
         }
-        url = authorization_endpoint + '?' + urlencode(params)
+
+        if self._use_par(metadata, par):
+            request_uri = await self._push_authorization_request(metadata, params)
+            redirect_params = {'client_id': self._config.client_id, 'request_uri': request_uri}
+        else:
+            redirect_params = {'client_id': self._config.client_id, **params}
+
+        url = authorization_endpoint + '?' + urlencode(redirect_params)
         return url, state, nonce, code_verifier
 
+    def _use_par(self, metadata: Mapping[str, Any], policy: PARPolicy) -> bool:
+        endpoint = metadata.get('pushed_authorization_request_endpoint')
+        required = metadata.get('require_pushed_authorization_requests') is True
+
+        if required:
+            if not isinstance(endpoint, str):
+                raise RuntimeError(
+                    'OIDC provider requires pushed authorization requests but does not '
+                    'advertise a pushed_authorization_request_endpoint'
+                )
+            if policy == 'never':
+                raise RuntimeError(
+                    "OIDC provider requires pushed authorization requests; set par to "
+                    "'auto' or 'always'"
+                )
+            return True
+
+        if policy == 'always':
+            if not isinstance(endpoint, str):
+                raise RuntimeError(
+                    'OIDC provider does not advertise a pushed_authorization_request_endpoint'
+                )
+            return True
+
+        return policy == 'auto' and isinstance(endpoint, str)
+
+    async def _push_authorization_request(
+        self, metadata: Mapping[str, Any], params: Mapping[str, str]
+    ) -> str:
+        endpoint = metadata.get('pushed_authorization_request_endpoint')
+        if not isinstance(endpoint, str):
+            raise RuntimeError(
+                'OIDC provider does not advertise a pushed_authorization_request_endpoint'
+            )
+
+        # RFC 9126 section 2.1: client_id is required in the body even when the
+        # client authenticates with the Authorization header.
+        data = {'client_id': self._config.client_id, **params}
+        resp = await self._request('POST', endpoint, data=data, auth=self._client_auth())
+        resp.raise_for_status()
+        body = resp.json()
+        request_uri = body.get('request_uri') if isinstance(body, Mapping) else None
+        if not isinstance(request_uri, str):
+            raise RuntimeError('PAR endpoint response is missing request_uri')
+        return request_uri
+
+    def _client_auth(self) -> Optional[tuple[str, str]]:
+        if self._config.client_secret is None:
+            return None
+        return (self._config.client_id, self._config.client_secret)
+
     async def fetch_token(self, code: str, code_verifier: str) -> Mapping[str, Any]:
         metadata = await self._ensure_metadata()
         token_endpoint = metadata.get('token_endpoint')
@@ -79,11 +147,8 @@ class OIDCClient:
             'client_id': self._config.client_id,
             'code_verifier': code_verifier,
         }
-        auth: Optional[tuple[str, str]] = None
-        if self._config.client_secret is not None:
-            auth = (self._config.client_id, self._config.client_secret)
 
-        resp = await self._request('POST', token_endpoint, data=data, auth=auth)
+        resp = await self._request('POST', token_endpoint, data=data, auth=self._client_auth())
         resp.raise_for_status()
         return cast(Mapping[str, Any], resp.json())
 
@@ -98,15 +163,89 @@ class OIDCClient:
             'refresh_token': refresh_token,
             'client_id': self._config.client_id,
         }
-        auth: Optional[tuple[str, str]] = None
-        if self._config.client_secret is not None:
-            auth = (self._config.client_id, self._config.client_secret)
 
-        resp = await self._request('POST', token_endpoint, data=data, auth=auth)
+        resp = await self._request('POST', token_endpoint, data=data, auth=self._client_auth())
         resp.raise_for_status()
         return cast(Mapping[str, Any], resp.json())
 
     async def validate_id_token(self, id_token: str, nonce: str) -> Mapping[str, Any]:
+        payload = await self._decode_id_token(id_token)
+        if payload.get('nonce') != nonce:
+            raise ValueError('ID token nonce mismatch')
+        return payload
+
+    async def validate_refreshed_id_token(
+        self,
+        id_token: str,
+        expected_nonce: Optional[str] = None,
+    ) -> Mapping[str, Any]:
+        """Validate an ID token returned by the refresh token grant.
+
+        Unlike the authorization code flow, the provider may omit the ``nonce``
+        claim; when it is present and the original nonce is known, the two must
+        match (OIDC Core 12.2).
+        """
+        payload = await self._decode_id_token(id_token)
+        token_nonce = payload.get('nonce')
+        if token_nonce is not None and expected_nonce is not None and token_nonce != expected_nonce:
+            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')
         if not isinstance(alg, str):
@@ -115,29 +254,59 @@ class OIDCClient:
         if alg == 'none':
             if not self._config.allow_unsigned_id_tokens:
                 raise ValueError('Unsigned ID tokens are not allowed')
+            metadata_issuer = await self._get_metadata_issuer()
             payload = jwt.decode(
                 id_token,
-                options={'verify_signature': False},
+                audience=self._config.client_id,
+                options={
+                    'verify_signature': False,
+                    'verify_aud': True,
+                    'verify_exp': True,
+                    'verify_iat': True,
+                    'verify_nbf': True,
+                },
             )
         else:
             if alg not in self._config.allowed_id_token_algorithms:
                 raise ValueError(f'ID 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(
                 id_token,
                 signing_key.key,
                 algorithms=[alg],
-                issuer=self._config.issuer,
                 audience=self._config.client_id,
             )
 
         if not isinstance(payload, dict):
             raise ValueError('ID token payload is not a JSON object')
-        if payload.get('nonce') != nonce:
-            raise ValueError('ID token nonce mismatch')
+        self._validate_issuer(payload, metadata_issuer)
         return payload
 
+    async def _get_metadata_issuer(self) -> str:
+        metadata = await self._ensure_metadata()
+        issuer = metadata.get('issuer')
+        if not isinstance(issuer, str):
+            raise RuntimeError('OIDC provider does not advertise an issuer')
+        return issuer
+
+    @staticmethod
+    def _validate_issuer(payload: Mapping[str, Any], metadata_issuer: str) -> None:
+        iss = payload.get('iss')
+        if not isinstance(iss, str):
+            raise jwt.MissingRequiredClaimError('iss')
+
+        expected = metadata_issuer
+        if _TENANT_ID_PLACEHOLDER.search(expected):
+            tenant_id = payload.get('tid')
+            if not isinstance(tenant_id, str):
+                raise jwt.InvalidIssuerError('Invalid issuer')
+            expected = _TENANT_ID_PLACEHOLDER.sub(tenant_id, expected, count=1)
+
+        if iss != expected:
+            raise jwt.InvalidIssuerError('Invalid issuer')
+
     def _get_signing_key(self, jwks: Mapping[str, Any], kid: Optional[str], alg: str) -> PyJWK:
         keys = jwks.get('keys')
         if not isinstance(keys, Sequence):

packages/kaya-oidc/src/kaya/oidc/_config.py

@@ -1,13 +1,37 @@
+import re
 from dataclasses import dataclass, field
-from typing import Any, Optional, Sequence
+from typing import Any, Literal, Optional, Sequence
 
 import httpx
 
+_PROVIDER_PATTERN = re.compile(r'^[A-Za-z0-9_-]+$')
+
+type PARPolicy = Literal['never', 'auto', 'always']
+
+_PAR_POLICIES: tuple[PARPolicy, ...] = ('never', 'auto', 'always')
+
 
 @dataclass
 class OIDCConfig:
     """Configuration for an OpenID Connect provider.
 
+    ``provider`` is a stable identifier for this provider. It namespaces the
+    session state (``oidc:{provider}:...``) so several providers can be used in
+    the same application without interfering with each other.
+
+    ``par`` selects when to use Pushed Authorization Requests (RFC 9126):
+    ``'never'`` uses a plain query-string authorization request, ``'auto'``
+    pushes the request whenever the provider advertises a
+    ``pushed_authorization_request_endpoint``, and ``'always'`` requires PAR.
+    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.
@@ -35,8 +59,28 @@ class OIDCConfig:
         'EdDSA',
     ))
     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:
+            raise ValueError(
+                'provider must be a non-empty string of letters, digits, '
+                "underscores or hyphens (got "
+                f'{self.provider!r})'
+            )
+        if self.par not in _PAR_POLICIES:
+            raise ValueError(
+                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)

packages/kaya-oidc/src/kaya/oidc/_mixin.py

@@ -1,5 +1,6 @@
-from typing import Any, Awaitable, Callable, Mapping, Optional, Sequence
+from typing import Any, Awaitable, Callable, Literal, Mapping, Optional, Sequence, cast
 
+import jwt
 from kaya.core import HttpContext, KayaApp, KayaMixin
 from kaya.session import Session, SessionMixin
 from urllib.parse import parse_qs
@@ -10,6 +11,14 @@ from ._config import OIDCConfig
 
 type HttpHandler = Callable[..., Awaitable[None]]
 
+type IdentityStrategy = Literal['single-identity', 'multi-identity']
+
+_SESSION_NAMESPACE = 'oidc'
+_USER_SUFFIX = 'user'
+_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]):
     """Read-only view of the OIDC userinfo / ID token claims."""
@@ -28,7 +37,7 @@ class OIDCUser(Mapping[str, Any]):
 
     @property
     def sub(self) -> str:
-        return self._data['sub']
+        return cast(str, self._data['sub'])
 
     @property
     def email(self) -> Optional[str]:
@@ -56,6 +65,21 @@ 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
+    :class:`~kaya.oidc.OIDCProviders` to wire several at once.
+
+    ``strategy`` controls how many identities may coexist in the session. The
+    default ``'single-identity'`` evicts every other provider's identity after a
+    successful login, so the session has exactly one user (last login wins).
+    ``'multi-identity'`` keeps each provider logged in independently, which is
+    useful for linked accounts, provider API aggregation, and step-up flows.
+
     Example::
 
         session = SessionMixin(InMemorySessionStore())
@@ -69,23 +93,115 @@ class OIDCMixin(KayaMixin):
             ...
     """
 
-    def __init__(self, config: OIDCConfig, session: SessionMixin) -> None:
+    def __init__(
+        self,
+        config: OIDCConfig,
+        session: SessionMixin,
+        strategy: IdentityStrategy = 'single-identity',
+    ) -> None:
+        if strategy not in _IDENTITY_STRATEGIES:
+            raise ValueError(
+                'strategy must be one of '
+                f'{", ".join(map(repr, _IDENTITY_STRATEGIES))} (got {strategy!r})'
+            )
         self._config = config
         self._session = session
         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:
+        return self._config.provider
+
+    @property
+    def strategy(self) -> IdentityStrategy:
+        return self._strategy
+
+    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()
+            auth_url, state, nonce, code_verifier = await self._client.build_authorization_url(
+                par=self._config.par
+            )
             session = self._session_of(ctx)
-            session['oidc_state'] = state
-            session['oidc_nonce'] = nonce
-            session['oidc_code_verifier'] = code_verifier
+            session[self._key('state')] = state
+            session[self._key('nonce')] = nonce
+            session[self._key('code_verifier')] = code_verifier
             await ctx.send_empty(302, {'Location': auth_url})
 
         @app.GET(self._config.callback_path)
@@ -108,13 +224,13 @@ class OIDCMixin(KayaMixin):
                 return
 
             session = self._session_of(ctx)
-            expected_state = session.get('oidc_state')
+            expected_state = session.get(self._key('state'))
             if state != expected_state:
                 await ctx.send_str(400, 'Invalid state')
                 return
 
-            nonce = session.get('oidc_nonce', '')
-            code_verifier = session.get('oidc_code_verifier', '')
+            nonce = session.get(self._key('nonce'), '')
+            code_verifier = session.get(self._key('code_verifier'), '')
 
             try:
                 token_response = await self._client.fetch_token(code, code_verifier)
@@ -124,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')
@@ -131,31 +253,151 @@ class OIDCMixin(KayaMixin):
                     userinfo = await self._client.fetch_userinfo(access_token)
                     user.update(userinfo)
 
-                session.pop('oidc_state', None)
-                session.pop('oidc_nonce', None)
-                session.pop('oidc_code_verifier', None)
+                session.pop(self._key('state'), None)
+                session.pop(self._key('code_verifier'), None)
+                # Keep this provider's nonce so refreshed ID tokens can be
+                # checked for nonce continuity (OIDC Core 12.2).
 
-                session['oidc_user'] = user
-                session['oidc_id_token'] = id_token
+                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['oidc_access_token'] = access_token
+                    session[self._key('access_token')] = access_token
                 if 'refresh_token' in token_response:
-                    session['oidc_refresh_token'] = token_response['refresh_token']
+                    session[self._key('refresh_token')] = token_response['refresh_token']
+
+                if self._strategy == 'single-identity':
+                    self._strip_other_identities(session)
 
                 session.regenerate_id()
                 await ctx.send_empty(302, {'Location': self._config.post_login_redirect})
-            except ValueError as exc:
+            except (ValueError, jwt.PyJWTError) as exc:
                 await ctx.send_str(400, f'Authentication failed: {exc}')
 
         @app.GET(self._config.logout_path)
         async def logout(ctx: HttpContext) -> None:
             session = self._session_of(ctx)
-            id_token = session.get('oidc_id_token')
-            session.invalidate()
+            id_token = session.get(self._key('id_token'))
+            self._logout_session(session)
             logout_url = await self._client.build_logout_url(id_token if isinstance(id_token, str) else None)
             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.
+
+        The shared session is only invalidated when no other provider remains
+        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):
+            session.invalidate()
+
+    def _strip_other_identities(self, session: Session) -> None:
+        """Remove every other provider's identity and tokens from the session.
+
+        Only identity keys are dropped; ephemeral flow keys (``state``,
+        ``nonce``, ``code_verifier``) are left alone so logins already in flight
+        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
@@ -169,14 +411,31 @@ class OIDCMixin(KayaMixin):
         return None
 
     def is_authenticated(self, ctx: HttpContext) -> bool:
-        return 'oidc_user' in self._session_of(ctx)
+        return self._key('user') in self._session_of(ctx)
 
     def get_user(self, ctx: HttpContext) -> Optional[OIDCUser]:
-        user = self._session_of(ctx).get('oidc_user')
+        user = self._session_of(ctx).get(self._key('user'))
         if user is None or not isinstance(user, Mapping):
             return None
         return OIDCUser(user)
 
+    def logged_in_providers(self, ctx: HttpContext) -> list[str]:
+        """Return the names of every provider with a user in this session."""
+        return [_provider_of_user_key(key)
+                for key in self._session_of(ctx) if _is_user_key(key)]
+
+    def get_users(self, ctx: HttpContext) -> dict[str, OIDCUser]:
+        """Return the users of every provider logged into this session."""
+        session = self._session_of(ctx)
+        users: dict[str, OIDCUser] = {}
+        for key in session:
+            if not _is_user_key(key):
+                continue
+            data = session.get(key)
+            if isinstance(data, Mapping):
+                users[_provider_of_user_key(key)] = OIDCUser(data)
+        return users
+
     def require_auth(self, handler: HttpHandler) -> HttpHandler:
         async def wrapper(ctx: HttpContext, *args: Any, **kwargs: Any) -> None:
             if not self.is_authenticated(ctx):
@@ -186,18 +445,53 @@ class OIDCMixin(KayaMixin):
         return wrapper
 
     async def refresh_access_token(self, session: Session) -> Optional[str]:
-        refresh_token = session.get('oidc_refresh_token')
+        refresh_token = session.get(self._key('refresh_token'))
         if not isinstance(refresh_token, str):
             return None
         try:
             token_response = await self._client.refresh_token(refresh_token)
             access_token = token_response.get('access_token')
+            id_token = token_response.get('id_token')
+            if isinstance(id_token, str):
+                stored_nonce = session.get(self._key('nonce'))
+                claims = await self._client.validate_refreshed_id_token(
+                    id_token,
+                    stored_nonce if isinstance(stored_nonce, str) else None,
+                )
+                user = session.get(self._key('user'))
+                original_sub = user.get('sub') if isinstance(user, Mapping) else None
+                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['oidc_access_token'] = access_token
-            if 'id_token' in token_response:
-                session['oidc_id_token'] = token_response['id_token']
+                session[self._key('access_token')] = access_token
+            if isinstance(id_token, str):
+                session[self._key('id_token')] = id_token
             if 'refresh_token' in token_response:
-                session['oidc_refresh_token'] = token_response['refresh_token']
+                session[self._key('refresh_token')] = token_response['refresh_token']
             return access_token if isinstance(access_token, str) else None
         except Exception:
             return None
+
+
+def _is_user_key(key: str) -> bool:
+    return key.startswith(f'{_SESSION_NAMESPACE}:') and key.endswith(f':{_USER_SUFFIX}')
+
+
+def _is_identity_key(key: str) -> bool:
+    return (key.startswith(f'{_SESSION_NAMESPACE}:')
+            and key.rsplit(':', 1)[-1] in _IDENTITY_FIELDS)
+
+
+def _provider_of_user_key(key: str) -> str:
+    return key[len(_SESSION_NAMESPACE) + 1:-(len(_USER_SUFFIX) + 1)]
+

packages/kaya-oidc/src/kaya/oidc/_providers.py

@@ -0,0 +1,141 @@
+from dataclasses import replace
+from types import MappingProxyType
+from typing import Callable, Mapping, Optional, Sequence
+
+from kaya.core import HttpContext, KayaApp, KayaMixin
+from kaya.session import Session, SessionMixin
+
+from ._config import OIDCConfig
+from ._mixin import (
+    _IDENTITY_STRATEGIES,
+    HttpHandler,
+    IdentityStrategy,
+    OIDCMixin,
+    OIDCUser,
+)
+
+
+class OIDCProviders(KayaMixin):
+    """Wire several OpenID Connect providers into one application.
+
+    Each config's ``provider`` becomes a route segment under ``base_path``:
+    ``{base_path}/{provider}/login``, ``{base_path}/{provider}/callback`` and
+    ``{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.
+
+    ``strategy`` controls how many identities may coexist in the session. The
+    default ``'single-identity'`` evicts every other provider's identity after a
+    successful login (last login wins). ``'multi-identity'`` keeps each provider
+    logged in independently, so logging out of one provider does not sign the
+    user out of the others.
+
+    Example::
+
+        session = SessionMixin(InMemorySessionStore())
+        oidc = OIDCProviders([google_config, github_config], session=session)
+        app = KayaApp(mixins=[session, oidc])
+
+        @app.GET('/profile')
+        @oidc.require_auth('google')
+        async def profile(ctx: HttpContext):
+            user = oidc.get_user(ctx, 'google')
+            ...
+    """
+
+    def __init__(
+        self,
+        configs: Sequence[OIDCConfig],
+        session: SessionMixin,
+        base_path: str = '/auth',
+        strategy: IdentityStrategy = 'single-identity',
+    ) -> None:
+        if not configs:
+            raise ValueError('At least one OIDCConfig is required')
+        if not base_path.startswith('/'):
+            raise ValueError('base_path must start with "/"')
+        if strategy not in _IDENTITY_STRATEGIES:
+            raise ValueError(
+                'strategy must be one of '
+                f'{", ".join(map(repr, _IDENTITY_STRATEGIES))} (got {strategy!r})'
+            )
+
+        self._session = session
+        self._strategy = strategy
+        self._base_path = base_path.rstrip('/')
+        self._providers: dict[str, OIDCMixin] = {}
+        for config in configs:
+            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,
+            )
+
+    @property
+    def base_path(self) -> str:
+        return self._base_path
+
+    @property
+    def strategy(self) -> IdentityStrategy:
+        return self._strategy
+
+    @property
+    def providers(self) -> Mapping[str, OIDCMixin]:
+        return MappingProxyType(self._providers)
+
+    def provider(self, name: str) -> OIDCMixin:
+        try:
+            return self._providers[name]
+        except KeyError:
+            raise KeyError(f'Unknown OIDC provider {name!r}') from None
+
+    def __getitem__(self, name: str) -> OIDCMixin:
+        return self.provider(name)
+
+    @property
+    def dependencies(self) -> Sequence[KayaMixin]:
+        return [self._session]
+
+    def apply(self, app: KayaApp) -> None:
+        for mixin in self._providers.values():
+            mixin.apply(app)
+
+    def is_authenticated(self, ctx: HttpContext, provider: str) -> bool:
+        return self.provider(provider).is_authenticated(ctx)
+
+    def get_user(self, ctx: HttpContext, provider: str) -> Optional[OIDCUser]:
+        return self.provider(provider).get_user(ctx)
+
+    def get_users(self, ctx: HttpContext) -> dict[str, OIDCUser]:
+        """Return the user of every provider with a session on ``ctx``."""
+        users: dict[str, OIDCUser] = {}
+        for name, mixin in self._providers.items():
+            user = mixin.get_user(ctx)
+            if user is not None:
+                users[name] = user
+        return users
+
+    def logged_in_providers(self, ctx: HttpContext) -> list[str]:
+        """Return the names of the providers logged into ``ctx``."""
+        return [name for name, mixin in self._providers.items() if mixin.is_authenticated(ctx)]
+
+    def require_auth(self, provider: str) -> Callable[[HttpHandler], HttpHandler]:
+        return self.provider(provider).require_auth
+
+    async def refresh_access_token(self, session: Session, provider: str) -> Optional[str]:
+        return await self.provider(provider).refresh_access_token(session)

packages/kaya-oidc/tests/test_backchannel_logout.py

@@ -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()

packages/kaya-openapi/README.md

@@ -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)
 )
 ```
 

packages/kaya-openapi/pyproject.toml

@@ -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"

packages/kaya-openapi/src/kaya/openapi/__init__.py

@@ -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',
 ]

packages/kaya-openapi/src/kaya/openapi/_metadata.py

@@ -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::
 

packages/kaya-openapi/src/kaya/openapi/_mixin.py

@@ -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)

packages/kaya-openapi/src/kaya/openapi/_spec.py

@@ -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

packages/kaya-openapi/tests/test_openapi.py

@@ -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()

packages/kaya-otel/README.md

@@ -69,9 +69,13 @@ async def home(ctx: HttpContext):
   `http.response.status_code` when the handler sends the response; 5xx
   statuses mark the span as failed. Exceptions escaping the handler are
   recorded on the span and also mark it as failed.
-- Every WebSocket connection gets one span for its whole lifetime. The close
-  code is recorded as `kaya.websocket.close_code`; exceptions and configured
-  error close codes mark the span as failed.
+- Every WebSocket connection gets one span for its whole lifetime, carrying
+  the applicable standard attributes: `url.scheme`, `url.path`, `url.query`,
+  `network.protocol.name` (`websocket`), `network.transport`,
+  `client.address`/`client.port` and `server.address`/`server.port`. The
+  close code is recorded as `kaya.websocket.close_code`; exceptions and
+  configured error close codes mark the span as failed and set `error.type`
+  (the exception class name, or the close code).
 - W3C `traceparent`/`tracestate` headers on incoming requests are honored,
   so traces propagate from upstream services.
 - Metrics:

packages/kaya-otel/pyproject.toml

@@ -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"

packages/kaya-otel/src/kaya/otel/_mixin.py

@@ -84,6 +84,30 @@ def _server_attributes(ctx: HttpContext) -> Dict[str, Any]:
     return attributes
 
 
+def _websocket_attributes(ws: WebSocket) -> Dict[str, Any]:
+    """Standard attributes applicable to a websocket connection span.
+
+    OpenTelemetry defines no websocket-specific conventions; these are the
+    general URL, network and peer attributes that apply to the connection.
+    """
+    attributes: Dict[str, Any] = {
+        'url.scheme': ws.scheme,
+        'url.path': ws.path,
+        'network.protocol.name': 'websocket',
+        'network.transport': 'tcp',
+    }
+    if ws.query_string:
+        attributes['url.query'] = ws.query_string
+    if ws.client is not None:
+        attributes['client.address'] = ws.client[0]
+        attributes['client.port'] = ws.client[1]
+    if ws.server is not None:
+        attributes['server.address'] = ws.server[0]
+        if ws.server[1] is not None:
+            attributes['server.port'] = ws.server[1]
+    return attributes
+
+
 def _normalize_header_name(name: str) -> str:
     return name.lower().replace('-', '_')
 
@@ -201,7 +225,11 @@ class OTelMixin(KayaMixin):
     usual HTTP semantic attributes; the response status code is recorded when
     the handler sends the response and 5xx statuses mark the span as failed.
     Exceptions escaping the handler are recorded and mark the span as failed.
-    WebSocket connections get one span for the whole connection lifetime.
+    WebSocket connections get one span for the whole connection lifetime,
+    carrying the applicable standard websocket URL, network and peer
+    attributes (``url.scheme``, ``url.path``, ``network.protocol.name``,
+    ``client.address``/``server.address``, ...); exceptions and configured
+    error close codes set ``error.type`` and mark the span as failed.
     W3C ``traceparent``/``tracestate`` headers on incoming requests are
     honored, so traces propagate from upstream services.
 
@@ -455,14 +483,7 @@ class OTelMixin(KayaMixin):
     async def _before_websocket(self, ws: WebSocket) -> Optional[WebSocket]:
         if self._is_excluded(ws.path):
             return None
-        holder = self._start_span(
-            f'WS {ws.path}',
-            {
-                'url.scheme': ws.scheme,
-                'url.path': ws.path,
-            },
-            ws.headers,
-        )
+        holder = self._start_span(f'WS {ws.path}', _websocket_attributes(ws), ws.headers)
         self._spans[id(ws)] = holder
         self._call_hook(self._websocket_connect_hook, holder.span, ws)
         return _TracedWebSocket(ws, holder)
@@ -475,10 +496,16 @@ class OTelMixin(KayaMixin):
         error = isinstance(exception, BaseException)
         if isinstance(exception, BaseException):
             holder.span.record_exception(exception)
+            holder.span.set_attribute('error.type', type(exception).__qualname__)
         close_code = holder.close_code
         if close_code is not None:
             holder.span.set_attribute('kaya.websocket.close_code', close_code)
-            error = error or close_code in self._websocket_error_close_codes
+            if close_code in self._websocket_error_close_codes:
+                error = True
+                if not isinstance(exception, BaseException):
+                    # Mirror the HTTP conventions, where server-side error
+                    # statuses are recorded as ``error.type`` values.
+                    holder.span.set_attribute('error.type', str(close_code))
         self._call_hook(self._websocket_close_hook, holder.span, ws)
         self._finish_span(holder, error)
 

packages/kaya-otel/tests/test_otel.py

@@ -279,7 +279,7 @@ class MetricsTest(unittest.TestCase):
 class WebSocketTracingTest(unittest.TestCase):
 
     @staticmethod
-    def _make_ws(headers=()):
+    def _make_ws(headers=(), query_string: bytes = b''):
         async def send(message):
             pass
 
@@ -289,7 +289,7 @@ class WebSocketTracingTest(unittest.TestCase):
         scope = {
             'type': 'websocket',
             'path': '/ws/games/abc',
-            'query_string': b'',
+            'query_string': query_string,
             'scheme': 'ws',
             'client': ('127.0.0.1', 12345),
             'server': ('127.0.0.1', 80),
@@ -308,7 +308,29 @@ class WebSocketTracingTest(unittest.TestCase):
         self.assertEqual(0, len(mixin._spans))
         spans = span_exporter.get_finished_spans()
         self.assertEqual(1, len(spans))
-        self.assertEqual('WS /ws/games/abc', spans[0].name)
+        span = spans[0]
+        self.assertEqual('WS /ws/games/abc', span.name)
+        assert span.attributes is not None
+        self.assertEqual('ws', span.attributes['url.scheme'])
+        self.assertEqual('/ws/games/abc', span.attributes['url.path'])
+        self.assertNotIn('url.query', span.attributes)
+        self.assertEqual('websocket', span.attributes['network.protocol.name'])
+        self.assertEqual('tcp', span.attributes['network.transport'])
+        self.assertEqual('127.0.0.1', span.attributes['client.address'])
+        self.assertEqual(12345, span.attributes['client.port'])
+        self.assertEqual('127.0.0.1', span.attributes['server.address'])
+        self.assertEqual(80, span.attributes['server.port'])
+
+    @async_test
+    async def test_websocket_query_string_recorded(self):
+        mixin, span_exporter, _ = make_mixin()
+        ws = self._make_ws(query_string=b'room=1&token=abc')
+        wrapped = await mixin._before_websocket(ws)
+        await mixin._after_websocket(wrapped)  # type: ignore[arg-type]
+        spans = span_exporter.get_finished_spans()
+        self.assertEqual(1, len(spans))
+        assert spans[0].attributes is not None
+        self.assertEqual('room=1&token=abc', spans[0].attributes['url.query'])
 
     @async_test
     async def test_websocket_traceparent_propagates(self):
@@ -344,6 +366,7 @@ class WebSocketTracingTest(unittest.TestCase):
         self.assertEqual(StatusCode.ERROR, spans[0].status.status_code)
         assert spans[0].attributes is not None
         self.assertEqual(1011, spans[0].attributes['kaya.websocket.close_code'])
+        self.assertEqual('1011', spans[0].attributes['error.type'])
 
     @async_test
     async def test_websocket_exception_marks_span_error(self):
@@ -358,6 +381,8 @@ class WebSocketTracingTest(unittest.TestCase):
         self.assertEqual(1, len(spans))
         self.assertEqual(StatusCode.ERROR, spans[0].status.status_code)
         self.assertTrue(any(event.name == 'exception' for event in spans[0].events))
+        assert spans[0].attributes is not None
+        self.assertEqual('RuntimeError', spans[0].attributes['error.type'])
 
     @async_test
     async def test_websocket_hooks_are_called(self):
@@ -386,6 +411,7 @@ class WebSocketTracingTest(unittest.TestCase):
         self.assertEqual(1, len(spans))
         self.assertEqual(StatusCode.UNSET, spans[0].status.status_code)
         assert spans[0].attributes is not None
+        self.assertNotIn('error.type', spans[0].attributes)
         self.assertTrue(spans[0].attributes['test.websocket_connect_hook'])
         self.assertTrue(spans[0].attributes['test.websocket_close_hook'])
 

packages/kaya-rsgi/pyproject.toml

@@ -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"

packages/kaya-session-memcache/README.md

@@ -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

packages/kaya-session-memcache/pyproject.toml

@@ -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"

packages/kaya-session-memcache/src/kaya/session/memcache/_store.py

@@ -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

packages/kaya-session-memcache/tests/test_indexes.py

@@ -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()

packages/kaya-session-memcache/tests/test_store.py

@@ -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

packages/kaya-session-redis/README.md

@@ -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

packages/kaya-session-redis/pyproject.toml

@@ -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"

packages/kaya-session-redis/src/kaya/session/redis/_store.py

@@ -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

packages/kaya-session-redis/tests/test_indexes.py

@@ -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()

packages/kaya-session-redis/tests/test_store.py

@@ -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

packages/kaya-session/README.md

@@ -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
 

packages/kaya-session/pyproject.toml

@@ -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"

packages/kaya-session/src/kaya/session/__init__.py

@@ -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',

packages/kaya-session/src/kaya/session/_mixin.py

@@ -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()

packages/kaya-session/src/kaya/session/_session.py

@@ -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]
 

packages/kaya-session/src/kaya/session/_store.py

@@ -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

packages/kaya-session/tests/test_indexes.py

@@ -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()

pyproject.toml

@@ -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

requirements-dev.in

@@ -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

requirements-dev.txt

@@ -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