Add Pushed Authorization Request (RFC 9126) support

OIDCConfig.par selects never (default), auto, or always. When enabled, build_authorization_url pushes the authorization parameters to the provider's PAR endpoint and redirects the browser with only client_id and request_uri. Providers that require PAR raise a clear error when configured with par='never'. Client authentication follows the token endpoint (HTTP Basic or public client).

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

@@ -4,7 +4,7 @@ from ._http_context import HttpContext
 from ._mixin import KayaMixin
 from ._tree import Tree, PathIterator
 from ._path_handler import PathHandler, Matches
-from ._websocket import WebSocket, WebSocketMessage
+from ._websocket import WebSocket, WebSocketDisconnect, WebSocketMessage
 
 
 __all__ = [
@@ -18,5 +18,6 @@ __all__ = [
     'Matches',
     'PathIterator',
     'WebSocket',
+    'WebSocketDisconnect',
     'WebSocketMessage'
 ]

packages/kaya-core/src/kaya/core/_app.py

@@ -9,7 +9,7 @@ from ._http_method import HttpMethod
 from ._mixin import KayaMixin
 from ._path_handler import Context
 from ._types import StrOrStrings
-from ._websocket import WebSocket, WebSocketMessage
+from ._websocket import WebSocket, WebSocketDisconnect, WebSocketMessage
 from ._asgi import AsgiContext, AsgiWebSocket
 from ._tree import Tree
 from ._types.asgi import LifespanScope, HTTPScope as ASGIHTTPScope, WebSocketScope as ASGIWebSocketScope
@@ -164,6 +164,12 @@ class KayaApp(AbstractKayaApp):
                 await handler.handle_request(ws, captured)
             else:
                 await ws.close(1000)
+        except WebSocketDisconnect:
+            # The peer went away while the handler was using the websocket
+            # (e.g. sending on a vanished transport). This is normal
+            # termination, not an application error, so it is not recorded
+            # on the socket nor re-raised.
+            pass
         except Exception as exc:
             ws.exception = exc
             raise

packages/kaya-core/src/kaya/core/_asgi.py

@@ -17,7 +17,7 @@ from pwo import Maybe
 from pathlib import Path
 from ._http_method import HttpMethod
 from ._http_context import HttpContext
-from ._websocket import WebSocket, WebSocketMessage
+from ._websocket import WebSocket, WebSocketDisconnect, WebSocketMessage
 from ._types import StrOrStrings
 from ._types.asgi import HTTPScope, WebSocketScope
 
@@ -179,7 +179,16 @@ class AsgiWebSocket(WebSocket):
             # Emit a list rather than a tuple: wsproto's AcceptConnection (used
             # by httpx_ws and others) requires list concatenation.
             message['headers'] = list(encode_headers(headers))
-        await self._send(message)
+        await self._send_message(message)
+
+    async def _send_message(self, message: Mapping[str, Any]) -> None:
+        try:
+            await self._send(message)
+        except OSError as exc:
+            # ASGI servers (e.g. uvicorn) report a vanished peer with an
+            # OSError subclass; surface it as a kaya-level disconnect so
+            # handlers do not need to know about server internals.
+            raise WebSocketDisconnect(code=1006) from exc
 
     async def receive(self) -> WebSocketMessage:
         message = await self._receive()
@@ -200,13 +209,13 @@ class AsgiWebSocket(WebSocket):
             return WebSocketMessage(kind='close')
 
     async def send_text(self, data: str) -> None:
-        await self._send({'type': 'websocket.send', 'text': data})
+        await self._send_message({'type': 'websocket.send', 'text': data})
 
     async def send_bytes(self, data: bytes) -> None:
-        await self._send({'type': 'websocket.send', 'bytes': data})
+        await self._send_message({'type': 'websocket.send', 'bytes': data})
 
     async def close(self, code: int = 1000) -> None:
-        await self._send({'type': 'websocket.close', 'code': code})
+        await self._send_message({'type': 'websocket.close', 'code': code})
 
     async def __anext__(self) -> WebSocketMessage:
         message = await self.receive()

packages/kaya-core/src/kaya/core/_websocket.py

@@ -13,6 +13,41 @@ class WebSocketMessage:
     data: WebSocketData = None
 
 
+class WebSocketDisconnect(Exception):
+    """Raised when using a websocket whose peer has disconnected.
+
+    Concrete websocket implementations raise this from ``send_text()`` and
+    ``send_bytes()`` when the transport reports that the peer is gone (for
+    example when a browser tab is closed without a close handshake), so
+    handlers can catch it and stop producing output:
+
+    .. code-block:: python
+
+        try:
+            await ws.send_text('hello')
+        except WebSocketDisconnect:
+            pass
+
+    Reading from a disconnected websocket is not an error: ``receive()``
+    reports a ``close`` message and iterating with ``async for`` simply ends,
+    which is what most handlers should rely on instead.
+
+    ``code`` is a websocket close code; ``1006`` means the connection was
+    closed abnormally, without a close handshake.
+    """
+
+    code: int
+    reason: str
+
+    def __init__(self, code: int = 1006, reason: str = '') -> None:
+        self.code = code
+        self.reason = reason
+        message = f'WebSocket disconnected (code={code})'
+        if reason:
+            message = f'{message}: {reason}'
+        super().__init__(message)
+
+
 class WebSocket(ABC):
     path: str
     query_string: str

packages/kaya-core/tests/test_websocket.py

@@ -1,9 +1,9 @@
 import unittest
 import httpx
 from pwo import async_test
-from httpx_ws import aconnect_ws, WebSocketDisconnect
+from httpx_ws import aconnect_ws, WebSocketDisconnect as HttpxWebSocketDisconnect
 from httpx_ws.transport import ASGIWebSocketTransport
-from kaya.core import KayaApp, WebSocket
+from kaya.core import KayaApp, WebSocket, WebSocketDisconnect, WebSocketMessage
 from kaya.core._asgi import AsgiWebSocket
 
 
@@ -66,7 +66,7 @@ class WebSocketTest(unittest.TestCase):
         transport = ASGIWebSocketTransport(app=self.app)
 
         async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
-            with self.assertRaises(WebSocketDisconnect) as cm:
+            with self.assertRaises(HttpxWebSocketDisconnect) as cm:
                 async with aconnect_ws("/unknown", client) as ws:
                     pass
 
@@ -151,3 +151,113 @@ class WebSocketTest(unittest.TestCase):
         }
         ws = AsgiWebSocket(scope, receive, send)
         self.assertEqual('ws', ws.scheme)
