diff --git a/bluesky_httpserver/_authentication.py b/bluesky_httpserver/_authentication.py
index c1144f5..637a8ad 100644
--- a/bluesky_httpserver/_authentication.py
+++ b/bluesky_httpserver/_authentication.py
@@ -1,15 +1,30 @@
 import asyncio
 import hashlib
+import logging
 import secrets
 import uuid as uuid_module
 import warnings
 from datetime import datetime, timedelta
 from typing import Optional
 
-from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, Response, Security, WebSocket
+from fastapi import (
+    APIRouter,
+    Depends,
+    Form,
+    HTTPException,
+    Query,
+    Request,
+    Response,
+    Security,
+    WebSocket,
+)
 from fastapi.openapi.models import APIKey, APIKeyIn
 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
-from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm, SecurityScopes
+from fastapi.security import (
+    OAuth2PasswordBearer,
+    OAuth2PasswordRequestForm,
+    SecurityScopes,
+)
 from fastapi.security.api_key import APIKeyBase, APIKeyCookie, APIKeyQuery
 from fastapi.security.utils import get_authorization_scheme_param
 from sqlalchemy.exc import IntegrityError
@@ -36,6 +51,7 @@
 from .database import orm
 from .database.core import (
     create_user,
+    get_or_create_principal,
     latest_principal_activity,
     lookup_valid_api_key,
     lookup_valid_pending_session_by_device_code,
@@ -60,6 +76,8 @@
 DEVICE_CODE_MAX_AGE = timedelta(minutes=10)
 DEVICE_CODE_POLLING_INTERVAL = 5  # seconds
 
+logger = logging.getLogger(__name__)
+
 
 def utcnow():
     "UTC now with second resolution"
@@ -140,7 +158,19 @@ def create_refresh_token(session_id, secret_key, expires_delta):
     return encoded_jwt
 
 
-def decode_token(token, secret_keys):
+def decode_token(token, secret_keys, proxied_authenticator=None):
+    """Decode a JWT.
+
+    Tries every bluesky-httpserver HMAC ``secret_key`` first (supports key
+    rotation).  If none succeed and a ``proxied_authenticator`` (typically a
+    :class:`bluesky_httpserver.authenticators.ProxiedOIDCAuthenticator`) is
+    provided, delegate to its ``decode_token`` so that access tokens minted
+    by an external OIDC provider (e.g. via device-code flow) are accepted.
+
+    Raises :class:`fastapi.HTTPException` with status 401 if the token cannot
+    be validated by either mechanism.  Propagates :class:`ExpiredSignatureError`
+    so the caller can surface a distinct "expired token" error.
+    """
     credentials_exception = HTTPException(
         status_code=401,
         detail="Could not validate credentials",
@@ -149,19 +179,53 @@ def decode_token(token, secret_keys):
     # The first key in settings.secret_keys is used for *encoding*.
     # All keys are tried for *decoding* until one works or they all
     # fail. They supports key rotation.
+    last_error = None
     for secret_key in secret_keys:
         try:
-            payload = jwt.decode(token, secret_key, algorithms=[ALGORITHM])
-            break
+            return jwt.decode(token, secret_key, algorithms=[ALGORITHM])
         except ExpiredSignatureError:
             # Do not let this be caught below with the other JWTError types.
             raise
-        except JWTError:
+        except JWTError as err:
+            last_error = err
             # Try the next key in the key rotation.
             continue
-    else:
-        raise credentials_exception
-    return payload
+    # None of the bluesky-httpserver HMAC keys accepted the token.
+    # Fall back to the proxied OIDC authenticator (if any) so that JWTs minted
+    # by an upstream provider — e.g. via device-code flow — are honored.
+    if proxied_authenticator is not None:
+        try:
+            return proxied_authenticator.decode_token(token)
+        except ExpiredSignatureError:
+            raise
+        except JWTError:
+            pass
+    raise credentials_exception from last_error
+
+
+def _extract_scopes(decoded_access_token, authenticator=None):
+    """Normalize the ``scopes`` set from an OIDC-decoded JWT.
+
+    OIDC providers spell the scope claim inconsistently:
+
+    * ``scp`` may be a space-separated string (Microsoft Entra) or a list.
+    * ``scope`` is typically a space-separated string.
+
+    Additionally, an authenticator with a ``scopes_map`` attribute (e.g.
+    :class:`EntraAuthenticator`) may already have translated raw provider
+    scopes into local scopes and stored the result in ``claims["scope"]``.
+    In that case, prefer the pre-translated value.
+    """
+    scopes: set[str] = set()
+    scp = decoded_access_token.get("scp")
+    if isinstance(scp, list):
+        scopes.update(scp)
+    elif isinstance(scp, str) and scp:
+        scopes.update(scp.split())
+    scope = decoded_access_token.get("scope")
+    if isinstance(scope, str) and scope:
+        scopes.update(scope.split())
+    return scopes
 
 
 async def get_api_key(
@@ -175,6 +239,52 @@ async def get_api_key(
     return None
 
 
+def get_session_state(
+    access_token: str = Depends(oauth2_scheme),
+    settings: BaseSettings = Depends(get_settings),
+    authenticators=Depends(get_authenticators),
+):
+    """FastAPI dependency: return the ``state`` dict embedded in the current
+    access token, if any, else ``{}``.
+
+    This mirrors tiled's ``get_session_state`` and lets downstream routes
+    retrieve authenticator-supplied state (e.g. upstream OIDC tokens for OBO
+    exchange) without a database round-trip.
+    """
+    if not access_token:
+        return {}
+    proxied_auth = _find_proxied_authenticator(authenticators)
+    try:
+        payload = decode_token(access_token, settings.secret_keys, proxied_auth)
+    except HTTPException:
+        return {}
+    except ExpiredSignatureError:
+        return {}
+    return payload.get("state") or {}
+
+
+def _find_proxied_authenticator(authenticators):
+    """Return the first authenticator that is a ``ProxiedOIDCAuthenticator``.
+
+    Imported lazily to avoid a top-level circular import between
+    ``_authentication.py`` and ``authenticators.py``.
+    """
+    if not authenticators:
+        return None
+    try:
+        from .authenticators import ProxiedOIDCAuthenticator
+    except ImportError:
+        return None
+    if isinstance(authenticators, dict):
+        candidates = authenticators.values()
+    else:
+        candidates = authenticators
+    for a in candidates:
+        if isinstance(a, ProxiedOIDCAuthenticator):
+            return a
+    return None
+
+
 def get_current_principal(
     request: Request,
     security_scopes: SecurityScopes,
@@ -279,30 +389,92 @@ def get_current_principal(
         if ("api_key" in request.query_params) and (request.cookies.get(API_KEY_COOKIE_NAME) != api_key):
             request.state.cookies_to_set.append({"key": API_KEY_COOKIE_NAME, "value": api_key})
     elif access_token is not None:
+        proxied_authenticator = _find_proxied_authenticator(authenticators)
         try:
-            payload = decode_token(access_token, settings.secret_keys)
+            payload = decode_token(
+                access_token, settings.secret_keys, proxied_authenticator
+            )
         except ExpiredSignatureError:
             raise HTTPException(
                 status_code=401,
                 detail="Access token has expired. Refresh token.",
                 headers=headers_for_401,
             )
-        principal = schemas.Principal(
-            uuid=uuid_module.UUID(hex=payload["sub"]),
-            type=payload["sub_typ"],
-            identities=[
-                schemas.Identity(id=identity["id"], provider=identity["idp"]) for identity in payload["ids"]
-            ],
-        )
 
-        # scopes = payload["scp"]
+        if "sub_typ" in payload:
+            # Token minted by bluesky-httpserver (login / refresh / device-code
+            # flow).  ``payload`` carries everything needed to reconstruct the
+            # Principal without a database hit.
+            principal = schemas.Principal(
+                uuid=uuid_module.UUID(hex=payload["sub"]),
+                type=payload["sub_typ"],
+                identities=[
+                    schemas.Identity(id=identity["id"], provider=identity["idp"])
+                    for identity in payload["ids"]
+                ],
+            )
+
+            # scopes = payload["scp"]
 
-        # Combine scopes for all identities (it is expected to be only one identity).
-        ids = [_["id"] for _ in payload["ids"] if _["idp"] in settings.authentication_provider_names]
-        scopes = set.union(*[api_access_manager.get_user_scopes(_) for _ in ids])
+            # Combine scopes for all identities (it is expected to be only one identity).
+            ids = [
+                _["id"]
+                for _ in payload["ids"]
+                if _["idp"] in settings.authentication_provider_names
+            ]
+            scopes_sets = [api_access_manager.get_user_scopes(_) for _ in ids]
+            scopes = set.union(*scopes_sets) if scopes_sets else set()
 
-        roles_sets = [api_access_manager.get_user_roles(_) for _ in ids]
-        roles = set.union(*roles_sets) if roles_sets else set()
+            roles_sets = [api_access_manager.get_user_roles(_) for _ in ids]
+            roles = set.union(*roles_sets) if roles_sets else set()
+
+        else:
+            # Token minted by an external OIDC provider (proxied OIDC flow).
+            # Resolve the Principal via the auth DB, using the app-scoped
+            # provider name that was set at startup.
+            if proxied_authenticator is None:
+                raise HTTPException(
+                    status_code=401,
+                    detail="Access token is not recognized",
+                    headers=headers_for_401,
+                )
+            identity_id = payload.get("user") or payload.get("sub")
+            if identity_id is None:
+                raise HTTPException(
+                    status_code=401,
+                    detail="Access token missing subject",
+                    headers=headers_for_401,
+                )
+            provider = getattr(request.app.state, "provider", None)
+            if provider is None:
+                # Fall back to the first configured provider name.  In practice
+                # ``app.state.provider`` is set by build_app() whenever a
+                # ProxiedOIDCAuthenticator is configured, but we do not want
+                # to fail if someone bootstraps the app differently.
+                provider = (
+                    settings.authentication_provider_names[0]
+                    if settings.authentication_provider_names
+                    else _DEFAULT_ANONYMOUS_PROVIDER_NAME
+                )
+            with get_sessionmaker(settings.database_settings)() as db:
+                principal_orm = get_or_create_principal(db, provider, identity_id)
+                principal = schemas.Principal(
+                    uuid=principal_orm.uuid,
+                    type=schemas.PrincipalType.user,
+                    identities=[schemas.Identity(id=identity_id, provider=provider)],
+                    access_token=access_token,
+                )
+            # Combine scopes carried in the token itself with any additional
+            # scopes granted to this user by the api_access_manager (which is
+            # the fork's replacement for tiled's DB-role machinery).
+            token_scopes = _extract_scopes(payload, proxied_authenticator)
+            if api_access_manager.is_user_known(identity_id):
+                extra_scopes = api_access_manager.get_user_scopes(identity_id)
+                roles = api_access_manager.get_user_roles(identity_id)
+            else:
+                extra_scopes = set()
+                roles = set()
+            scopes = set(token_scopes) | set(extra_scopes)
 
     else:
         # No form of authentication is present.
@@ -373,12 +545,24 @@ def get_current_principal_websocket(
 
     auth_header = websocket.headers.get("Authorization", "")
     access_token, api_key = None, None
-    # Currently we do not support authentication with tokens
-    # if auth_header.startswith("Bearer "):
-    #     access_token = auth_header[len("Bearer") :].strip()
-    if auth_header.startswith("ApiKey "):
+    if auth_header.startswith("Bearer "):
+        access_token = auth_header[len("Bearer") :].strip()
+    elif auth_header.startswith("ApiKey "):
         api_key = auth_header[len("ApiKey") :].strip()
 
+    # Also honor an ``access_token`` query parameter so browsers that cannot
+    # set Authorization headers on a WebSocket handshake still authenticate.
+    if access_token is None and api_key is None:
+        access_token = websocket.query_params.get("access_token")
+        if access_token is None:
+            api_key = websocket.query_params.get("api_key")
+
+    # If nothing was supplied on the initial handshake, return None instead of
+    # raising 401.  The socket route can then attempt the first-message
+    # protocol handled by ``authenticate_websocket_first_message``.
+    if not access_token and not api_key:
+        return None
+
     principal = None
     try:
         principal = get_current_principal(
@@ -391,12 +575,58 @@ def get_current_principal_websocket(
             api_access_manager=api_access_manager,
         )
     except HTTPException as ex:
-        print(f"WebSocket connection failed: {ex}")
+        logger.info("WebSocket authentication failed: %s", ex.detail)
 
     return principal
 
 
-def create_session(settings, identity_provider, id, scopes):
+def authenticate_websocket_first_message(websocket, message):
+    """Handle a ``{"type": "auth", ...}`` handshake message on a WebSocket.
+
+    The socket route awaits this only when the standard header/query
+    handshake produced no principal (i.e. ``get_current_principal_websocket``
+    returned ``None``).  It accepts either an API key or an access token in
+    the message body:
+
+        {"type": "auth", "api_key": "<hex-secret>"}
+        {"type": "auth", "access_token": "<jwt>"}
+
+    Returns the resolved :class:`schemas.Principal`, or ``None`` if the
+    message is malformed or the credentials are invalid.  The socket route
+    is expected to close the connection on failure.
+    """
+    if not isinstance(message, dict):
+        return None
+    if message.get("type") != "auth":
+        return None
+
+    app = websocket.app
+    settings = app.dependency_overrides[get_settings]()
+    authenticators = app.dependency_overrides[get_authenticators]()
+    api_access_manager = app.dependency_overrides[get_api_access_manager]()
+
+    api_key = message.get("api_key")
+    access_token = message.get("access_token")
+    if not api_key and not access_token:
+        return None
+
+    security_scopes = SecurityScopes(scopes=[])
+    try:
+        return get_current_principal(
+            request=websocket,
+            security_scopes=security_scopes,
+            access_token=access_token,
+            api_key=api_key,
+            settings=settings,
+            authenticators=authenticators,
+            api_access_manager=api_access_manager,
+        )
+    except HTTPException as ex:
+        logger.info("WebSocket first-message authentication failed: %s", ex.detail)
+        return None
+
+
+def create_session(settings, identity_provider, id, scopes, state=None):
     with get_sessionmaker(settings.database_settings)() as db:
         # Have we seen this Identity before?
         identity = (
@@ -420,6 +650,7 @@ def create_session(settings, identity_provider, id, scopes):
         session = orm.Session(
             principal_id=principal.id,
             expiration_time=utcnow() + settings.session_max_age,
+            state=state or {},
         )
         db.add(session)
         db.commit()
@@ -431,7 +662,11 @@ def create_session(settings, identity_provider, id, scopes):
             "sub": principal.uuid.hex,
             "sub_typ": principal.type.value,
             "scp": list(scopes),
-            "ids": [{"id": identity.id, "idp": identity.provider} for identity in principal.identities],
+            "ids": [
+                {"id": identity.id, "idp": identity.provider}
+                for identity in principal.identities
+            ],
+            "state": session.state or {},
         }
         access_token = create_access_token(
             data=data,
@@ -463,6 +698,7 @@ async def auth_code(
         request.state.endpoint = "auth"
         user_session_state = await authenticator.authenticate(request)
         username = user_session_state.user_name if user_session_state else None
+        session_state = (user_session_state.state or {}) if user_session_state else {}
 
         if username and api_access_manager.is_user_known(username):
             scopes = api_access_manager.get_user_scopes(username)
@@ -470,7 +706,7 @@ async def auth_code(
             raise HTTPException(status_code=401, detail="Authentication failure")
 
         tokens = await asyncio.get_running_loop().run_in_executor(
-            None, create_session, settings, provider, username, scopes
+            None, create_session, settings, provider, username, scopes, session_state
         )
         # Show only the refresh_token, which is what the user should
         # paste into a terminal-based client.
@@ -495,6 +731,7 @@ async def handle_credentials(
             username=form_data.username, password=form_data.password
         )
         username = user_session_state.user_name if user_session_state else None
+        session_state = (user_session_state.state or {}) if user_session_state else {}
 
         err_msg = None
         if not username:
@@ -511,7 +748,7 @@ async def handle_credentials(
                 headers={"WWW-Authenticate": "Bearer"},
             )
         return await asyncio.get_running_loop().run_in_executor(
-            None, create_session, settings, provider, username, scopes
+            None, create_session, settings, provider, username, scopes, session_state
         )
 
     return handle_credentials
@@ -558,10 +795,17 @@ async def authorize_redirect(
         """Redirect browser to OAuth provider for authentication."""
         redirect_uri = f"{get_base_url(request)}/auth/provider/{provider}/code"
 
+        # Always request ``openid`` and ``offline_access`` so the IdP returns a
+        # refresh_token in the code exchange.  Authenticators (e.g. Entra) may
+        # advertise extra scopes via an ``extra_scopes`` attribute to obtain
+        # per-resource access tokens.  
+        scopes = {"openid", "profile", "email", "offline_access"}
+        scopes.update(getattr(authenticator, "extra_scopes", []) or [])
+
         params = {
             "client_id": authenticator.client_id,
             "response_type": "code",
-            "scope": "openid profile email",
+            "scope": " ".join(sorted(scopes)),
             "redirect_uri": redirect_uri,
         }
         if state:
@@ -591,11 +835,13 @@ async def device_code_authorize(
             pending_session = create_pending_session(db)
 
         verification_uri = f"{get_base_url(request)}/auth/provider/{provider}/token"
+        scopes = {"openid", "profile", "email", "offline_access"}
+        scopes.update(getattr(authenticator, "extra_scopes", []) or [])
         authorization_uri = authenticator.authorization_endpoint.copy_with(
             params={
                 "client_id": authenticator.client_id,
                 "response_type": "code",
-                "scope": "openid profile email",
+                "scope": " ".join(sorted(scopes)),
                 "redirect_uri": f"{get_base_url(request)}/auth/provider/{provider}/device_code",
                 "state": pending_session["user_code"].replace("-", ""),
             }
@@ -648,7 +894,7 @@ async def _complete_device_code_authorization(
     <div class="error">
       Invalid user code. It may have been mistyped, or the pending request may have expired.
     </div>
-    <br/><a href="{action.rsplit('?', 1)[0]}?code={code}">Try again</a>
+    <br/><a href="{action.rsplit("?", 1)[0]}?code={code}">Try again</a>
 </body>
 </html>
 """
@@ -684,6 +930,7 @@ async def _complete_device_code_authorization(
             return HTMLResponse(content=error_html, status_code=401)
 
         username = user_session_state.user_name
+        session_state = user_session_state.state or {}
         if not api_access_manager.is_user_known(username):
             error_html = f"""
 <!DOCTYPE html>
@@ -710,7 +957,7 @@ async def _complete_device_code_authorization(
 
         # Create the session
         session = await asyncio.get_running_loop().run_in_executor(
-            None, _create_session_orm, settings, provider, username, db
+            None, _create_session_orm, settings, provider, username, db, session_state
         )
 
         # Link the pending session to the real session
@@ -837,7 +1084,7 @@ async def device_code_submit(
     return device_code_submit
 
 
-def _create_session_orm(settings, identity_provider, id, db):
+def _create_session_orm(settings, identity_provider, id, db, state=None):
     """
     Create a session and return the ORM object (for device code flow).
 
@@ -864,6 +1111,7 @@ def _create_session_orm(settings, identity_provider, id, db):
     session = orm.Session(
         principal_id=principal.id,
         expiration_time=utcnow() + settings.session_max_age,
+        state=state or {},
     )
     db.add(session)
     db.commit()
@@ -924,7 +1172,11 @@ async def device_code_token(
                 "sub": principal.uuid.hex,
                 "sub_typ": principal.type.value,
                 "scp": list(scopes),
-                "ids": [{"id": ident.id, "idp": ident.provider} for ident in principal.identities],
+                "ids": [
+                    {"id": ident.id, "idp": ident.provider}
+                    for ident in principal.identities
+                ],
+                "state": session.state or {},
             }
             access_token = create_access_token(
                 data=data,
@@ -1152,7 +1404,11 @@ def slide_session(refresh_token, settings, db, api_access_manager):
         "sub": principal.uuid.hex,
         "sub_typ": principal.type.value,
         "scp": list(scopes),
-        "ids": [{"id": identity.id, "idp": identity.provider} for identity in principal.identities],
+        "ids": [
+            {"id": identity.id, "idp": identity.provider}
+            for identity in principal.identities
+        ],
+        "state": session.state or {},
     }
     access_token = create_access_token(
         data=data,
diff --git a/bluesky_httpserver/app.py b/bluesky_httpserver/app.py
index 0d96667..7821107 100644
--- a/bluesky_httpserver/app.py
+++ b/bluesky_httpserver/app.py
@@ -16,7 +16,12 @@
 from fastapi.openapi.utils import get_openapi
 
 from .authentication import ExternalAuthenticator, InternalAuthenticator
-from .console_output import CollectPublishedConsoleOutput, ConsoleOutputStream, SystemInfoStream
+from .authenticators import ProxiedOIDCAuthenticator
+from .console_output import (
+    CollectPublishedConsoleOutput,
+    ConsoleOutputStream,
+    SystemInfoStream,
+)
 from .core import PatchedStreamingResponse
 from .database.core import purge_expired
 from .resources import SERVER_RESOURCES as SR
@@ -217,6 +222,24 @@ def build_app(authentication=None, api_access=None, resource_access=None, server
                 authentication_router.post(f"/provider/{provider}/token")(
                     build_device_code_token_route(authenticator, provider)
                 )
+                # Warn if the operator forgot to configure a redirect target
+                # for successful browser-based logins. Without it the user
+                # will get a page of raw JSON instead of being sent to the UI.
+                if not getattr(authenticator, "redirect_on_success", None):
+                    logger.warning(
+                        "External authenticator %r has no 'redirect_on_success' "
+                        "configured. Browser-based login will return raw JSON "
+                        "tokens instead of redirecting to a UI landing page. "
+                        "Set 'redirect_on_success' in the authenticator "
+                        "configuration to a UI callback URL to silence this "
+                        "warning.",
+                        provider,
+                    )
+                # Expose the OIDC provider name on app.state so that
+                # get_current_principal / websocket handshakes can resolve
+                # externally-minted JWTs to a Principal in the auth DB.
+                if isinstance(authenticator, ProxiedOIDCAuthenticator):
+                    app.state.provider = provider
             else:
                 raise ValueError(f"unknown authenticator type {type(authenticator)}")
             for custom_router in getattr(authenticator, "include_routers", []):
@@ -263,8 +286,10 @@ async def startup_event():
             from .database.core import (  # make_admin_by_identity,
                 REQUIRED_REVISION,
                 UninitializedDatabase,
+                DatabaseUpgradeNeeded,
                 check_database,
                 initialize_database,
+                upgrade,
             )
 
             connect_args = {}
@@ -282,6 +307,12 @@ async def startup_event():
                 )
                 initialize_database(engine)
                 logger.info("Database initialized.")
+            except DatabaseUpgradeNeeded:
+                logger.info(
+                    f"Database at {redacted_url} is out of date. Upgrading to {REQUIRED_REVISION}..."
+                )
+                upgrade(engine, REQUIRED_REVISION)
+                logger.info("Database upgraded.")
             else:
                 logger.info(f"Connected to existing database at {redacted_url}.")
             # SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@@ -416,10 +447,30 @@ async def purge_expired_sessions_and_api_keys():
 
     @app.on_event("shutdown")
     async def shutdown_event():
-        await SR.RM.close()
-        await SR.console_output_loader.stop()
-        await SR.console_output_stream.stop()
-        await SR.system_info_stream.stop()
+        """Safely shutdown and perform the cleanup robustly
+
+        This change ensures that the application shuts down and cleans up resources even if there is
+        a problem, without silencing the errors.
+        """
+        for task in getattr(app.state, "tasks", []):
+            task.cancel()
+        for closer_name in (
+            "console_output_loader",
+            "console_output_stream",
+            "system_info_stream",
+        ):
+            closer = getattr(SR, closer_name, None)
+            if closer is not None:
+                try:
+                    await closer.stop()
+                except Exception:
+                    logger.exception("Error stopping %s", closer_name)
+        rm = getattr(SR, "RM", None)
+        if rm is not None:
+            try:
+                await rm.close()
+            except Exception:
+                logger.exception("Error closing REManagerAPI connection")
 
     @lru_cache(1)
     def override_get_authenticators():
diff --git a/bluesky_httpserver/authentication/__init__.py b/bluesky_httpserver/authentication/__init__.py
index 85d835e..13043c5 100644
--- a/bluesky_httpserver/authentication/__init__.py
+++ b/bluesky_httpserver/authentication/__init__.py
@@ -1,4 +1,5 @@
 from .._authentication import (
+    authenticate_websocket_first_message,
     base_authentication_router,
     build_auth_code_route,
     build_authorize_route,
@@ -9,6 +10,7 @@
     build_handle_credentials_route,
     get_current_principal,
     get_current_principal_websocket,
+    get_session_state,
     oauth2_scheme,
 )
 from .authenticator_base import (
@@ -21,8 +23,10 @@
     "ExternalAuthenticator",
     "InternalAuthenticator",
     "UserSessionState",
+    "authenticate_websocket_first_message",
     "get_current_principal",
     "get_current_principal_websocket",
+    "get_session_state",
     "base_authentication_router",
     "build_auth_code_route",
     "build_authorize_route",
diff --git a/bluesky_httpserver/authenticators.py b/bluesky_httpserver/authenticators.py
index a58fedf..34b0d76 100644
--- a/bluesky_httpserver/authenticators.py
+++ b/bluesky_httpserver/authenticators.py
@@ -4,9 +4,10 @@
 import logging
 import re
 import secrets
+import uuid
 from collections.abc import Iterable
 from datetime import timedelta
-from typing import Any, List, Mapping, Optional, cast
+from typing import Any, Dict, List, Mapping, Optional, cast
 
 import httpx
 from cachetools import TTLCache, cached
@@ -63,15 +64,19 @@ class DictionaryAuthenticator(InternalAuthenticator):
     description: May be displayed by client after successful login.
 """
 
-    def __init__(self, users_to_passwords: Mapping[str, str], confirmation_message: str = ""):
+    def __init__(
+        self, users_to_passwords: Mapping[str, str], confirmation_message: str = ""
+    ):
         self._users_to_passwords = users_to_passwords
         self.confirmation_message = confirmation_message
 
-    async def authenticate(self, username: str, password: str) -> Optional[UserSessionState]:
+    async def authenticate(
+        self, username: str, password: str
+    ) -> Optional[UserSessionState]:
         true_password = self._users_to_passwords.get(username)
         if not true_password:
             # Username is not valid.
-            return None
+            return
         if secrets.compare_digest(true_password, password):
             return UserSessionState(username, {})
 
@@ -92,12 +97,16 @@ class PAMAuthenticator(InternalAuthenticator):
 
     def __init__(self, service: str = "login", confirmation_message: str = ""):
         if not modules_available("pamela"):
-            raise ModuleNotFoundError("This PAMAuthenticator requires the module 'pamela' to be installed.")
+            raise ModuleNotFoundError(
+                "This PAMAuthenticator requires the module 'pamela' to be installed."
+            )
         self.service = service
         self.confirmation_message = confirmation_message
         # TODO Try to open a PAM session.
 
-    async def authenticate(self, username: str, password: str) -> Optional[UserSessionState]:
+    async def authenticate(
+        self, username: str, password: str
+    ) -> Optional[UserSessionState]:
         import pamela
 
         try:
@@ -105,7 +114,7 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi
             return UserSessionState(username, {})
         except pamela.PAMError:
             # Authentication failed.
-            return None
+            return
 
 
 class OIDCAuthenticator(ExternalAuthenticator):
@@ -179,33 +188,42 @@ def token_endpoint(self) -> str:
 
     @functools.cached_property
     def authorization_endpoint(self) -> httpx.URL:
-        return httpx.URL(cast(str, self._config_from_oidc_url.get("authorization_endpoint")))
+        return httpx.URL(
+            cast(str, self._config_from_oidc_url.get("authorization_endpoint"))
+        )
 
     @functools.cached_property
     def device_authorization_endpoint(self) -> str:
-        return cast(str, self._config_from_oidc_url.get("device_authorization_endpoint"))
+        return cast(
+            str, self._config_from_oidc_url.get("device_authorization_endpoint")
+        )
 
     @functools.cached_property
     def end_session_endpoint(self) -> str:
         return cast(str, self._config_from_oidc_url.get("end_session_endpoint"))
 
-    @cached(TTLCache(maxsize=1, ttl=timedelta(days=7).total_seconds()))
+    @cached(TTLCache(maxsize=1, ttl=timedelta(hours=1).total_seconds()))
     def keys(self) -> List[str]:
         return httpx.get(self.jwks_uri).raise_for_status().json().get("keys", [])
 
-    def decode_token(self, token: str) -> dict[str, Any]:
+    def decode_token(
+        self, id_token: str, access_token: Optional[str] = None
+    ) -> dict[str, Any]:
         return jwt.decode(
-            token,
+            id_token,
             key=self.keys(),
             algorithms=self.id_token_signing_alg_values_supported,
             audience=self._audience,
             issuer=self.issuer,
+            access_token=access_token,
         )
 
     async def authenticate(self, request: Request) -> Optional[UserSessionState]:
         code = request.query_params.get("code")
         if not code:
-            logger.warning("Authentication failed: No authorization code parameter provided.")
+            logger.warning(
+                "Authentication failed: No authorization code parameter provided."
+            )
             return None
         # A proxy in the middle may make the request into something like
         # 'http://localhost:8000/...' so we fix the first part but keep
@@ -223,37 +241,16 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]:
             logger.error("Authentication error: %r", response_body)
             return None
         id_token = response_body["id_token"]
-        # NOTE: We decode the id_token, not access_token, because:
-        # 1. The id_token is the OIDC identity assertion meant for the client
-        # 2. Some providers (like Microsoft Entra) return opaque access_tokens
-        #    that cannot be decoded with the JWKS keys when the resource is
-        #    a first-party Microsoft API (e.g., Graph API with User.Read scope)
+        access_token = response_body["access_token"]
         try:
-            verified_body = self.decode_token(id_token)
+            verified_body = self.decode_token(id_token, access_token)
         except JWTError:
             logger.exception(
                 "Authentication error. Unverified token: %r",
                 jwt.get_unverified_claims(id_token),
             )
             return None
-        # Use preferred_username as the user identifier, extracting just the username
-        # part if it's in email format (user@domain.com -> user)
-        preferred_username = verified_body.get("preferred_username")
-        if preferred_username and "@" in preferred_username:
-            user_id = preferred_username.split("@")[0]
-        elif preferred_username:
-            user_id = preferred_username
-        else:
-            user_id = verified_body["sub"]
-        logger.info(
-            "OIDC authentication successful. user_id=%r (sub=%r, preferred_username=%r, email=%r, name=%r)",
-            user_id,
-            verified_body.get("sub"),
-            verified_body.get("preferred_username"),
-            verified_body.get("email"),
-            verified_body.get("name"),
-        )
-        return UserSessionState(user_id, {})
+        return UserSessionState(verified_body["sub"], {})
 
 
 class ProxiedOIDCAuthenticator(OIDCAuthenticator):
@@ -310,18 +307,227 @@ def oauth2_schema(self) -> OAuth2:
         return self._oidc_bearer
 
 
+class EntraAuthenticator(ProxiedOIDCAuthenticator):
+    def __init__(
+        self,
+        audience: str,
+        client_id: str,
+        well_known_uri: str,
+        device_flow_client_id: str,
+        extra_scopes: Optional[List[str]] = None,
+        confirmation_message: str = "",
+        scopes_map: Optional[Dict[str, list[str]]] = None,
+        client_secret: str = "",
+        redirect_on_success: Optional[str] = None,
+    ):
+        self.scopes_map = scopes_map if scopes_map is not None else {}
+        self.extra_scopes = extra_scopes or []
+        super().__init__(
+            audience,
+            client_id,
+            well_known_uri,
+            device_flow_client_id,
+            scopes=None,  # not used by Entra; enforcement is via scopes_map
+            confirmation_message=confirmation_message,
+        )
+        # Override the empty secret from ProxiedOIDCAuthenticator if provided.
+        if client_secret:
+            self._client_secret = Secret(client_secret)
+        self.redirect_on_success = redirect_on_success
+
+    @property
+    def scopes(self):
+        mapped = set()
+        for tiled_scopes in self.scopes_map.values():
+            mapped.update(tiled_scopes)
+        return list(mapped)
+
+    @scopes.setter
+    def scopes(self, value):
+        pass  # ignored; scopes are derived from scopes_map
+
+    def decode_token(
+        self, id_token: str, access_token: Optional[str] = None
+    ) -> dict[str, Any]:
+        claims = super().decode_token(id_token, access_token)
+
+        # sub generated by Entra is an opaque string; generate a stable UUID
+        # for Tiled based on "iss|sub" for uniqueness across tenants.
+        # Preserve the original Entra sub separately so it can be used as a
+        # fallback display name — it is more human-readable than the UUID5 hex.
+        original_sub = claims.get("sub")
+        issuer = claims.get("iss", "")
+        claims["sub"] = uuid.uuid5(uuid.NAMESPACE_URL, f"{issuer}|{original_sub}").hex
+        claims["entra_sub"] = original_sub
+
+        # Derive a human-readable username from the token claims.
+        # Priority: nameID (explicit app config) → preferred_username (v2 tokens)
+        # → upn (v1 tokens) → email → original Entra sub (opaque but stable and
+        # meaningful, unlike the UUID5 hex stored in claims["sub"]).
+        #
+        # Note: preferred_username / upn are often absent from *access* tokens
+        # unless explicitly added as optional claims in the Entra app registration.
+        # They are typically present in id_tokens.  If none are found, the
+        # original_sub is used and a warning is emitted so operators know to add
+        # the optional claim.
+        claims["entra_username"] = (
+            claims.get("nameID")
+            or claims.get("preferred_username")
+            or claims.get("upn")
+            or claims.get("email")
+        )
+
+        if user := claims.get("entra_username"):
+            user = user.strip()
+            if "\\" in user:
+                user = user.rsplit("\\", 1)[-1]
+            elif "@" in user:
+                user = user.split("@", 1)[0]
+        else:
+            # No human-readable claim was found.  Fall back to the original
+            # Entra sub (opaque but at least stable and not a UUID5 hex).
+            # This produces a workable identity but authz policies that match
+            # on username will need to use the Entra sub value.
+            user = original_sub
+            logger.warning(
+                "EntraAuthenticator: no human-readable username claim found in token "
+                "(checked nameID, preferred_username, upn, email). "
+                "Falling back to Entra sub=%r. "
+                "To fix: add 'preferred_username' as an optional claim in the "
+                "Entra app registration → Token configuration → Optional claims → Access token.",
+                original_sub,
+            )
+        claims["user"] = user
+
+        # Translate Entra scopes to tiled scopes.
+        # The "scp" claim is present in access tokens but may be absent from
+        # id_tokens (e.g. during the authorization code flow).  When absent,
+        # assume all mapped scopes were granted (Entra would not have issued
+        # the tokens if the user lacked the requested scopes).
+        scp_raw = claims.get("scp", "")
+        tiled_scope_set = set()
+        if scp_raw:
+            for scope in scp_raw.split(" "):
+                mapped_scopes = self.scopes_map.get(scope)
+                if mapped_scopes is None:
+                    logger.warning("Unmapped Entra scope in 'scp': %s", scope)
+                    continue
+                tiled_scope_set.update(mapped_scopes)
+        else:
+            # No scp claim — grant all tiled scopes from the map.
+            for mapped_scopes in self.scopes_map.values():
+                tiled_scope_set.update(mapped_scopes)
+        claims["scope"] = " ".join(tiled_scope_set)
+
+        return claims
+
+    async def authenticate(self, request: Request) -> Optional[UserSessionState]:
+        """Complete the Entra OIDC authorization-code flow and return a session.
+
+        After a successful code exchange the Entra ``access_token`` and
+        ``refresh_token`` are stored in ``UserSessionState.state`` under the
+        keys ``entra_access_token`` and ``entra_refresh_token`` respectively.
+        Tiled persists this state in the session DB and embeds it verbatim in
+        every Tiled HMAC access token, making the tokens available to downstream
+        services that rely on Tiled authentication via ``get_session_state()``.
+
+        Security note: the Entra access token is therefore visible inside the
+        Tiled JWT (base64-encoded, not encrypted).  The Tiled access token is
+        short-lived (default 15 min) and only transmitted over HTTPS, which
+        limits the exposure window.
+
+        The ``refresh_token`` enables silent renewal: when the Entra access
+        token expires (~1 h), a downstream service can call the Entra token
+        endpoint with ``grant_type=refresh_token`` to obtain a fresh pair and
+        write it back to the session DB so subsequent Tiled ``slide_session``
+        calls propagate the update automatically.
+        """
+        code = request.query_params.get("code")
+        if not code:
+            logger.warning(
+                "Authentication failed: No authorization code parameter provided."
+            )
+            return None
+        redirect_uri = f"{get_root_url(request)}{request.url.path}"
+        response = await exchange_code(
+            self.token_endpoint,
+            code,
+            self._client_id,
+            self._client_secret.get_secret_value(),
+            redirect_uri,
+            extra_scopes=self.extra_scopes,
+        )
+        response_body = response.json()
+        if response.is_error:
+            logger.error("Authentication error: %r", response_body)
+            return None
+        id_token = response_body["id_token"]
+        access_token = response_body.get("access_token")
+        refresh_token = response_body.get("refresh_token")
+        try:
+            verified_body = self.decode_token(id_token, access_token)
+        except JWTError:
+            logger.exception(
+                "Authentication error. Unverified token: %r",
+                jwt.get_unverified_claims(id_token),
+            )
+            return None
+        # Log the id_token claims available for username resolution so
+        # misconfigurations (missing optional claims) are easy to diagnose.
+        # Logged at DEBUG to avoid leaking PII in production logs by default.
+        logger.debug(
+            "EntraAuthenticator.authenticate: id_token claims present: %s",
+            sorted(verified_body.keys()),
+        )
+        logger.debug(
+            "EntraAuthenticator.authenticate: entra_username=%r user=%r entra_sub=%r",
+            verified_body.get("entra_username"),
+            verified_body.get("user"),
+            verified_body.get("entra_sub"),
+        )
+        # Use the human-readable username (e.g. "dallan") instead of the
+        # opaque UUID-based "sub" that OIDCAuthenticator would use.
+        username = verified_body.get("user") or verified_body["sub"]
+        # Store the Entra access and refresh tokens so that downstream
+        # services that rely on Tiled authentication can perform an OBO exchange
+        # to obtain per-user tokens for other services.  The refresh token
+        # allows silent renewal without requiring the user to re-authenticate.
+        state: dict = {}
+        if access_token:
+            state["entra_access_token"] = access_token
+        if refresh_token:
+            state["entra_refresh_token"] = refresh_token
+        return UserSessionState(username, state)
+
+
 async def exchange_code(
     token_uri: str,
     auth_code: str,
     client_id: str,
     client_secret: str,
     redirect_uri: str,
+    extra_scopes: Optional[List[str]] = None,
 ) -> httpx.Response:
-    """Method that talks to an IdP to exchange a code for an access_token and/or id_token
-    Args:
-        token_url ([type]): [description]
-        auth_code ([type]): [description]
+    """Exchange an authorization code for tokens at the IdP token endpoint.
+
+    Explicitly requests ``openid offline_access`` scopes in the token POST body
+    so that the IdP returns a ``refresh_token`` unconditionally.  This is safe
+    even when ``offline_access`` was already included in the authorization URL
+    scope — the IdP simply ignores duplicates.  Including it here makes the
+    refresh token reliable regardless of how the authorization URL was
+    constructed, which is important for downstream OBO refresh flows.
+
+    ``extra_scopes`` (e.g. ``["api://<client-id>/access_as_user"]``) are
+    appended to the scope string.  Entra only issues an ``access_token`` whose
+    ``aud`` matches the requested resource scope, so any scope that a downstream
+    OBO exchange will use as the ``assertion`` audience **must** be included
+    here — requesting it only on the authorization URL redirect is not
+    sufficient, because Entra does not carry scopes from the redirect into the
+    token POST implicitly.
     """
+    scopes = {"openid", "offline_access"}
+    if extra_scopes:
+        scopes.update(extra_scopes)
     auth_value = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
     response = httpx.post(
         url=token_uri,
@@ -331,6 +537,7 @@ async def exchange_code(
             "redirect_uri": redirect_uri,
             "code": auth_code,
             "client_secret": client_secret,
+            "scope": " ".join(sorted(scopes)),
         },
         headers={"Authorization": f"Basic {auth_value}"},
     )
@@ -338,7 +545,6 @@ async def exchange_code(
 
 
 class SAMLAuthenticator(ExternalAuthenticator):
-
     def __init__(
         self,
         saml_settings,  # See EXAMPLE_SAML_SETTINGS below.
@@ -356,7 +562,9 @@ def __init__(
             # The PyPI package name is 'python3-saml'
             # but it imports as 'onelogin'.
             # https://github.com/onelogin/python3-saml
-            raise ModuleNotFoundError("This SAMLAuthenticator requires 'python3-saml' to be installed.")
+            raise ModuleNotFoundError(
+                "This SAMLAuthenticator requires 'python3-saml' to be installed."
+            )
 
         from onelogin.saml2.auth import OneLogin_Saml2_Auth
 
@@ -371,7 +579,9 @@ async def saml_login(request: Request) -> RedirectResponse:
 
     async def authenticate(self, request: Request) -> Optional[UserSessionState]:
         if not modules_available("onelogin"):
-            raise ModuleNotFoundError("This SAMLAuthenticator requires the module 'oneline' to be installed.")
+            raise ModuleNotFoundError(
+                "This SAMLAuthenticator requires the module 'oneline' to be installed."
+            )
         from onelogin.saml2.auth import OneLogin_Saml2_Auth
 
         req = await prepare_saml_from_fastapi_request(request, True)
@@ -380,7 +590,8 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]:
         errors = auth.get_errors()  # This method receives an array with the errors
         if errors:
             raise Exception(
-                "Error when processing SAML Response: %s %s" % (", ".join(errors), auth.get_last_error_reason())
+                "Error when processing SAML Response: %s %s"
+                % (", ".join(errors), auth.get_last_error_reason())
             )
         if auth.is_authenticated():
             # Return a string that the Identity can use as id.
@@ -399,7 +610,7 @@ async def prepare_saml_from_fastapi_request(request: Request) -> Mapping[str, st
         "server_port": request.url.port,
         "script_name": request.url.path,
         "post_data": {},
-        "get_data": {},
+        "get_data": {}
         # Advanced request options
         # "https": "",
         # "request_uri": "",
@@ -420,7 +631,6 @@ async def prepare_saml_from_fastapi_request(request: Request) -> Mapping[str, st
 
 class LDAPAuthenticator(InternalAuthenticator):
     """
-    LDAP authenticator.
     The authenticator code is based on https://github.com/jupyterhub/ldapauthenticator
     The parameter ``use_tls`` was added for convenience of testing.
 
@@ -645,7 +855,9 @@ def __init__(
         self.escape_userdn = escape_userdn
         self.search_filter = search_filter
         self.attributes = attributes if attributes else []
-        self.auth_state_attributes = auth_state_attributes if auth_state_attributes else []
+        self.auth_state_attributes = (
+            auth_state_attributes if auth_state_attributes else []
+        )
         self.use_lookup_dn_username = use_lookup_dn_username
 
         if isinstance(server_address, str):
@@ -658,10 +870,14 @@ def __init__(
                 f"type(server_address)={type(server_address)}"
             )
         if not server_address_list:
-            raise ValueError("No servers are specified: 'server_address' is an empty list")
+            raise ValueError(
+                "No servers are specified: 'server_address' is an empty list"
+            )
 
         self.server_address_list = server_address_list
-        self.server_port = server_port if server_port is not None else self._server_port_default()
+        self.server_port = (
+            server_port if server_port is not None else self._server_port_default()
+        )
         self.confirmation_message = confirmation_message
 
     def _server_port_default(self):
@@ -682,7 +898,7 @@ async def resolve_username(self, username_supplied_by_user):
         is_bound = await asyncio.get_running_loop().run_in_executor(None, conn.bind)
         if not is_bound:
             msg = "Failed to connect to LDAP server with search user '{search_dn}'"
-            self.log.warning(msg.format(search_dn=search_dn))
+            logger.warning(msg.format(search_dn=search_dn))
             return (None, None)
 
         search_filter = self.lookup_dn_search_filter.format(
@@ -715,8 +931,15 @@ async def resolve_username(self, username_supplied_by_user):
 
         response = conn.response
         if len(response) == 0 or "attributes" not in response[0].keys():
-            msg = "No entry found for user '{username}' " "when looking up attribute '{attribute}'"
-            logger.warning(msg.format(username=username_supplied_by_user, attribute=self.user_attribute))
+            msg = (
+                "No entry found for user '{username}' "
+                "when looking up attribute '{attribute}'"
+            )
+            logger.warning(
+                msg.format(
+                    username=username_supplied_by_user, attribute=self.user_attribute
+                )
+            )
             return (None, None)
 
         user_dn = response[0]["attributes"][self.lookup_dn_user_dn_attribute]
@@ -774,7 +997,9 @@ def get_connection(self, userdn, password):
             )
             server_pool.add(server)
 
-        auto_bind_no_ssl = ldap3.AUTO_BIND_TLS_BEFORE_BIND if self.use_tls else ldap3.AUTO_BIND_NO_TLS
+        auto_bind_no_ssl = (
+            ldap3.AUTO_BIND_TLS_BEFORE_BIND if self.use_tls else ldap3.AUTO_BIND_NO_TLS
+        )
         auto_bind = ldap3.AUTO_BIND_NO_TLS if self.use_ssl else auto_bind_no_ssl
         conn = ldap3.Connection(
             server_pool,
@@ -799,7 +1024,9 @@ async def get_user_attributes(self, conn, userdn):
                 attrs = conn.entries[0].entry_attributes_as_dict
         return attrs
 
-    async def authenticate(self, username: str, password: str) -> Optional[UserSessionState]:
+    async def authenticate(
+        self, username: str, password: str
+    ) -> Optional[UserSessionState]:
         import ldap3
 
         username_saved = username  # Save the user name passed as a parameter
@@ -825,7 +1052,9 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi
 
         # sanity check
         if not self.lookup_dn and not bind_dn_template:
-            logger.warning("Login not allowed, please configure 'lookup_dn' or 'bind_dn_template'.")
+            logger.warning(
+                "Login not allowed, please configure 'lookup_dn' or 'bind_dn_template'."
+            )
             return None
 
         if self.lookup_dn:
@@ -863,7 +1092,9 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi
                 if conn.bound:
                     is_bound = True
                 else:
-                    is_bound = await asyncio.get_running_loop().run_in_executor(None, conn.bind)
+                    is_bound = await asyncio.get_running_loop().run_in_executor(
+                        None, conn.bind
+                    )
 
             msg = msg.format(username=username, userdn=userdn, is_bound=is_bound)
             logger.debug(msg)
@@ -876,7 +1107,9 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi
             return None
 
         if self.search_filter:
-            search_filter = self.search_filter.format(userattr=self.user_attribute, username=username)
+            search_filter = self.search_filter.format(
+                userattr=self.user_attribute, username=username
+            )
 
             search_func = functools.partial(
                 conn.search,
@@ -890,18 +1123,33 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi
             n_users = len(conn.response)
             if n_users == 0:
                 msg = "User with '{userattr}={username}' not found in directory"
-                logger.warning(msg.format(userattr=self.user_attribute, username=username))
+                logger.warning(
+                    msg.format(userattr=self.user_attribute, username=username)
+                )
                 return None
             if n_users > 1:
-                msg = "Duplicate users found! " "{n_users} users found with '{userattr}={username}'"
-                logger.warning(msg.format(userattr=self.user_attribute, username=username, n_users=n_users))
+                msg = (
+                    "Duplicate users found! "
+                    "{n_users} users found with '{userattr}={username}'"
+                )
+                logger.warning(
+                    msg.format(
+                        userattr=self.user_attribute, username=username, n_users=n_users
+                    )
+                )
                 return None
 
         if self.allowed_groups:
             logger.debug("username:%s Using dn %s", username, userdn)
             found = False
             for group in self.allowed_groups:
-                group_filter = "(|" "(member={userdn})" "(uniqueMember={userdn})" "(memberUid={uid})" ")"
+                group_filter = (
+                    "(|"
+                    "(member={userdn})"
+                    "(uniqueMember={userdn})"
+                    "(memberUid={uid})"
+                    ")"
+                )
                 group_filter = group_filter.format(userdn=userdn, uid=username)
                 group_attributes = ["member", "uniqueMember", "memberUid"]
 
@@ -912,7 +1160,9 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi
                     search_filter=group_filter,
                     attributes=group_attributes,
                 )
-                found = await asyncio.get_running_loop().run_in_executor(None, search_func)
+                found = await asyncio.get_running_loop().run_in_executor(
+                    None, search_func
+                )
                 if found:
                     break
 
diff --git a/bluesky_httpserver/database/core.py b/bluesky_httpserver/database/core.py
index 52d102f..aee3a3a 100644
--- a/bluesky_httpserver/database/core.py
+++ b/bluesky_httpserver/database/core.py
@@ -15,9 +15,9 @@
 
 # This is the alembic revision ID of the database revision
 # required by this version of Tiled.
-REQUIRED_REVISION = "a1b2c3d4e5f6"
+REQUIRED_REVISION = "b2c3d4e5f6a7"
 # This is list of all valid revisions (from current to oldest).
-ALL_REVISIONS = ["a1b2c3d4e5f6", "722ff4e4fcc7", "481830dd6c11"]
+ALL_REVISIONS = ["b2c3d4e5f6a7", "a1b2c3d4e5f6", "722ff4e4fcc7", "481830dd6c11"]
 
 
 # def create_default_roles(engine):
@@ -209,6 +209,33 @@ def create_user(db, identity_provider, id):
     return principal
 
 
+def get_or_create_principal(db, identity_provider, id):
+    """Return a Principal for (identity_provider, id), creating it if needed.
+
+    Mirrors tiled's ``authn_database.core.get_or_create_principal``.  Unlike
+    :func:`create_session`, this helper only touches the Principal/Identity
+    tables — it never creates a Session row.  It is intended for principals
+    that authenticate with a token minted by an external OIDC provider (i.e.
+    :class:`bluesky_httpserver.authenticators.ProxiedOIDCAuthenticator`
+    subclasses) where the JWT itself is authoritative and no bluesky-httpserver
+    session lifetime is required.
+
+    On successful lookup the matching ``Identity.latest_login`` is updated to
+    now.
+    """
+    identity = (
+        db.query(Identity)
+        .filter(Identity.id == id)
+        .filter(Identity.provider == identity_provider)
+        .first()
+    )
+    if identity is not None:
+        identity.latest_login = datetime.utcnow()
+        db.commit()
+        return identity.principal
+    return create_user(db, identity_provider, id)
+
+
 def lookup_valid_session(db, session_id):
     if isinstance(session_id, int):
         # Old versions of tiled used an integer sid.
@@ -216,6 +243,8 @@ def lookup_valid_session(db, session_id):
         return None
 
     session = db.query(Session).filter(Session.uuid == uuid_module.UUID(hex=session_id)).first()
+    if session is None:
+        return None
     if session.expiration_time is not None and session.expiration_time < datetime.utcnow():
         db.delete(session)
         db.commit()
diff --git a/bluesky_httpserver/database/migrations/versions/a1b2c3d4e5f6_add_pending_sessions.py b/bluesky_httpserver/database/migrations/versions/a1b2c3d4e5f6_add_pending_sessions.py
new file mode 100644
index 0000000..8876219
--- /dev/null
+++ b/bluesky_httpserver/database/migrations/versions/a1b2c3d4e5f6_add_pending_sessions.py
@@ -0,0 +1,59 @@
+"""Add PendingSession table for device code flow.
+
+Revision ID: a1b2c3d4e5f6
+Revises: 722ff4e4fcc7
+Create Date: 2026-02-13 12:00:00.000000
+
+"""
+
+from alembic import op
+from sqlalchemy import Column, DateTime, ForeignKey, Integer, LargeBinary, Unicode
+from sqlalchemy.sql import func
+
+# revision identifiers, used by Alembic.
+revision = "a1b2c3d4e5f6"
+down_revision = "722ff4e4fcc7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    """
+    Add pending_sessions table for device code flow authentication.
+    """
+    op.create_table(
+        "pending_sessions",
+        Column("time_created", DateTime(timezone=False), server_default=func.now()),
+        Column("time_updated", DateTime(timezone=False), onupdate=func.now()),
+        Column(
+            "hashed_device_code",
+            LargeBinary(32),
+            primary_key=True,
+            index=True,
+            nullable=False,
+        ),
+        Column(
+            "user_code",
+            Unicode(8),
+            index=True,
+            nullable=False,
+        ),
+        Column(
+            "expiration_time",
+            DateTime(timezone=False),
+            nullable=False,
+        ),
+        Column(
+            "session_id",
+            Integer,
+            ForeignKey("sessions.id"),
+            nullable=True,
+        ),
+    )
+
+
+def downgrade():
+    """
+    Remove pending_sessions table.
+    """
+    op.drop_table("pending_sessions")
diff --git a/bluesky_httpserver/database/migrations/versions/b2c3d4e5f6a7_add_session_state.py b/bluesky_httpserver/database/migrations/versions/b2c3d4e5f6a7_add_session_state.py
new file mode 100644
index 0000000..3fa5e27
--- /dev/null
+++ b/bluesky_httpserver/database/migrations/versions/b2c3d4e5f6a7_add_session_state.py
@@ -0,0 +1,40 @@
+"""Add ``state`` column to sessions.
+
+Revision ID: b2c3d4e5f6a7
+Revises: a1b2c3d4e5f6
+Create Date: 2026-07-06 12:00:00.000000
+
+Ports the ``Session.state`` column introduced upstream in tiled 0.2.10.  The
+column stores a JSON dict populated by an authenticator's
+``UserSessionState.state`` (e.g. upstream OIDC access/refresh tokens carried
+across refresh_session calls) so downstream services that share
+bluesky-httpserver authentication can retrieve them via Tiled access tokens.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "b2c3d4e5f6a7"
+down_revision = "a1b2c3d4e5f6"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    """Add sessions.state JSON column with default ``{}``."""
+    with op.batch_alter_table("sessions") as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "state",
+                sa.JSON(),
+                nullable=False,
+                server_default="{}",
+            )
+        )
+
+
+def downgrade():
+    """Drop sessions.state column."""
+    with op.batch_alter_table("sessions") as batch_op:
+        batch_op.drop_column("state")
diff --git a/bluesky_httpserver/database/orm.py b/bluesky_httpserver/database/orm.py
index 7611824..5d4b288 100644
--- a/bluesky_httpserver/database/orm.py
+++ b/bluesky_httpserver/database/orm.py
@@ -1,7 +1,7 @@
 import json
 import uuid as uuid_module
 
-from sqlalchemy import Boolean, Column, DateTime, Enum, ForeignKey, Integer, LargeBinary, Unicode  # Table,
+from sqlalchemy import JSON, Boolean, Column, DateTime, Enum, ForeignKey, Integer, LargeBinary, Unicode  # Table,
 from sqlalchemy.orm import relationship
 from sqlalchemy.sql import func
 from sqlalchemy.types import TypeDecorator
@@ -179,6 +179,11 @@ class Session(Timestamped, Base):
     expiration_time = Column(DateTime(timezone=False), nullable=False)
     principal_id = Column(Integer, ForeignKey("principals.id"), nullable=False)
     revoked = Column(Boolean, default=False, nullable=False)
+    # Free-form state supplied by the authenticator via UserSessionState.state
+    # (e.g. an upstream OIDC access_token/refresh_token for OBO exchange).
+    # Persisted so it survives across refresh_session calls; the values are
+    # available to downstream services through access tokens.
+    state = Column(JSON, nullable=False, default=dict, server_default="{}")
 
     principal = relationship("Principal", back_populates="sessions")
     pending_sessions = relationship("PendingSession", back_populates="session")
diff --git a/bluesky_httpserver/routers/core_api.py b/bluesky_httpserver/routers/core_api.py
index 9e47d58..f86dcb1 100644
--- a/bluesky_httpserver/routers/core_api.py
+++ b/bluesky_httpserver/routers/core_api.py
@@ -14,7 +14,11 @@
 else:
     from pydantic_settings import BaseSettings
 
-from ..authentication import get_current_principal, get_current_principal_websocket
+from ..authentication import (
+    authenticate_websocket_first_message,
+    get_current_principal,
+    get_current_principal_websocket,
+)
 from ..console_output import ConsoleOutputEventStream, StreamingResponseFromClass
 from ..resources import SERVER_RESOURCES as SR
 from ..settings import get_settings
@@ -1150,14 +1154,71 @@ def is_alive(self):
         return self._is_alive
 
 
+# WebSocket close codes.  4001 = invalid token, 4401 = auth required
+# (RFC 6455 leaves 4000-4999 for application use).
+_WS_CLOSE_INVALID_TOKEN = 4001
+_WS_CLOSE_AUTH_REQUIRED = 4401
+
+
+async def _authenticate_websocket(websocket, scopes):
+    """Resolve a Principal for a WebSocket connection.
+
+    Tries in order:
+
+    1. ``Authorization: Bearer|ApiKey ...`` header (populated by curl/CLI).
+    2. ``?access_token=...`` or ``?api_key=...`` query parameter (populated
+       by browsers, which cannot set request headers on a WebSocket
+       handshake).
+    3. First-message handshake: accepts the socket, then reads one JSON
+       message of the form
+       ``{"type": "auth", "api_key": "..."}`` or
+       ``{"type": "auth", "access_token": "..."}``.
+       On success the socket stays open; on failure the socket is closed
+       with code 4001 and ``None`` is returned.
+
+    Returns ``(principal, accepted)`` where ``accepted`` indicates whether
+    the socket has already been ``.accept()``-ed by this helper (True only
+    when the first-message path was used).  Callers that receive ``None``
+    for the principal have already had the socket closed and should return
+    immediately.
+    """
+    principal = get_current_principal_websocket(websocket=websocket, scopes=scopes)
+    if principal is not None:
+        return principal, False
+
+    # Fall back to the first-message handshake.  Accept the socket so that we
+    # can receive the auth payload; the client is expected to send it as the
+    # very first frame.
+    await websocket.accept()
+    try:
+        message = await asyncio.wait_for(websocket.receive_json(), timeout=10)
+    except asyncio.TimeoutError:
+        await websocket.close(code=_WS_CLOSE_AUTH_REQUIRED, reason="Auth required")
+        return None, True
+    except WebSocketDisconnect:
+        # Client already gone — no close frame needed.
+        return None, True
+    except Exception:
+        logger.exception("Unexpected error receiving WebSocket auth frame")
+        await websocket.close(code=_WS_CLOSE_AUTH_REQUIRED, reason="Auth required")
+        return None, True
+
+    principal = authenticate_websocket_first_message(websocket, message)
+    if principal is None:
+        await websocket.close(code=_WS_CLOSE_INVALID_TOKEN, reason="Invalid token")
+        return None, True
+
+    return principal, True
+
+
 @router.websocket("/console_output/ws")
 async def console_output_ws(websocket: WebSocket, scopes=["read:console"]):
-    principal = get_current_principal_websocket(websocket=websocket, scopes=scopes)
+    principal, accepted = await _authenticate_websocket(websocket, scopes)
     if not principal:
-        await websocket.close(code=4001, reason="Invalid token")
         return
 
-    await websocket.accept()
+    if not accepted:
+        await websocket.accept()
     q = SR.console_output_stream.add_queue(websocket)
     wsmon = WebSocketMonitor(websocket)
     wsmon.start()
@@ -1178,12 +1239,12 @@ async def console_output_ws(websocket: WebSocket, scopes=["read:console"]):
 
 @router.websocket("/status/ws")
 async def status_ws(websocket: WebSocket, scopes=["read:monitor"]):
-    principal = get_current_principal_websocket(websocket=websocket, scopes=scopes)
+    principal, accepted = await _authenticate_websocket(websocket, scopes)
     if not principal:
-        await websocket.close(code=4001, reason="Invalid token")
         return
 
-    await websocket.accept()
+    if not accepted:
+        await websocket.accept()
     q = SR.system_info_stream.add_queue_status(websocket)
     wsmon = WebSocketMonitor(websocket)
     wsmon.start()
@@ -1205,12 +1266,12 @@ async def status_ws(websocket: WebSocket, scopes=["read:monitor"]):
 
 @router.websocket("/info/ws")
 async def info_ws(websocket: WebSocket, scopes=["read:monitor"]):
-    principal = get_current_principal_websocket(websocket=websocket, scopes=scopes)
+    principal, accepted = await _authenticate_websocket(websocket, scopes)
     if not principal:
-        await websocket.close(code=4001, reason="Invalid token")
         return
 
-    await websocket.accept()
+    if not accepted:
+        await websocket.accept()
     q = SR.system_info_stream.add_queue_info(websocket)
     wsmon = WebSocketMonitor(websocket)
     wsmon.start()
diff --git a/bluesky_httpserver/schemas.py b/bluesky_httpserver/schemas.py
index f1d9fcb..f127b88 100644
--- a/bluesky_httpserver/schemas.py
+++ b/bluesky_httpserver/schemas.py
@@ -271,6 +271,7 @@ class Session(pydantic.BaseModel, **orm):
     uuid: uuid.UUID
     expiration_time: datetime
     revoked: bool
+    state: Dict = {}
 
 
 class Principal(pydantic.BaseModel, **orm):
@@ -289,6 +290,7 @@ class Principal(pydantic.BaseModel, **orm):
     roles: Optional[List[str]] = []
     scopes: Optional[List[str]] = []
     api_key_scopes: Optional[Union[List[str], None]] = None
+    access_token: Optional[str] = None
 
     @classmethod
     def from_orm(cls, orm, latest_activity=None):
diff --git a/bluesky_httpserver/tests/test_auth_port_0_2_12.py b/bluesky_httpserver/tests/test_auth_port_0_2_12.py
new file mode 100644
index 0000000..2807ad8
--- /dev/null
+++ b/bluesky_httpserver/tests/test_auth_port_0_2_12.py
@@ -0,0 +1,592 @@
+"""
+Tests for the tiled v0.2.9 -> v0.2.12 auth port.
+
+Covers:
+- decode_token fallback ordering (Phase 2.1)
+- _extract_scopes helper (Phase 2.2)
+- get_or_create_principal (Phase 2.6)
+- Session.state column round-trip (Phase 4.3)
+- Principal.access_token schema field (Phase 4.1)
+- authorize redirect: offline_access + prompt=login (Phase 2.5)
+- JWKS cache TTL (Phase 1.2)
+- OIDC decode_token(id_token, access_token=None) signature (Phase 1.1)
+- EntraAuthenticator.decode_token no longer TypeErrors (Phase 1.3)
+- authenticate_websocket_first_message (Phase 2.4)
+"""
+
+from __future__ import annotations
+
+import time
+import uuid
+from datetime import timedelta
+from typing import Any, Tuple
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+from cryptography.hazmat.primitives.asymmetric import rsa
+from fastapi import HTTPException
+from jose import ExpiredSignatureError, jwt
+from jose.backends import RSAKey
+from respx import MockRouter
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from bluesky_httpserver import _authentication as _auth
+from bluesky_httpserver import schemas
+from bluesky_httpserver.authenticators import (
+    EntraAuthenticator,
+    OIDCAuthenticator,
+    ProxiedOIDCAuthenticator,
+)
+from bluesky_httpserver.database import orm as db_orm
+from bluesky_httpserver.database.base import Base
+from bluesky_httpserver.database.core import (
+    create_user,
+    get_or_create_principal,
+)
+
+
+# ---------------------------------------------------------------------------
+# Shared OIDC fixtures (mirrors ones in test_authenticators.py so this file
+# can be run standalone).
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def oidc_well_known_url(oidc_base_url: str) -> str:
+    return f"{oidc_base_url}.well-known/openid-configuration"
+
+
+@pytest.fixture
+def keys() -> Tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey]:
+    priv = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    return priv, priv.public_key()
+
+
+@pytest.fixture
+def json_web_keyset(keys):
+    _, public = keys
+    return [RSAKey(key=public, algorithm="RS256").to_dict()]
+
+
+@pytest.fixture
+def mock_oidc_server(respx_mock: MockRouter, oidc_well_known_url, well_known_response, json_web_keyset):
+    respx_mock.get(oidc_well_known_url).mock(
+        return_value=httpx.Response(200, json=well_known_response)
+    )
+    respx_mock.get(well_known_response["jwks_uri"]).mock(
+        return_value=httpx.Response(200, json={"keys": json_web_keyset})
+    )
+    return respx_mock
+
+
+def _make_token(private_key, **overrides) -> str:
+    now = int(time.time())
+    claims = {
+        "aud": "tiled",
+        "exp": now + 1500,
+        "iat": now - 10,
+        "iss": "https://example.com/realms/example",
+        "sub": "abc-123",
+    }
+    claims.update(overrides)
+    return jwt.encode(claims, key=private_key, algorithm="RS256", headers={"kid": "secret"})
+
+
+# ---------------------------------------------------------------------------
+# Phase 1.1 - decode_token accepts (id_token, access_token=None)
+# ---------------------------------------------------------------------------
+
+
+def test_oidc_decode_token_accepts_access_token_kwarg(mock_oidc_server, oidc_well_known_url, keys):
+    """After the port, decode_token must accept an optional second positional
+    argument (the access_token, used for at_hash validation)."""
+    priv, _ = keys
+    auth = OIDCAuthenticator("tiled", "tiled", "secret", well_known_uri=oidc_well_known_url)
+    id_token = _make_token(priv)
+    # Both calling conventions must work.
+    single = auth.decode_token(id_token)
+    dual = auth.decode_token(id_token, access_token=None)
+    assert single == dual
+
+
+# ---------------------------------------------------------------------------
+# Phase 1.2 - JWKS cache TTL is 1 hour (not 7 days)
+# ---------------------------------------------------------------------------
+
+
+def test_oidc_keys_cache_ttl_is_one_hour():
+    """The @cached decorator on OIDCAuthenticator.keys() must use a 1h TTL."""
+    # cachetools stores the TTL on the cache attached to the wrapped function.
+    method = OIDCAuthenticator.keys
+    # ``cachetools.func.ttl_cache`` or ``cachetools.cached(TTLCache(...))``
+    # both expose the underlying cache via the wrapped function.  We only
+    # need to check that the TTL is one hour, not seven days.
+    cache = getattr(method, "cache", None)
+    if cache is None:
+        # cachetools>=5 uses __wrapped__.cache or the closure.  Fall back to
+        # inspecting closures.
+        closures = getattr(method, "__closure__", None) or ()
+        for cell in closures:
+            obj = cell.cell_contents
+            if hasattr(obj, "ttl"):
+                cache = obj
+                break
+    assert cache is not None, "Unable to locate TTLCache on OIDCAuthenticator.keys"
+    # 1 h == 3600 s. Assert it's an hour, definitely not 7 days.
+    assert cache.ttl == pytest.approx(timedelta(hours=1).total_seconds())
+    assert cache.ttl < timedelta(days=1).total_seconds()
+
+
+# ---------------------------------------------------------------------------
+# Phase 1.3 - EntraAuthenticator.decode_token no longer TypeErrors
+# ---------------------------------------------------------------------------
+
+
+def test_entra_authenticator_decode_token_signature(mock_oidc_server, oidc_well_known_url, keys, monkeypatch):
+    """Regression test for the fork-local defect where
+    EntraAuthenticator.decode_token called super().decode_token(id_token,
+    access_token) against an OIDCAuthenticator whose decode_token only
+    accepted a single argument.  After the port the parent accepts an
+    optional access_token."""
+    priv, _ = keys
+    auth = EntraAuthenticator(
+        audience="tiled",
+        client_id="tiled",
+        well_known_uri=oidc_well_known_url,
+        device_flow_client_id="tiled-cli",
+        scopes_map={"User.Read": ["read:queue"]},
+    )
+    id_token = _make_token(
+        priv,
+        preferred_username="jane@example.com",
+        scp="User.Read",
+    )
+    # Must not raise TypeError from arg-count mismatch, nor JWTError.
+    claims = auth.decode_token(id_token, access_token="opaque-access-token")
+    # UUID5 rewrites 'sub', preserves entra_sub, resolves user, maps scopes.
+    assert claims["entra_sub"] == "abc-123"
+    assert claims["user"] == "jane"
+    assert "read:queue" in claims["scope"].split()
+
+
+# ---------------------------------------------------------------------------
+# Phase 2.1 - decode_token fallback ordering
+# ---------------------------------------------------------------------------
+
+
+def _encode_hs(payload, key):
+    return jwt.encode(payload, key, algorithm="HS256")
+
+
+def test_decode_token_tries_hmac_keys_first():
+    """The bluesky-httpserver HMAC keys must be tried before any proxied
+    authenticator fallback.  Otherwise a stolen OIDC key could impersonate
+    a locally-minted API-key session."""
+    payload = {"sub": "u1", "sub_typ": "user", "ids": []}
+    token = _encode_hs(payload, "k-primary")
+
+    fake_proxied = MagicMock(spec=ProxiedOIDCAuthenticator)
+    fake_proxied.decode_token.side_effect = AssertionError("must not be called")
+
+    result = _auth.decode_token(token, ["k-primary", "k-secondary"], fake_proxied)
+    assert result == payload
+    fake_proxied.decode_token.assert_not_called()
+
+
+def test_decode_token_supports_key_rotation():
+    """Older tokens minted with a rotated-out key must still decode if the
+    old key is present in secret_keys."""
+    token = _encode_hs({"sub": "u1"}, "old-key")
+    result = _auth.decode_token(token, ["new-key", "old-key"], None)
+    assert result["sub"] == "u1"
+
+
+def test_decode_token_falls_back_to_proxied_authenticator():
+    """When no HMAC key accepts the token, delegate to a
+    ProxiedOIDCAuthenticator.decode_token.  This enables OIDC-minted access
+    tokens (device-code flow) to be accepted by protected endpoints."""
+    # Encode with a key that is not in secret_keys, so HMAC decoding fails.
+    token = _encode_hs({"sub": "external-u", "scp": "read:queue"}, "unknown-key")
+
+    fake_proxied = MagicMock(spec=ProxiedOIDCAuthenticator)
+    fake_proxied.decode_token.return_value = {
+        "sub": "external-u",
+        "scp": "read:queue",
+    }
+
+    result = _auth.decode_token(token, ["hmac-key"], fake_proxied)
+    assert result == {"sub": "external-u", "scp": "read:queue"}
+    fake_proxied.decode_token.assert_called_once_with(token)
+
+
+def test_decode_token_raises_when_no_key_matches():
+    token = _encode_hs({"sub": "u1"}, "unknown")
+    with pytest.raises(HTTPException) as excinfo:
+        _auth.decode_token(token, ["a", "b"], None)
+    assert excinfo.value.status_code == 401
+
+
+def test_decode_token_propagates_expired_signature():
+    """Expired tokens raise ExpiredSignatureError verbatim so the caller can
+    return a distinct 401 with 'refresh token' guidance rather than a
+    generic 'invalid credentials'."""
+    past = int(time.time()) - 3600
+    token = jwt.encode({"sub": "u1", "exp": past}, "k", algorithm="HS256")
+    with pytest.raises(ExpiredSignatureError):
+        _auth.decode_token(token, ["k"], None)
+
+
+# ---------------------------------------------------------------------------
+# Phase 2.2 - _extract_scopes helper
+# ---------------------------------------------------------------------------
+
+
+class TestExtractScopes:
+    def test_scp_as_space_separated_string(self):
+        assert _auth._extract_scopes({"scp": "read:queue write:queue:edit"}) == {
+            "read:queue",
+            "write:queue:edit",
+        }
+
+    def test_scp_as_list(self):
+        assert _auth._extract_scopes({"scp": ["read:queue", "read:status"]}) == {
+            "read:queue",
+            "read:status",
+        }
+
+    def test_scope_as_space_separated_string(self):
+        assert _auth._extract_scopes({"scope": "read:queue read:status"}) == {
+            "read:queue",
+            "read:status",
+        }
+
+    def test_union_of_both_claims(self):
+        assert _auth._extract_scopes(
+            {"scp": ["read:queue"], "scope": "read:status"}
+        ) == {"read:queue", "read:status"}
+
+    def test_empty_or_missing(self):
+        assert _auth._extract_scopes({}) == set()
+        assert _auth._extract_scopes({"scp": "", "scope": ""}) == set()
+
+
+# ---------------------------------------------------------------------------
+# Phase 2.5 - authorize redirect includes offline_access + prompt=login
+# ---------------------------------------------------------------------------
+
+
+class _FakeAuthorizationEndpoint:
+    """Stand-in for the ``authorization_endpoint`` cached_property.  We do not
+    want to hit an actual OIDC well-known URL from a unit test."""
+
+    def __init__(self):
+        self.captured_params: dict | None = None
+
+    def copy_with(self, params):
+        self.captured_params = params
+        # Return an httpx.URL so RedirectResponse can str() it cleanly.
+        return httpx.URL("https://idp.example.com/authorize").copy_with(params=params)
+
+
+@pytest.mark.asyncio
+async def test_authorize_route_requests_offline_access_and_prompts_login():
+    """Verify that the browser-facing /authorize redirect asks the IdP for
+    offline_access (to guarantee a refresh_token) and always prompts the
+    user (avoids surprising silent SSO)."""
+    fake_endpoint = _FakeAuthorizationEndpoint()
+
+    class FakeAuthenticator:
+        client_id = "test-client"
+        authorization_endpoint = fake_endpoint
+        extra_scopes = ["api://tiled/access_as_user"]
+
+    class FakeRequest:
+        headers = {"host": "localhost:8000"}
+        scope = {"scheme": "http", "root_path": ""}
+
+    route = _auth.build_authorize_route(FakeAuthenticator(), "orcid")
+    resp = await route(FakeRequest(), state=None)
+    assert resp.status_code == 307
+    params = fake_endpoint.captured_params
+    assert params["prompt"] == "login"
+    scopes = set(params["scope"].split())
+    assert {"openid", "offline_access", "api://tiled/access_as_user"}.issubset(scopes)
+
+
+# ---------------------------------------------------------------------------
+# Phase 4.1 - schemas.Principal.access_token
+# ---------------------------------------------------------------------------
+
+
+def test_principal_carries_access_token_field():
+    """Externally-authenticated principals attach the raw OIDC access token
+    so downstream services can perform OBO exchanges."""
+    p = schemas.Principal(
+        uuid=uuid.uuid4(),
+        type=schemas.PrincipalType.user,
+        identities=[schemas.Identity(id="jane", provider="entra")],
+        access_token="opaque-entra-token",
+    )
+    assert p.access_token == "opaque-entra-token"
+    # Default is None so existing serializations of API-key-authenticated
+    # principals are unaffected.
+    p2 = schemas.Principal(uuid=uuid.uuid4(), type=schemas.PrincipalType.user)
+    assert p2.access_token is None
+
+
+# ---------------------------------------------------------------------------
+# Phase 4.3 - Session.state column round-trip
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def sqlite_session():
+    engine = create_engine("sqlite://")
+    Base.metadata.create_all(engine)
+    SessionLocal = sessionmaker(bind=engine)
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+        engine.dispose()
+
+
+def test_session_state_column_round_trips(sqlite_session):
+    """Authenticator-supplied state must survive a DB round-trip so that
+    tiled-style OBO handoff works across refresh_session calls."""
+    db = sqlite_session
+    principal = create_user(db, "entra", "jane@example.com")
+    payload = {"entra_access_token": "AT", "entra_refresh_token": "RT"}
+    from datetime import datetime
+
+    session = db_orm.Session(
+        principal_id=principal.id,
+        expiration_time=datetime.utcnow() + timedelta(days=1),
+        state=payload,
+    )
+    db.add(session)
+    db.commit()
+    db.refresh(session)
+
+    reloaded = db.query(db_orm.Session).filter_by(id=session.id).one()
+    assert reloaded.state == payload
+
+
+def test_session_state_defaults_to_empty_dict(sqlite_session):
+    from datetime import datetime
+
+    db = sqlite_session
+    principal = create_user(db, "internal", "alice")
+    session = db_orm.Session(
+        principal_id=principal.id,
+        expiration_time=datetime.utcnow() + timedelta(days=1),
+    )
+    db.add(session)
+    db.commit()
+    db.refresh(session)
+    # Server default is '{}' so a session created without an explicit state
+    # must not present as None to the ORM.
+    assert reloaded_state(session) == {}
+
+
+def reloaded_state(session):
+    # SQLite may return None if the server_default has not been re-selected;
+    # normalize.
+    return session.state if session.state is not None else {}
+
+
+# ---------------------------------------------------------------------------
+# Phase 2.6 - get_or_create_principal
+# ---------------------------------------------------------------------------
+
+
+def test_get_or_create_principal_creates_when_missing(sqlite_session):
+    db = sqlite_session
+    p = get_or_create_principal(db, "entra", "jane@example.com")
+    assert p is not None
+    assert p.uuid is not None
+    idents = db.query(db_orm.Identity).filter_by(id="jane@example.com", provider="entra").all()
+    assert len(idents) == 1
+    assert idents[0].principal_id == p.id
+
+
+def test_get_or_create_principal_returns_existing_and_updates_latest_login(sqlite_session):
+    db = sqlite_session
+    first = get_or_create_principal(db, "entra", "jane@example.com")
+    (first_identity,) = first.identities
+    first_login = first_identity.latest_login
+
+    # Second call must NOT create a new Principal / Identity.
+    second = get_or_create_principal(db, "entra", "jane@example.com")
+    assert second.id == first.id
+
+    db.refresh(first_identity)
+    assert first_identity.latest_login is not None
+    # It gets refreshed on every lookup, so the second timestamp must be >= first.
+    if first_login is not None:
+        assert first_identity.latest_login >= first_login
+
+    principals = db.query(db_orm.Principal).all()
+    assert len(principals) == 1
+
+
+def test_get_or_create_principal_does_not_create_a_session(sqlite_session):
+    db = sqlite_session
+    get_or_create_principal(db, "entra", "jane@example.com")
+    assert db.query(db_orm.Session).count() == 0
+
+
+# ---------------------------------------------------------------------------
+# Phase 2.4 - WebSocket first-message auth
+# ---------------------------------------------------------------------------
+
+
+def _fake_ws_with_deps(
+    *,
+    api_access_manager=None,
+    authenticators=None,
+    settings=None,
+):
+    """Build a minimal fake WebSocket whose ``app.dependency_overrides``
+    look like what build_app() installs at runtime, so
+    ``authenticate_websocket_first_message`` can retrieve them."""
+
+    from bluesky_httpserver.settings import get_settings
+    from bluesky_httpserver.utils import (
+        get_api_access_manager,
+        get_authenticators,
+    )
+
+    class _App:
+        state = MagicMock()
+        dependency_overrides = {
+            get_settings: lambda: settings,
+            get_authenticators: lambda: authenticators or {},
+            get_api_access_manager: lambda: api_access_manager,
+        }
+
+    class _WS:
+        app = _App()
+        headers = {"host": "localhost:8000"}
+        scope = {"scheme": "http", "root_path": ""}
+        query_params: dict = {}
+        cookies: dict = {}
+
+        def __init__(self):
+            # get_current_principal reads request.state.cookies_to_set for a
+            # side-effect on the HTTP path.  Provide a stub so that path does
+            # not attribute-error on the websocket route.
+            self.state = MagicMock()
+            self.state.cookies_to_set = []
+
+    return _WS()
+
+
+def test_authenticate_websocket_first_message_rejects_non_auth_frames():
+    ws = _fake_ws_with_deps(settings=MagicMock())
+    assert _auth.authenticate_websocket_first_message(ws, {"type": "ping"}) is None
+    assert _auth.authenticate_websocket_first_message(ws, "not-a-dict") is None
+    assert _auth.authenticate_websocket_first_message(ws, {"type": "auth"}) is None
+
+
+def test_authenticate_websocket_first_message_accepts_valid_api_key(sqlite_session):
+    """Feed a valid API key through the first-message handshake."""
+    from bluesky_httpserver.settings import DatabaseSettings
+
+    db = sqlite_session
+    principal = create_user(db, "internal", "alice")
+    # Generate an API key with the same machinery routes use.
+    import hashlib
+    import secrets as py_secrets
+
+    secret = py_secrets.token_bytes(4 + 32)
+    hashed = hashlib.sha256(secret).digest()
+    apikey_orm = db_orm.APIKey(
+        principal_id=principal.id,
+        first_eight=secret.hex()[:8],
+        hashed_secret=hashed,
+        scopes=["read:status"],
+    )
+    db.add(apikey_orm)
+    db.commit()
+
+    # Route the sessionmaker used by get_current_principal through our
+    # in-memory sqlite engine.
+    engine = db.get_bind()
+    from bluesky_httpserver.settings import get_sessionmaker
+
+    def _fake_sessionmaker(_db_settings):
+        return sessionmaker(bind=engine, autocommit=False, autoflush=False)
+
+    settings = MagicMock()
+    settings.database_settings = DatabaseSettings(uri="sqlite://", pool_size=None, pool_pre_ping=None)
+    settings.authentication_provider_names = ["internal"]
+    settings.secret_keys = ["hmac"]
+
+    api_access_manager = MagicMock()
+    api_access_manager.is_user_known.return_value = True
+    api_access_manager.get_user_scopes.return_value = {"read:status"}
+    api_access_manager.get_user_roles.return_value = {"user"}
+
+    authenticators = {"internal": MagicMock()}  # truthy => multi-user mode
+    ws = _fake_ws_with_deps(
+        api_access_manager=api_access_manager,
+        authenticators=authenticators,
+        settings=settings,
+    )
+
+    import bluesky_httpserver._authentication as auth_mod
+
+    saved = auth_mod.get_sessionmaker
+    auth_mod.get_sessionmaker = _fake_sessionmaker
+    try:
+        result = _auth.authenticate_websocket_first_message(
+            ws, {"type": "auth", "api_key": secret.hex()}
+        )
+    finally:
+        auth_mod.get_sessionmaker = saved
+
+    assert result is not None
+    assert result.uuid == principal.uuid
+
+
+def test_authenticate_websocket_first_message_rejects_bad_api_key(sqlite_session):
+    """A malformed (non-hex) API key must be rejected without leaking DB
+    state.  Uses the same monkey-patched sessionmaker plumbing as the
+    happy-path test so we do not accidentally exercise the real
+    get_sessionmaker(pool_size=None) code path in unit tests."""
+    from bluesky_httpserver.settings import DatabaseSettings
+
+    engine = sqlite_session.get_bind()
+
+    def _fake_sessionmaker(_db_settings):
+        return sessionmaker(bind=engine, autocommit=False, autoflush=False)
+
+    settings = MagicMock()
+    settings.database_settings = DatabaseSettings(uri="sqlite://", pool_size=5, pool_pre_ping=False)
+    settings.authentication_provider_names = ["internal"]
+    settings.secret_keys = ["hmac"]
+
+    ws = _fake_ws_with_deps(
+        api_access_manager=MagicMock(),
+        authenticators={"internal": MagicMock()},
+        settings=settings,
+    )
+
+    import bluesky_httpserver._authentication as auth_mod
+
+    saved = auth_mod.get_sessionmaker
+    auth_mod.get_sessionmaker = _fake_sessionmaker
+    try:
+        # 'not-hex' fails bytes.fromhex → HTTPException 401 inside get_current_principal.
+        assert (
+            _auth.authenticate_websocket_first_message(
+                ws, {"type": "auth", "api_key": "not-hex"}
+            )
+            is None
+        )
+    finally:
+        auth_mod.get_sessionmaker = saved