+
+
+class FakeWebSocket(WebSocket):
+    """Minimal in-memory websocket for exercising ``KayaApp`` directly."""
+
+    def __init__(self, path: str) -> None:
+        self.path = path
+        self.query_string = ''
+        self.scheme = 'ws'
+        self.headers: dict[str, list[str]] = {}
+        self.client = ('127.0.0.1', 12345)
+        self.server = ('127.0.0.1', 80)
+        self.closed: list[int] = []
+
+    async def accept(self, headers=None) -> None:
+        pass
+
+    async def receive(self) -> WebSocketMessage:
+        return WebSocketMessage(kind='close')
+
+    async def send_text(self, data: str) -> None:
+        pass
+
+    async def send_bytes(self, data: bytes) -> None:
+        pass
+
+    async def close(self, code: int = 1000) -> None:
+        self.closed.append(code)
+
+    async def __anext__(self) -> WebSocketMessage:
+        raise StopAsyncIteration
+
+
+class WebSocketDisconnectTest(unittest.TestCase):
+
+    def setUp(self):
+        self.app = KayaApp()
+
+        @self.app.websocket('/disconnect')
+        async def disconnect(ws: WebSocket) -> None:
+            await ws.accept()
+            await ws.send_text('before')
+            raise WebSocketDisconnect()
+
+        @self.app.websocket('/failure')
+        async def failure(ws: WebSocket) -> None:
+            await ws.accept()
+            raise RuntimeError('boom')
+
+    @async_test
+    async def test_disconnect_terminates_quietly(self):
+        after_hook_calls = []
+
+        async def after(ws: WebSocket) -> None:
+            after_hook_calls.append(ws)
+
+        self.app.add_after_websocket_hook(after)
+        ws = FakeWebSocket('/disconnect')
+        await self.app.handle_websocket(ws)
+        self.assertEqual([ws], after_hook_calls)
+        self.assertIsNone(ws.exception)
+
+    @async_test
+    async def test_other_errors_are_recorded_and_reraised(self):
+        ws = FakeWebSocket('/failure')
+        with self.assertRaises(RuntimeError):
+            await self.app.handle_websocket(ws)
+        self.assertIsInstance(ws.exception, RuntimeError)
+
+    @async_test
+    async def test_asgi_send_after_disconnect_raises_websocket_disconnect(self):
+        async def send(message):
+            raise OSError('client gone')
+
+        async def receive():
+            return {'type': 'websocket.disconnect', 'code': 1006}
+
+        ws = AsgiWebSocket(self._scope(), receive, send)
+        with self.assertRaises(WebSocketDisconnect) as ctx:
+            await ws.send_text('hello')
+        self.assertEqual(1006, ctx.exception.code)
+        with self.assertRaises(WebSocketDisconnect):
+            await ws.send_bytes(b'hello')
+        with self.assertRaises(WebSocketDisconnect):
+            await ws.close()
+
+    @async_test
+    async def test_asgi_accept_after_disconnect_raises_websocket_disconnect(self):
+        async def send(message):
+            raise OSError('client gone')
+
+        async def receive():
+            return {'type': 'websocket.connect'}
+
+        ws = AsgiWebSocket(self._scope(), receive, send)
+        with self.assertRaises(WebSocketDisconnect) as ctx:
+            await ws.accept()
+        self.assertEqual(1006, ctx.exception.code)
+
+    @staticmethod
+    def _scope() -> dict:
+        return {
+            'type': 'websocket',
+            'path': '/echo',
+            'query_string': b'',
+            'scheme': 'ws',
+            'client': ('127.0.0.1', 12345),
+            'server': ('127.0.0.1', 80),
+            'headers': [],
+        }

packages/kaya-oidc/README.md

@@ -40,6 +40,125 @@ 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.
+
+## 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
@@ -50,6 +169,9 @@ also works because the app applies mixin dependencies automatically.
 - Optional userinfo endpoint fetch
 - Refresh token support
 - RP-initiated logout (when provider advertises `end_session_endpoint`)
+- 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.)
 
 ## Security notes

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

@@ -1,11 +1,15 @@
 from ._client import OIDCClient
-from ._config import OIDCConfig
+from ._config import OIDCConfig, PARPolicy
-from ._mixin import OIDCMixin, OIDCUser
+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

@@ -5,7 +5,7 @@ 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
 
 
@@ -42,8 +42,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]:
+    async def build_authorization_url(self, par: PARPolicy = 'never') -> tuple[str, str, str, str]:
-        """Return (authorization_url, state, nonce, code_verifier)."""
+        """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 +60,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 +68,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 +142,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,11 +158,8 @@ 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())
 

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

@@ -1,13 +1,31 @@
+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.
+
     ``client_secret`` is optional; omit it for public clients.
     ``fetch_userinfo`` controls whether the userinfo endpoint is queried after
     token exchange.
@@ -35,8 +53,21 @@ class OIDCConfig:
         'EdDSA',
     ))
     http_client: Optional[httpx.AsyncClient] = None
+    provider: str = 'default'
+    par: PARPolicy = 'never'
 
     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})'
+            )
         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,4 +1,4 @@
-from typing import Any, Awaitable, Callable, Mapping, Optional, Sequence
+from typing import Any, Awaitable, Callable, Literal, Mapping, Optional, Sequence, cast
 
 from kaya.core import HttpContext, KayaApp, KayaMixin
 from kaya.session import Session, SessionMixin
@@ -10,6 +10,13 @@ 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'})
+_IDENTITY_STRATEGIES: tuple[IdentityStrategy, ...] = ('single-identity', 'multi-identity')
+
 
 class OIDCUser(Mapping[str, Any]):
     """Read-only view of the OIDC userinfo / ID token claims."""
@@ -28,7 +35,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 +63,17 @@ class OIDCMixin(KayaMixin):
     - ``callback_path`` (default ``/auth/callback``): handles the provider callback.
     - ``logout_path`` (default ``/auth/logout``): logs the user out.
 
+    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,10 +87,33 @@ 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}:'
+
+    @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}'
 
     @property
     def dependencies(self) -> Sequence[KayaMixin]:
@@ -81,11 +122,13 @@ class OIDCMixin(KayaMixin):
     def apply(self, app: KayaApp) -> None:
         @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[self._key('state')] = state
-            session['oidc_nonce'] = nonce
+            session[self._key('nonce')] = nonce
-            session['oidc_code_verifier'] = code_verifier
+            session[self._key('code_verifier')] = code_verifier
             await ctx.send_empty(302, {'Location': auth_url})
 
         @app.GET(self._config.callback_path)
@@ -108,13 +151,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', '')
+            nonce = session.get(self._key('nonce'), '')
-            code_verifier = session.get('oidc_code_verifier', '')
+            code_verifier = session.get(self._key('code_verifier'), '')
 
             try:
                 token_response = await self._client.fetch_token(code, code_verifier)
@@ -131,16 +174,19 @@ class OIDCMixin(KayaMixin):
                     userinfo = await self._client.fetch_userinfo(access_token)
                     user.update(userinfo)
 
-                session.pop('oidc_state', None)
+                session.pop(self._key('state'), None)
-                session.pop('oidc_nonce', None)
+                session.pop(self._key('nonce'), None)
-                session.pop('oidc_code_verifier', None)
+                session.pop(self._key('code_verifier'), None)
 
-                session['oidc_user'] = user
+                session[self._key('user')] = user
-                session['oidc_id_token'] = id_token
+                session[self._key('id_token')] = id_token
                 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})
@@ -150,12 +196,36 @@ class OIDCMixin(KayaMixin):
         @app.GET(self._config.logout_path)
         async def logout(ctx: HttpContext) -> None:
             session = self._session_of(ctx)
-            id_token = session.get('oidc_id_token')
+            id_token = session.get(self._key('id_token'))
-            session.invalidate()
+            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})
 
+    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.
+        """
+        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.
+        """
+        for key in [key for key in session
+                    if _is_identity_key(key) and not key.startswith(self._prefix)]:
+            del session[key]
+
     @staticmethod
     def _session_of(ctx: HttpContext) -> Session:
         session = ctx.session
@@ -169,14 +239,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 +273,32 @@ 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')
             if isinstance(access_token, str):
-                session['oidc_access_token'] = access_token
+                session[self._key('access_token')] = access_token
             if 'id_token' in token_response:
-                session['oidc_id_token'] = token_response['id_token']
+                session[self._key('id_token')] = token_response['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,134 @@
+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``. 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}')
+            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',
+                ),
+                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_oidc.py

@@ -1,7 +1,8 @@
 import base64
 import unittest
+from dataclasses import replace
 from time import time
-from typing import Mapping
+from typing import Mapping, Optional
 from urllib.parse import parse_qs, urlparse
 
 import httpx
@@ -11,17 +12,32 @@ from pwo import async_test
 from kaya.core import HttpContext, KayaApp
 from kaya.session import InMemorySessionStore, SessionMixin
 
-from kaya.oidc import OIDCConfig, OIDCMixin
+from kaya.oidc import OIDCConfig, OIDCMixin, OIDCProviders
 from kaya.oidc._client import OIDCClient
 
 
 class MockOIDCProvider:
-    def __init__(self) -> None:
+    def __init__(
+        self,
+        name: str = 'mock',
+        *,
+        issuer: Optional[str] = None,
+        sub: str = 'user123',
+        email: str = 'user@example.com',
+        supports_par: bool = False,
+        require_par: bool = False,
+        par_error: bool = False,
+    ) -> None:
         self.private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
         self.public_key = self.private_key.public_key()
         self.kid = 'mock-key'
-        self.issuer = 'https://mock-oidc.local'
+        self.issuer = issuer or f'https://{name}-oidc.local'
-        self.nonce = 'nonce-value'
+        self.sub = sub
+        self.email = email
+        self.nonce = f'{name}-nonce-value'
+        self.par_error = par_error
+        self.par_requests: list[dict[str, list[str]]] = []
+        self.par_auth_headers: list[Optional[str]] = []
         self.discovery = {
             'issuer': self.issuer,
             'authorization_endpoint': f'{self.issuer}/auth',
@@ -30,6 +46,10 @@ class MockOIDCProvider:
             'end_session_endpoint': f'{self.issuer}/logout',
             'jwks_uri': f'{self.issuer}/jwks',
         }
+        if supports_par or require_par:
+            self.discovery['pushed_authorization_request_endpoint'] = f'{self.issuer}/par'
+        if require_par:
+            self.discovery['require_pushed_authorization_requests'] = True
 
     def _public_key_to_jwk(self) -> dict[str, str]:
         numbers = self.public_key.public_numbers()
@@ -48,7 +68,7 @@ class MockOIDCProvider:
         now = time()
         exp = now - 3600 if expired else now + 3600
         payload = {
-            'sub': 'user123',
+            'sub': self.sub,
             'iss': self.issuer,
             'aud': audience,
             'iat': now,
@@ -70,6 +90,20 @@ class MockOIDCProvider:
             return httpx.Response(200, json=self.discovery)
         if path == '/jwks':
             return httpx.Response(200, json={'keys': [self._public_key_to_jwk()]})
+        if path == '/par':
+            if request.method != 'POST':
+                return httpx.Response(405)
+            self.par_requests.append(parse_qs(request.content.decode()))
+            self.par_auth_headers.append(request.headers.get('Authorization'))
+            if self.par_error:
+                return httpx.Response(400, json={
+                    'error': 'invalid_request',
+                    'error_description': 'mock PAR failure',
+                })
+            return httpx.Response(201, json={
+                'request_uri': f'urn:ietf:params:oauth:request_uri:par-{len(self.par_requests)}',
+                'expires_in': 60,
+            })
         if path == '/token' and request.method == 'POST':
             body = request.content.decode()
             data = dict(part.split('=') for part in body.split('&')) if body else {}
@@ -87,7 +121,7 @@ class MockOIDCProvider:
                 'token_type': 'Bearer',
             })
         if path == '/userinfo':
-            return httpx.Response(200, json={'sub': 'user123', 'email': 'user@example.com'})
+            return httpx.Response(200, json={'sub': self.sub, 'email': self.email})
         if path == '/logout':
             return httpx.Response(200)
         return httpx.Response(404)
@@ -194,6 +228,236 @@ class OIDCClientTest(unittest.TestCase):
         self.assertEqual(['id-token'], query['id_token_hint'])
         self.assertEqual(['/'], query['post_logout_redirect_uri'])
 
+    @async_test
+    async def test_build_authorization_url_with_par(self) -> None:
+        provider = MockOIDCProvider('par', supports_par=True)
+        http_client = httpx.AsyncClient(transport=MockTransport(provider))
+        config = OIDCConfig(
+            issuer=provider.issuer,
+            client_id='client',
+            client_secret='secret',
+            redirect_uri='http://localhost:8000/auth/callback',
+            http_client=http_client,
+            par='always',
+        )
+        client = OIDCClient(config)
+
+        url, state, nonce, verifier = await client.build_authorization_url(par='always')
+
+        query = parse_qs(urlparse(url).query)
+        self.assertEqual(['client'], query['client_id'])
+        self.assertEqual(['urn:ietf:params:oauth:request_uri:par-1'], query['request_uri'])
+        self.assertNotIn('state', query)
+        self.assertNotIn('nonce', query)
+        self.assertNotIn('code_challenge', query)
+
+        self.assertEqual(1, len(provider.par_requests))
+        data = provider.par_requests[0]
+        self.assertEqual(['client'], data['client_id'])
+        self.assertEqual(['code'], data['response_type'])
+        self.assertEqual([state], data['state'])
+        self.assertEqual([nonce], data['nonce'])
+        self.assertEqual(['S256'], data['code_challenge_method'])
+        self.assertIn('openid', data['scope'][0])
+        self.assertIn('code_challenge', data)
+        self.assertTrue(len(verifier) > 0)
+
+        auth = provider.par_auth_headers[0]
+        self.assertIsNotNone(auth)
+        assert auth is not None
+        self.assertTrue(auth.startswith('Basic '))
+
+    @async_test
+    async def test_par_auto_uses_par_only_when_advertised(self) -> None:
+        provider = MockOIDCProvider('par', supports_par=True)
+        http_client = httpx.AsyncClient(transport=MockTransport(provider))
+        config = OIDCConfig(
+            issuer=provider.issuer,
+            client_id='client',
+            redirect_uri='http://localhost:8000/auth/callback',
+            http_client=http_client,
+        )
+        client = OIDCClient(config)
+
+        url, *_ = await client.build_authorization_url(par='auto')
+        query = parse_qs(urlparse(url).query)
+        self.assertIn('request_uri', query)
+        self.assertEqual(1, len(provider.par_requests))
+
+        # The default mock provider does not advertise a PAR endpoint.
+        url, *_ = await self.client.build_authorization_url(par='auto')
+        query = parse_qs(urlparse(url).query)
+        self.assertNotIn('request_uri', query)
+        self.assertIn('state', query)
+
+    @async_test
+    async def test_par_always_without_endpoint_raises(self) -> None:
+        with self.assertRaises(RuntimeError):
+            await self.client.build_authorization_url(par='always')
+
+    @async_test
+    async def test_required_par_with_never_raises(self) -> None:
+        provider = MockOIDCProvider('req', require_par=True)
+        http_client = httpx.AsyncClient(transport=MockTransport(provider))
+        config = OIDCConfig(
+            issuer=provider.issuer,
+            client_id='client',
+            redirect_uri='http://localhost:8000/auth/callback',
+            http_client=http_client,
+        )
+        client = OIDCClient(config)
+        with self.assertRaises(RuntimeError) as ctx:
+            await client.build_authorization_url(par='never')
+        self.assertIn('set par', str(ctx.exception))
+        self.assertEqual(0, len(provider.par_requests))
+
+    @async_test
+    async def test_required_par_with_auto_uses_par(self) -> None:
+        provider = MockOIDCProvider('req', require_par=True)
+        http_client = httpx.AsyncClient(transport=MockTransport(provider))
+        config = OIDCConfig(
+            issuer=provider.issuer,
+            client_id='client',
+            redirect_uri='http://localhost:8000/auth/callback',
+            http_client=http_client,
+        )
+        client = OIDCClient(config)
+        url, *_ = await client.build_authorization_url(par='auto')
+        self.assertIn('request_uri', parse_qs(urlparse(url).query))
+        self.assertEqual(1, len(provider.par_requests))
+
+    @async_test
+    async def test_par_error_response_propagates(self) -> None:
+        provider = MockOIDCProvider('par', supports_par=True, par_error=True)
+        http_client = httpx.AsyncClient(transport=MockTransport(provider))
+        config = OIDCConfig(
+            issuer=provider.issuer,
+            client_id='client',
+            redirect_uri='http://localhost:8000/auth/callback',
+            http_client=http_client,
+            par='always',
+        )
+        client = OIDCClient(config)
+        with self.assertRaises(httpx.HTTPStatusError):
+            await client.build_authorization_url(par='always')
+        self.assertEqual(1, len(provider.par_requests))
+
+    @async_test
+    async def test_par_public_client_uses_no_basic_auth(self) -> None:
+        provider = MockOIDCProvider('par', supports_par=True)
+        http_client = httpx.AsyncClient(transport=MockTransport(provider))
+        config = OIDCConfig(
+            issuer=provider.issuer,
+            client_id='client',
+            redirect_uri='http://localhost:8000/auth/callback',
+            http_client=http_client,
+            par='always',
+        )
+        client = OIDCClient(config)
+        await client.build_authorization_url(par='always')
+        self.assertIsNone(provider.par_auth_headers[0])
+        self.assertEqual(['client'], provider.par_requests[0]['client_id'])
+
+    def test_config_par_defaults_to_never_and_validates(self) -> None:
+        config = OIDCConfig(
+            issuer='https://issuer.local',
+            client_id='client',
+            redirect_uri='http://localhost/callback',
+        )
+        self.assertEqual('never', config.par)
+        with self.assertRaises(ValueError):
+            OIDCConfig(
+                issuer='https://issuer.local',
+                client_id='client',
+                redirect_uri='http://localhost/callback',
+                par='bogus',
+            )
+
+
+class OIDCParTest(unittest.TestCase):
+
+    @async_test
+    async def test_login_and_callback_with_par(self) -> None:
+        provider = MockOIDCProvider('par', email='par@example.com', supports_par=True)
+        http_client = httpx.AsyncClient(transport=MockTransport(provider))
+        store = InMemorySessionStore()
+        session = SessionMixin(store)
+        config = OIDCConfig(
+            issuer=provider.issuer,
+            client_id='client',
+            client_secret='secret',
+            redirect_uri='http://localhost:8000/auth/callback',
+            http_client=http_client,
+            par='auto',
+            fetch_userinfo=True,
+        )
+        oidc = OIDCMixin(config, session=session)
+        app = KayaApp(mixins=[session, oidc])
+
+        @app.GET('/profile')
+        @oidc.require_auth
+        async def profile(ctx: HttpContext) -> None:
+            user = oidc.get_user(ctx)
+            await ctx.send_str(200, user.email if user and user.email else 'none')
+
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            r = await client.get('/auth/login', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            query = parse_qs(urlparse(r.headers['Location']).query)
+            self.assertIn('request_uri', query)
+            self.assertNotIn('state', query)
+            self.assertEqual(1, len(provider.par_requests))
+
+            state = provider.par_requests[0]['state'][0]
+            provider.nonce = provider.par_requests[0]['nonce'][0]
+
+            r = await client.get('/auth/callback', params={'code': 'mock-code', 'state': state}, follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+
+            r = await client.get('/profile')
+            self.assertEqual('par@example.com', r.text)
+
+    @async_test
+    async def test_container_applies_policy_per_provider(self) -> None:
+        par_provider = MockOIDCProvider('par', supports_par=True)
+        plain_provider = MockOIDCProvider('plain')
+        par_http = httpx.AsyncClient(transport=MockTransport(par_provider))
+        plain_http = httpx.AsyncClient(transport=MockTransport(plain_provider))
+
+        par_config = OIDCConfig(
+            issuer=par_provider.issuer,
+            client_id='par-client',
+            redirect_uri='http://localhost:8000/auth/withpar/callback',
+            http_client=par_http,
+            provider='withpar',
+            par='always',
+        )
+        plain_config = OIDCConfig(
+            issuer=plain_provider.issuer,
+            client_id='plain-client',
+            redirect_uri='http://localhost:8000/auth/plain/callback',
+            http_client=plain_http,
+            provider='plain',
+            par='never',
+        )
+        session = SessionMixin(InMemorySessionStore())
+        oidc = OIDCProviders([par_config, plain_config], session=session)
+        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:
+            r = await client.get('/auth/withpar/login', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            self.assertIn('request_uri', parse_qs(urlparse(r.headers['Location']).query))
+
+            r = await client.get('/auth/plain/login', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            plain_query = parse_qs(urlparse(r.headers['Location']).query)
+            self.assertNotIn('request_uri', plain_query)
+            self.assertIn('state', plain_query)
+
+
 
 class OIDCMixinTest(unittest.TestCase):
 
@@ -356,5 +620,362 @@ class OIDCMixinTest(unittest.TestCase):
             self.assertIn('Set-Cookie', r.headers)
 
 
+class OIDCMultiProviderTest(unittest.TestCase):
+
+    def _provider_config(self, provider: MockOIDCProvider, name: str) -> tuple[OIDCConfig, httpx.AsyncClient]:
+        http_client = httpx.AsyncClient(transport=MockTransport(provider))
+        config = OIDCConfig(
+            issuer=provider.issuer,
+            client_id=f'{name}-client',
+            client_secret='secret',
+            provider=name,
+            redirect_uri=f'http://localhost:8000/auth/{name}/callback',
+            http_client=http_client,
+            fetch_userinfo=True,
+        )
+        return config, http_client
+
+    def _build(self, strategy: str = 'multi-identity') -> tuple[KayaApp, OIDCProviders, InMemorySessionStore]:
+        self.google = MockOIDCProvider('google', sub='g-sub', email='g@example.com')
+        self.github = MockOIDCProvider('github', sub='h-sub', email='h@example.com')
+        google_config, self.google_http = self._provider_config(self.google, 'google')
+        github_config, self.github_http = self._provider_config(self.github, 'github')
+
+        store = InMemorySessionStore()
+        session = SessionMixin(store)
+        oidc = OIDCProviders([google_config, github_config], session=session, strategy=strategy)
+        app = KayaApp(mixins=[session, oidc])
+
+        @app.GET('/who')
+        async def who(ctx: HttpContext) -> None:
+            users = oidc.get_users(ctx)
+            await ctx.send_str(200, ','.join(sorted(users)) or 'none')
+
+        @app.GET('/google-profile')
+        @oidc.require_auth('google')
+        async def google_profile(ctx: HttpContext) -> None:
+            user = oidc.get_user(ctx, 'google')
+            assert user is not None
+            await ctx.send_str(200, user.email or '')
+
+        @app.GET('/github-profile')
+        @oidc.require_auth('github')
+        async def github_profile(ctx: HttpContext) -> None:
+            user = oidc.get_user(ctx, 'github')
+            assert user is not None
+            await ctx.send_str(200, user.email or '')
+
+        @app.GET('/refresh-github')
+        @oidc.require_auth('github')
+        async def refresh_github(ctx: HttpContext) -> None:
+            token = await oidc.refresh_access_token(ctx.session, 'github')
+            await ctx.send_str(200, token or 'no-token')
+
+        return app, oidc, store
+
+    async def _complete_login(self, client: httpx.AsyncClient, name: str, state: str) -> None:
+        r = await client.get(
+            f'/auth/{name}/callback',
+            params={'code': 'mock-code', 'state': state},
+            follow_redirects=False,
+        )
+        self.assertEqual(302, r.status_code)
+
+    async def _login(self, client: httpx.AsyncClient, provider: MockOIDCProvider, name: str) -> None:
+        r = await client.get(f'/auth/{name}/login', follow_redirects=False)
+        self.assertEqual(302, r.status_code)
+        query = parse_qs(urlparse(r.headers['Location']).query)
+        provider.nonce = query['nonce'][0]
+        await self._complete_login(client, name, query['state'][0])
+
+    @async_test
+    async def test_independent_logins(self) -> None:
+        app, oidc, store = self._build()
+        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')
+
+            r = await client.get('/google-profile')
+            self.assertEqual('g@example.com', r.text)
+
+            # The github provider is untouched by the google login.
+            r = await client.get('/github-profile', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            self.assertEqual('/auth/github/login', r.headers['Location'])
+
+            r = await client.get('/who')
+            self.assertEqual('google', r.text)
+
+    @async_test
+    async def test_interleaved_login_flows(self) -> None:
+        app, oidc, store = self._build()
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            # Start both flows before completing either one.
+            r = await client.get('/auth/google/login', follow_redirects=False)
+            google_query = parse_qs(urlparse(r.headers['Location']).query)
+            r = await client.get('/auth/github/login', follow_redirects=False)
+            github_query = parse_qs(urlparse(r.headers['Location']).query)
+
+            self.google.nonce = google_query['nonce'][0]
+            self.github.nonce = github_query['nonce'][0]
+
+            # Completing google must not invalidate the pending github state.
+            await self._complete_login(client, 'google', google_query['state'][0])
+            await self._complete_login(client, 'github', github_query['state'][0])
+
+            r = await client.get('/who')
+            self.assertEqual('github,google', r.text)
+
+    @async_test
+    async def test_logout_one_provider_keeps_the_other(self) -> None:
+        app, oidc, store = self._build()
+        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')
+            await self._login(client, self.github, 'github')
+
+            session_id = client.cookies['session_id']
+            self.assertIn(session_id, store._data)
+
+            r = await client.get('/auth/github/logout', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            self.assertIn(self.github.discovery['end_session_endpoint'], r.headers['Location'])
+
+            # Google is still logged in and the shared session survives.
+            r = await client.get('/google-profile')
+            self.assertEqual('g@example.com', r.text)
+            r = await client.get('/github-profile', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            self.assertIn(session_id, store._data)
+
+            # Logging out the last provider invalidates the session.
+            r = await client.get('/auth/google/logout', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            self.assertNotIn(session_id, store._data)
+
+    @async_test
+    async def test_provider_refresh_is_namespaced(self) -> None:
+        app, oidc, store = self._build()
+        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')
+            await self._login(client, self.github, 'github')
+
+            # Refreshing github must not clobber google's session state.
+            r = await client.get('/refresh-github')
+            self.assertEqual('new-access-token', r.text)
+            r = await client.get('/google-profile')
+            self.assertEqual('g@example.com', r.text)
+
+    @async_test
+    async def test_multiple_mixin_instances_compose(self) -> None:
+        google = MockOIDCProvider('google', sub='g-sub', email='g@example.com')
+        github = MockOIDCProvider('github', sub='h-sub', email='h@example.com')
+        google_config, google_http = self._provider_config(google, 'google')
+        github_config, github_http = self._provider_config(github, 'github')
+
+        store = InMemorySessionStore()
+        session = SessionMixin(store)
+        google_mixin = OIDCMixin(
+            replace(google_config, login_path='/google/login', callback_path='/google/callback', logout_path='/google/logout'),
+            session=session,
+        )
+        github_mixin = OIDCMixin(
+            replace(github_config, login_path='/github/login', callback_path='/github/callback', logout_path='/github/logout'),
+            session=session,
+        )
+        app = KayaApp(mixins=[session, google_mixin, github_mixin])
+
+        @app.GET('/g')
+        @google_mixin.require_auth
+        async def g(ctx: HttpContext) -> None:
+            user = google_mixin.get_user(ctx)
+            assert user is not None
+            await ctx.send_str(200, user.email or '')
+
+        @app.GET('/h')
+        @github_mixin.require_auth
+        async def h(ctx: HttpContext) -> None:
+            user = github_mixin.get_user(ctx)
+            assert user is not None
+            await ctx.send_str(200, user.email or '')
+
+        @app.GET('/who')
+        async def who(ctx: HttpContext) -> None:
+            users = google_mixin.get_users(ctx)
+            providers = google_mixin.logged_in_providers(ctx)
+            summary = ','.join(f'{name}={users[name].email}' for name in sorted(users))
+            await ctx.send_str(200, f'{",".join(sorted(providers))}|{summary}')
+
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            r = await client.get('/google/login', follow_redirects=False)
+            query = parse_qs(urlparse(r.headers['Location']).query)
+            google.nonce = query['nonce'][0]
+            r = await client.get('/google/callback', params={'code': 'c', 'state': query['state'][0]}, follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+
+            r = await client.get('/g')
+            self.assertEqual('g@example.com', r.text)
+            r = await client.get('/h', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+
+            # Mixin-level helpers see the namespaced user across providers.
+            r = await client.get('/who')
+            self.assertEqual('google|google=g@example.com', r.text)
+
+    @async_test
+    async def test_single_identity_evicts_previous_provider(self) -> None:
+        app, oidc, store = self._build(strategy='single-identity')
+        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')
+            r = await client.get('/google-profile')
+            self.assertEqual('g@example.com', r.text)
+
+            # Logging in with github replaces the google identity.
+            await self._login(client, self.github, 'github')
+
+            r = await client.get('/who')
+            self.assertEqual('github', r.text)
+            r = await client.get('/google-profile', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            self.assertEqual('/auth/google/login', r.headers['Location'])
+            r = await client.get('/github-profile')
+            self.assertEqual('h@example.com', r.text)
+
+            data = store._data[client.cookies['session_id']]
+            self.assertNotIn('oidc:google:user', data)
+            self.assertNotIn('oidc:google:id_token', data)
+            self.assertIn('oidc:github:user', data)
+
+    @async_test
+    async def test_single_identity_last_completed_login_wins(self) -> None:
+        app, oidc, store = self._build(strategy='single-identity')
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            # Start both flows before completing either one; pending flow state
+            # is not evicted, so both callbacks can complete.
+            r = await client.get('/auth/google/login', follow_redirects=False)
+            google_query = parse_qs(urlparse(r.headers['Location']).query)
+            r = await client.get('/auth/github/login', follow_redirects=False)
+            github_query = parse_qs(urlparse(r.headers['Location']).query)
+
+            self.google.nonce = google_query['nonce'][0]
+            self.github.nonce = github_query['nonce'][0]
+
+            await self._complete_login(client, 'google', google_query['state'][0])
+            await self._complete_login(client, 'github', github_query['state'][0])
+
+            r = await client.get('/who')
+            self.assertEqual('github', r.text)
+
+    @async_test
+    async def test_standalone_mixins_default_to_single_identity(self) -> None:
+        google = MockOIDCProvider('google', sub='g-sub', email='g@example.com')
+        github = MockOIDCProvider('github', sub='h-sub', email='h@example.com')
+        google_config, google_http = self._provider_config(google, 'google')
+        github_config, github_http = self._provider_config(github, 'github')
+
+        session = SessionMixin(InMemorySessionStore())
+        google_mixin = OIDCMixin(
+            replace(google_config, login_path='/google/login', callback_path='/google/callback', logout_path='/google/logout'),
+            session=session,
+        )
+        github_mixin = OIDCMixin(
+            replace(github_config, login_path='/github/login', callback_path='/github/callback', logout_path='/github/logout'),
+            session=session,
+        )
+        app = KayaApp(mixins=[session, google_mixin, github_mixin])
+
+        self.assertEqual('single-identity', google_mixin.strategy)
+
+        @app.GET('/g')
+        @google_mixin.require_auth
+        async def g(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'google')
+
+        @app.GET('/h')
+        @github_mixin.require_auth
+        async def h(ctx: HttpContext) -> None:
+            await ctx.send_str(200, 'github')
+
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
+            r = await client.get('/google/login', follow_redirects=False)
+            query = parse_qs(urlparse(r.headers['Location']).query)
+            google.nonce = query['nonce'][0]
+            r = await client.get('/google/callback', params={'code': 'c', 'state': query['state'][0]}, follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            r = await client.get('/g')
+            self.assertEqual('google', r.text)
+
+            r = await client.get('/github/login', follow_redirects=False)
+            query = parse_qs(urlparse(r.headers['Location']).query)
+            github.nonce = query['nonce'][0]
+            r = await client.get('/github/callback', params={'code': 'c', 'state': query['state'][0]}, follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+
+            r = await client.get('/g', follow_redirects=False)
+            self.assertEqual(302, r.status_code)
+            r = await client.get('/h')
+            self.assertEqual('github', r.text)
+
+    def test_default_strategy_is_single_identity(self) -> None:
+        provider = MockOIDCProvider()
+        config, http = self._provider_config(provider, 'google')
+        session = SessionMixin(InMemorySessionStore())
+
+        providers = OIDCProviders([config], session=session)
+        self.assertEqual('single-identity', providers.strategy)
+
+        mixin = OIDCMixin(config, session=session)
+        self.assertEqual('single-identity', mixin.strategy)
+
+    def test_invalid_strategy_rejected(self) -> None:
+        provider = MockOIDCProvider()
+        config, http = self._provider_config(provider, 'google')
+        session = SessionMixin(InMemorySessionStore())
+        with self.assertRaises(ValueError):
+            OIDCMixin(config, session=session, strategy='bogus')
+        with self.assertRaises(ValueError):
+            OIDCProviders([config], session=session, strategy='bogus')
+
+    def test_config_rejects_invalid_provider(self) -> None:
+        with self.assertRaises(ValueError):
+            OIDCConfig(
+                issuer='https://issuer.local',
+                client_id='client',
+                redirect_uri='http://localhost/callback',
+                provider='bad:name',
+            )
+
+    def test_providers_reject_duplicates(self) -> None:
+        config = OIDCConfig(
+            issuer='https://issuer.local',
+            client_id='client',
+            redirect_uri='http://localhost/callback',
+        )
+        session = SessionMixin(InMemorySessionStore())
+        with self.assertRaises(ValueError):
+            OIDCProviders([config, config], session=session)
+
+    def test_providers_require_at_least_one_config(self) -> None:
+        session = SessionMixin(InMemorySessionStore())
+        with self.assertRaises(ValueError):
+            OIDCProviders([], session=session)
+
+    def test_providers_reject_bad_base_path(self) -> None:
+        config = OIDCConfig(
+            issuer='https://issuer.local',
+            client_id='client',
+            redirect_uri='http://localhost/callback',
+        )
+        session = SessionMixin(InMemorySessionStore())
+        with self.assertRaises(ValueError):
+            OIDCProviders([config], session=session, base_path='auth')
+
+
 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
+- Every WebSocket connection gets one span for its whole lifetime, carrying
-  code is recorded as `kaya.websocket.close_code`; exceptions and configured
+  the applicable standard attributes: `url.scheme`, `url.path`, `url.query`,
-  error close codes mark the span as failed.
+  `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/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(
+        holder = self._start_span(f'WS {ws.path}', _websocket_attributes(ws), ws.headers)
-            f'WS {ws.path}',
-            {
-                'url.scheme': ws.scheme,
-                'url.path': ws.path,
-            },
-            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/src/kaya/rsgi/_rsgi.py

@@ -17,13 +17,21 @@ from typing import (
 from granian._granian import (  # type: ignore[attr-defined]
     RSGIHTTPProtocol,
     RSGIHTTPScope,
+    RSGIProtocolClosed,
     RSGIWebsocketProtocol,
     RSGIWebsocketScope,
     RSGIWebsocketTransport,
 )
 from pwo import Maybe
 
-from kaya.core import AbstractKayaApp, HttpContext, HttpMethod, WebSocket, WebSocketMessage
+from kaya.core import (
+    AbstractKayaApp,
+    HttpContext,
+    HttpMethod,
+    WebSocket,
+    WebSocketDisconnect,
+    WebSocketMessage,
+)
 from kaya.core._types import StrOrStrings
 
 
@@ -109,6 +117,7 @@ class RsgiContext(HttpContext):
 class RsgiWebSocket(WebSocket):
     _protocol: RSGIWebsocketProtocol
     _transport: Optional[RSGIWebsocketTransport]
+    _closed: bool
     scheme: str
     path: str
     query_string: str
@@ -124,6 +133,7 @@ class RsgiWebSocket(WebSocket):
             )
         self._protocol = protocol
         self._transport = None
+        self._closed = False
         self.scheme = scope.scheme
         self.path = scope.path
         self.query_string = scope.query_string
@@ -144,13 +154,28 @@ class RsgiWebSocket(WebSocket):
     async def accept(self, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
         # RSGI's websocket accept() takes no arguments (https://github.com/emmett-framework/granian/blob/master/docs/spec/RSGI.md),
         # so handshake headers (e.g. Set-Cookie) cannot be sent on RSGI; they are silently ignored here.
-        self._transport = await self._protocol.accept()
+        try:
+            self._transport = await self._protocol.accept()
+        except RSGIProtocolClosed as exc:
+            self._closed = True
+            raise WebSocketDisconnect(code=1006) from exc
 
     async def receive(self) -> WebSocketMessage:
         if self._transport is None:
             raise RuntimeError('WebSocket connection has not been accepted yet')
-        message = await self._transport.receive()
+        if self._closed:
+            return WebSocketMessage(kind='close')
+        try:
+            message = await self._transport.receive()
+        except RSGIProtocolClosed:
+            # Granian raises this when the peer is gone without a close
+            # handshake (browser tab closed, network dropped, ...). Surface it
+            # as a close message so receive loops terminate normally instead
+            # of seeing an exception.
+            self._closed = True
+            return WebSocketMessage(kind='close')
         if message.kind == 0:
+            self._closed = True
             return WebSocketMessage(kind='close')
         elif message.kind == 1:
             return WebSocketMessage(kind='binary', data=message.data)
@@ -162,15 +187,32 @@ class RsgiWebSocket(WebSocket):
     async def send_text(self, data: str) -> None:
         if self._transport is None:
             raise RuntimeError('WebSocket connection has not been accepted yet')
-        await self._transport.send_str(data)
+        if self._closed:
+            raise WebSocketDisconnect(code=1006)
+        try:
+            await self._transport.send_str(data)
+        except RSGIProtocolClosed as exc:
+            self._closed = True
+            raise WebSocketDisconnect(code=1006) from exc
 
     async def send_bytes(self, data: bytes) -> None:
         if self._transport is None:
             raise RuntimeError('WebSocket connection has not been accepted yet')
-        await self._transport.send_bytes(data)
+        if self._closed:
+            raise WebSocketDisconnect(code=1006)
+        try:
+            await self._transport.send_bytes(data)
+        except RSGIProtocolClosed as exc:
+            self._closed = True
+            raise WebSocketDisconnect(code=1006) from exc
 
     async def close(self, code: int = 1000) -> None:
-        self._protocol.close(code)
+        if self._closed:
+            return
+        try:
+            self._protocol.close(code)
+        except RSGIProtocolClosed:
+            self._closed = True
 
     async def __anext__(self) -> WebSocketMessage:
         message = await self.receive()

packages/kaya-rsgi/tests/test_rsgi.py

@@ -1,22 +1,150 @@
 import unittest
+from types import SimpleNamespace
+from typing import Any
+
+from granian.rsgi import ProtocolClosed
+from pwo import async_test
+
+from kaya.core import WebSocketDisconnect
 from kaya.rsgi import RsgiWebSocket
 
 
+class FakeScope:
+    scheme = 'ws'
+    path = '/ws'
+    query_string = ''
+    headers: dict[str, list[str]] = {}
+    client = '127.0.0.1:12345'
+    server = '127.0.0.1:80'
+
+
+class FakeTransport:
+    """A transport whose peer vanished without a close handshake."""
+
+    def __init__(self) -> None:
+        self.receive_calls = 0
+        self.sent: list[Any] = []
+
+    async def receive(self) -> Any:
+        self.receive_calls += 1
+        raise ProtocolClosed()
+
+    async def send_str(self, data: str) -> None:
+        self.sent.append(data)
+        raise ProtocolClosed()
+
+    async def send_bytes(self, data: bytes) -> None:
+        self.sent.append(data)
+        raise ProtocolClosed()
+
+
+class CleanCloseTransport(FakeTransport):
+    """A transport that reports a close message (kind 0)."""
+
+    async def receive(self) -> Any:
+        self.receive_calls += 1
+        return SimpleNamespace(kind=0, data=None)
+
+
+class FakeProtocol:
+    def __init__(self, transport: FakeTransport) -> None:
+        self._transport = transport
+        self.closed: list[int] = []
+
+    async def accept(self) -> FakeTransport:
+        return self._transport
+
+    def close(self, code: int) -> None:
+        self.closed.append(code)
+
+
+class ClosedOnAcceptProtocol(FakeProtocol):
+    """A protocol whose peer vanished during the handshake."""
+
+    async def accept(self) -> FakeTransport:
+        raise ProtocolClosed()
+
+
 class RsgiWebSocketTest(unittest.TestCase):
 
     def test_misconfigured_granian(self):
-        class FakeScope:
+        class FakeProtocolWithoutWebsockets:
-            scheme = 'ws'
-            path = '/ws'
-            query_string = ''
-            headers = {}
-            client = '127.0.0.1:12345'
-            server = '127.0.0.1:80'
-
-        class FakeProtocol:
             pass
 
         with self.assertRaises(RuntimeError) as ctx:
-            RsgiWebSocket(FakeScope(), FakeProtocol())  # type: ignore[arg-type]
+            RsgiWebSocket(FakeScope(), FakeProtocolWithoutWebsockets())  # type: ignore[arg-type]
 
         self.assertIn('Granian was not configured for websockets', str(ctx.exception))
+
+    async def _socket(self, transport: FakeTransport) -> tuple[RsgiWebSocket, FakeProtocol]:
+        protocol = FakeProtocol(transport)
+        ws = RsgiWebSocket(FakeScope(), protocol)  # type: ignore[arg-type]
+        await ws.accept()
+        return ws, protocol
+
+    @async_test
+    async def test_receive_returns_close_when_transport_is_gone(self):
+        ws, _ = await self._socket(FakeTransport())
+        message = await ws.receive()
+        self.assertEqual('close', message.kind)
+
+    @async_test
+    async def test_iteration_ends_when_transport_is_gone(self):
+        ws, _ = await self._socket(FakeTransport())
+        messages = [message async for message in ws]
+        self.assertEqual([], messages)
+
+    @async_test
+    async def test_receive_remembers_the_close(self):
+        transport = FakeTransport()
+        ws, _ = await self._socket(transport)
+        self.assertEqual('close', (await ws.receive()).kind)
+        self.assertEqual('close', (await ws.receive()).kind)
+        self.assertEqual(1, transport.receive_calls)
+
+    @async_test
+    async def test_send_text_raises_disconnect_when_transport_is_gone(self):
+        ws, _ = await self._socket(FakeTransport())
+        with self.assertRaises(WebSocketDisconnect) as ctx:
+            await ws.send_text('hello')
+        self.assertEqual(1006, ctx.exception.code)
+
+    @async_test
+    async def test_send_bytes_raises_disconnect_when_transport_is_gone(self):
+        ws, _ = await self._socket(FakeTransport())
+        with self.assertRaises(WebSocketDisconnect) as ctx:
+            await ws.send_bytes(b'hello')
+        self.assertEqual(1006, ctx.exception.code)
+
+    @async_test
+    async def test_sends_after_disconnect_do_not_touch_the_transport(self):
+        transport = FakeTransport()
+        ws, _ = await self._socket(transport)
+        with self.assertRaises(WebSocketDisconnect):
+            await ws.send_text('first')
+        with self.assertRaises(WebSocketDisconnect):
+            await ws.send_text('second')
+        self.assertEqual(['first'], transport.sent)
+
+    @async_test
+    async def test_send_after_a_clean_close_raises_disconnect(self):
+        transport = CleanCloseTransport()
+        ws, _ = await self._socket(transport)
+        self.assertEqual('close', (await ws.receive()).kind)
+        with self.assertRaises(WebSocketDisconnect):
+            await ws.send_text('hello')
+        self.assertEqual([], transport.sent)
+
+    @async_test
+    async def test_accept_raises_disconnect_when_transport_is_gone(self):
+        ws = RsgiWebSocket(FakeScope(), ClosedOnAcceptProtocol(FakeTransport()))  # type: ignore[arg-type]
+        with self.assertRaises(WebSocketDisconnect) as ctx:
+            await ws.accept()
+        self.assertEqual(1006, ctx.exception.code)
+
+    @async_test
+    async def test_close_is_noop_once_the_transport_is_gone(self):
+        ws, protocol = await self._socket(FakeTransport())
+        await ws.receive()
+        await ws.close(1000)
+        self.assertEqual([], protocol.closed)