Re-run search whenever file list updates.

- Successful SSO /auth/api/validate responses are cached per credential
  and perm/renew URL for 10s, so watch websocket re-checks do not hammer
  the auth backend. A POST to the logout endpoint purges all cached
  entries for the request's credentials immediately, so logout/login
  flows are not served stale successes.
- The watch websocket now re-validates auth before each forwarded
  message and every 10s when idle (SSO and built-in sessions alike).
  When the session is gone the client gets an auth error message and
  the socket is closed, instead of streaming updates forever.
- Token-authenticated (API/share token) sockets are exempt from
  re-validation; they are checked once at handshake.

cista/api.py

@@ -5,7 +5,6 @@ import msgspec
 from mediapreview.office import is_available_cached
 from sanic import Blueprint, json
 from sanic.exceptions import BadRequest
-from sanic.log import logger
 
 from cista import __version__, auth, config, sharefs, sso, watching
 from cista.auth import (
@@ -15,7 +14,11 @@ from cista.auth import (
     list_tokens_handler,
 )
 from cista.fileio import FileServer
-from cista.util.apphelpers import websocket_wrapper
+from cista.util.apphelpers import (
+    get_watch_user_info,
+    run_auth_checked_watch,
+    websocket_wrapper,
+)
 
 bp = Blueprint("api", url_prefix="/api")
 fileserver = FileServer()
@@ -36,29 +39,7 @@ async def stop_fileserver(app):
 @bp.websocket("watch")
 @websocket_wrapper
 async def watch(req, ws):
-    # Build user info from either built-in auth or SSO
-    user_info = None
-    if sso.paskia_enabled():
-        # SSO auth: call validation to get user info (don't enforce auth in public mode)
-        try:
-            # WebSocket cannot forward Set-Cookie, so ask the auth backend not to
-            # renew the session here; renewal happens on the HTTP side instead.
-            await sso.validate_sso_request(req, renew=False)
-        except Exception as e:
-            logger.debug("watch SSO validation failed: %s", e)
-        if sso_user := getattr(req.ctx, "sso_user", None):
-            ctx = sso_user.get("ctx", {})
-            perms = ctx.get("permissions", [])
-            user_info = {
-                "username": ctx.get("user", {}).get("display_name", ""),
-                "privileged": "cista:admin" in perms,
-            }
-    elif req.ctx.user:
-        # Built-in auth: use local user database
-        user_info = {
-            "username": req.ctx.username,
-            "privileged": req.ctx.user.privileged,
-        }
+    user_info = await get_watch_user_info(req)
 
     await ws.send(
         msgspec.json.encode(
@@ -85,17 +66,7 @@ async def watch(req, ws):
             await ws.send(root)
         else:
             await ws.send(watching.format_root(sharefs.build_virtual_root(share_token)))
-        # Send updates
-        while True:
-            msg = await q.get()
-            if share_token is None or (
-                isinstance(msg, str) and msg.startswith('{"space"')
-            ):
-                await ws.send(msg)
-            else:
-                await ws.send(
-                    watching.format_root(sharefs.build_virtual_root(share_token))
-                )
+        await run_auth_checked_watch(req, ws, q, share_token)
     except RuntimeError as e:
         if str(e) == "cannot schedule new futures after shutdown":
             return  # Server shutting down, drop the WebSocket

cista/app.py

@@ -13,7 +13,7 @@ from blake3 import blake3
 from mediapreview.office import close_oo_client, log_reachable_info
 from mediapreview.pool import shutdown_preview_workers, start_preview_workers
 from sanic import Sanic, empty, raw, redirect
-from sanic.exceptions import Forbidden, NotFound
+from sanic.exceptions import Forbidden, NotFound, RequestCancelled
 from sanic.log import logger
 from setproctitle import setproctitle
 from stream_zip import ZIP_AUTO, stream_zip
@@ -35,6 +35,7 @@ from cista.sanic_logging import (
     configure_access_logging,
     configure_main_logging,
     format_access_log,
+    reset_sanic_log_levels,
 )
 from cista.sanic_logging import logger as access_logger
 from cista.util.apphelpers import handle_sanic_exception
@@ -101,6 +102,19 @@ async def forward_sso_cookies(req, res):
             res.headers.add("set-cookie", cookie)
 
 
+@app.on_response
+async def invalidate_sso_cache_on_logout(req, _res):
+    """Purge cached SSO validations after a logout request."""
+    # Convenience for logout/login flows, not a security feature: cached
+    # entries expire after 10 seconds anyway if the logout happened elsewhere.
+    if (
+        sso.paskia_enabled()
+        and req.method == "POST"
+        and req.path in {"/auth/api/logout", "/auth/logout"}
+    ):
+        sso.invalidate_validation_cache(req)
+
+
 @app.on_response
 async def persist_auth_session(req, res):
     """Persist a session cookie after successful Authorization-based auth."""
@@ -124,11 +138,28 @@ app.blueprint(fileserver.bp)
 app.exception(Exception)(handle_sanic_exception)
 
 
+@app.exception(asyncio.CancelledError)
+async def request_cancelled(req, e):
+    """Request cancelled mid-flight (client disconnect or server shutdown).
+
+    Sanic wraps this as RequestCancelled (client disconnect only) — a
+    BaseException, so the generic Exception handler above never sees it — and
+    its default handler renders a 500 error page. Report 499 for client
+    disconnects and 503 for server-side cancellation instead; no traceback
+    (quiet=True), since there is nothing to fix. The connection is usually
+    already gone.
+    """
+    if not getattr(req.ctx, "log_extra", None):
+        req.ctx.log_extra = "cancelled"
+    return empty(499 if isinstance(e, RequestCancelled) else 503)
+
+
 setproctitle("cista-main")
 
 
 @app.before_server_start
 async def main_start(app):
+    reset_sanic_log_levels()
     config.load_config()
     onlyoffice.configure()
     setproctitle(f"cista {config.config.path.name}")

cista/onlyoffice.py

@@ -20,7 +20,7 @@ def configure() -> None:
     )
 
 
-def setup_docker(confdir: Path | None = None) -> int:
+def setup_docker(confdir: Path | None = None) -> str:
     """Build and run the patched OnlyOffice Docker image (via mediapreview)."""
     if confdir is not None:
         os.environ["CISTA_HOME"] = confdir.as_posix()

cista/preview.py

@@ -11,15 +11,14 @@ from pathlib import PurePosixPath
 from urllib.parse import unquote
 from wsgiref.handlers import format_date_time
 
-import httpx
-from mediapreview.formats import OFFICE_PREVIEW_SUFFIXES
-from mediapreview.formats import expected_backend as _expected_preview_backend
-from mediapreview.office import onlyoffice_error_short_text
-from mediapreview.pool import (
-    PREVIEW_TIMEOUT,
+from mediapreview import CachedPreview, PreviewCache, is_previewable_path
+from mediapreview.exceptions import (
+    PreviewBackendError,
+    PreviewCancelledError,
     PreviewError,
-    PreviewPoolClosedError,
-    PreviewTimeoutError,
+)
+from mediapreview.formats import OFFICE_PREVIEW_SUFFIXES
+from mediapreview.pool import (
     generate_office_preview,
     run_preview,
 )
@@ -30,7 +29,6 @@ from sanic.log import logger
 from cista import auth, config, sharefs, watching
 from cista.fileio import fuid
 from cista.util.filename import sanitize
-from mediapreview import CachedPreview, PreviewCache, is_previewable_path
 
 bp = Blueprint("preview", url_prefix="/preview")
 
@@ -85,49 +83,22 @@ async def preview(req, path):
     # Generate preview
     try:
         if filepath.suffix.lower() in OFFICE_PREVIEW_SUFFIXES:
-            img, preview_resp = await asyncio.wait_for(
-                generate_office_preview(filepath, quality, maxsize, maxzoom),
-                timeout=PREVIEW_TIMEOUT,
+            img, preview_resp = await generate_office_preview(
+                filepath, quality, maxsize, maxzoom
             )
         else:
-            img, preview_resp = await asyncio.wait_for(
-                run_preview(filepath, quality, maxsize, maxzoom),
-                timeout=PREVIEW_TIMEOUT,
-            )
-    except TimeoutError:
-        req.ctx.log_extra = f"{_expected_preview_backend(filepath)} timeout"
-        return empty(503)
-    except PreviewTimeoutError as e:
-        req.ctx.log_extra = (
-            f"{(e.backend or _expected_preview_backend(filepath))} timeout"
-        )
-        return empty(503)
-    except httpx.HTTPStatusError:
-        req.ctx.log_extra = "onlyoffice N/A"
-        return empty(503)
-    except httpx.RequestError:
-        req.ctx.log_extra = "onlyoffice N/A"
-        return empty(503)
-    except RuntimeError as e:
-        detail = str(e)
-        if detail.startswith("OnlyOffice"):
-            req.ctx.log_extra = onlyoffice_error_short_text(detail)
-            return empty(503)
-        raise
-    except PreviewPoolClosedError:
-        # Server is shutting down; not an error, just a cancelled preview.
-        req.ctx.log_extra = "preview cancelled"
-        return empty(503)
+            img, preview_resp = await run_preview(filepath, quality, maxsize, maxzoom)
     except PreviewError as e:
-        detail = str(e)
-        if detail == "preview worker error" and e.stderr:
-            captured = e.stderr.strip()
-            if captured:
-                detail = captured.splitlines()[0]
-        # The worker already logged the failure (with traceback where the
-        # error occurred) — annotate the access log instead of re-logging.
-        req.ctx.log_extra = e.backend or detail
-        return empty(422)
+        # mediapreview is responsible for backend-specific diagnostics; cista only
+        # needs the backend name, a short access-log reason, and a response status.
+        if isinstance(e, PreviewCancelledError):
+            req.ctx.log_extra = e.short or "preview cancelled"
+            raise asyncio.CancelledError from e
+        status = 422 if isinstance(e, PreviewBackendError) else 503
+        req.ctx.log_extra = f"{e.backend}: {e.short}" if e.backend else e.short
+        if req.app.debug:
+            logger.warning("%s", str(e))
+        return empty(status)
     except asyncio.CancelledError:
         # Server shutdown or client disconnect: the connection is being torn
         # down, so responding is impossible — just annotate the access log.

cista/sanic_logging.py

@@ -274,11 +274,44 @@ def configure_access_logging() -> None:
 
 
 def configure_main_logging() -> None:
-    """Replace Sanic's verbose 'Main yyyy-mm-dd INFO:' prefix with emoji-only format.
+    """Replace Sanic's verbose 'Main yyyy-mm-dd INFO:' prefix with emoji-only format
 
-    Patches LOGGING_CONFIG_DEFAULTS so the formatter survives every dictConfig
-    call Sanic makes during serve_single() / serve().
+    and make sure the root logger catches unhandled loggers instead of falling back
+    to logging.lastResort (which prints a bare message with no level prefix).
+
+    Patches LOGGING_CONFIG_DEFAULTS so the formatter and root logger survive every
+    dictConfig call Sanic makes during serve_single() / serve().
     """
+    # Give the root logger a real handler so third-party warnings (e.g.
+    # mediapreview.office) are formatted with the emoji prefix instead of being
+    # printed plain by logging.lastResort.
+    root = logging.getLogger()
+    root.setLevel(logging.WARNING)
+    if not root.handlers:
+        root_handler = ReentrantSafeStreamHandler(sys.stderr)
+        root_handler.setFormatter(_EmojiFormatter())
+        root.addHandler(root_handler)
+
+    # Ensure future dictConfig calls keep a root logger so unhandled loggers still
+    # get the emoji formatter rather than falling back to logging.lastResort.
+    LOGGING_CONFIG_DEFAULTS["root"] = {
+        "level": "WARNING",
+        "handlers": ["error_console"],
+    }
+
+    # Sanic's loggers already have their own handlers; stop them from bubbling up
+    # to the root handler we just added so messages are not duplicated.
+    for name in (
+        "sanic.root",
+        "sanic.error",
+        "sanic.access",
+        "sanic.server",
+        "sanic.websockets",
+    ):
+        logging.getLogger(name).propagate = False
+        if name in LOGGING_CONFIG_DEFAULTS["loggers"]:
+            LOGGING_CONFIG_DEFAULTS["loggers"][name]["propagate"] = False
+
     for handler_name in ("console", "error_console", "access_console"):
         LOGGING_CONFIG_DEFAULTS["handlers"][handler_name]["class"] = (
             "cista.sanic_logging.ReentrantSafeStreamHandler"
@@ -286,10 +319,30 @@ def configure_main_logging() -> None:
     LOGGING_CONFIG_DEFAULTS["formatters"]["generic"] = {
         "class": "cista.sanic_logging._EmojiFormatter",
     }
-    # Silence websockets' built-in "connection closed" INFO messages; we log WS
-    # open/close ourselves in the custom access log instead.
-    logging.getLogger("websockets.server").setLevel(logging.WARNING)
+    # Sanic passes its "sanic.websockets" logger to websockets' ServerProtocol,
+    # so "connection closed" (websockets >= 17, INFO) and Sanic's own
+    # "Websocket timed out waiting for pong" (WARNING) both emit via
+    # sanic.websockets, not websockets.server. Raise it to ERROR so these
+    # routine disconnect messages are dropped while real errors still show.
+    # Patch the config defaults too, so the level survives Sanic's dictConfig.
+    LOGGING_CONFIG_DEFAULTS["loggers"]["sanic.websockets"]["level"] = "ERROR"
+    logging.getLogger("sanic.websockets").setLevel(logging.ERROR)
+    # Preview worker timeouts are already annotated in the access log extra;
+    # keep the pool's own warnings quiet so they are not logged twice.
+    logging.getLogger("mediapreview.pool").setLevel(logging.ERROR)
     # Also reformat any handlers already attached (covers the initial Sanic() call)
     for name in ("sanic.root", "sanic.error", "sanic.server", "sanic.websockets"):
         for handler in logging.getLogger(name).handlers:
             handler.setFormatter(_EmojiFormatter())
+
+
+def reset_sanic_log_levels() -> None:
+    """Force Sanic's loggers back to INFO in debug/dev mode.
+
+    Debug mode enables DEBUG on sanic.root at runtime
+    (ApplicationState.set_mode calls logger.setLevel(DEBUG)), which unleashes
+    useless noise like the 'Error Page:' content-negotiation messages. Call
+    from before_server_start so the override lands after Sanic's own setup.
+    """
+    for name in ("sanic.root", "sanic.error", "sanic.server"):
+        logging.getLogger(name).setLevel(logging.INFO)

cista/sso.py

@@ -10,8 +10,10 @@ Environment variables:
 """
 
 import asyncio
+import hashlib
 import os
 import re
+from time import time
 
 import httpx
 import websockets
@@ -62,6 +64,40 @@ async def close_client():
         _client = None
 
 
+# In-memory cache for successful SSO /auth/api/validate responses.
+# Keyed by (credential hash, validation URL) so that entries for different
+# perms/renew flags coexist and all entries for a credential can be purged
+# on logout.
+_VALIDATE_CACHE_TTL = 10
+_validate_cache: dict[tuple[str, str], tuple[float, dict]] = {}
+
+
+def _validate_credential_key(request) -> str:
+    """Return a stable key for the credential material in the request."""
+    cookie = request.headers.get("cookie", "")
+    authorization = request.headers.get("authorization", "")
+    return hashlib.sha256(f"{cookie}\x00{authorization}".encode()).hexdigest()
+
+
+def _cleanup_validate_cache() -> None:
+    """Drop expired cache entries."""
+    now = time()
+    for key, (timestamp, _) in list(_validate_cache.items()):
+        if now - timestamp >= _VALIDATE_CACHE_TTL:
+            del _validate_cache[key]
+
+
+def invalidate_validation_cache(request) -> None:
+    """Remove cached SSO validations for the credentials carried by *request*.
+
+    Called after a logout request so the next request is forced to the
+    backend instead of being served from a stale success cache.
+    """
+    credential_key = _validate_credential_key(request)
+    for key in [key for key in _validate_cache if key[0] == credential_key]:
+        del _validate_cache[key]
+
+
 async def validate_sso_request(
     request, *, perm: str = "cista:login", renew: bool = True
 ) -> dict | None:
@@ -105,6 +141,17 @@ async def validate_sso_request(
     if not renew:
         url += "&renew=0"
 
+    credential_key = _validate_credential_key(request)
+    cache_key = (credential_key, url)
+
+    cached = _validate_cache.get(cache_key)
+    if cached is not None:
+        timestamp, data = cached
+        if time() - timestamp < _VALIDATE_CACHE_TTL:
+            request.ctx.sso_user = data
+            return data
+        del _validate_cache[cache_key]
+
     try:
         response = await client.post(
             url,
@@ -121,6 +168,8 @@ async def validate_sso_request(
                 request.ctx.sso_user = {}
                 return {}
             else:
+                _cleanup_validate_cache()
+                _validate_cache[cache_key] = (time(), data)
                 return data
 
         try:

cista/util/apphelpers.py

@@ -1,14 +1,15 @@
+import asyncio
 import time
 from functools import wraps
 
 import msgspec
 import websockets.exceptions
 from sanic import errorpages
-from sanic.exceptions import SanicException
+from sanic.exceptions import SanicException, Unauthorized
 from sanic.log import logger
 from sanic.response import raw, redirect
 
-from cista import auth
+from cista import auth, config, session, sharefs, sso, watching
 from cista.protocol import ErrorMsg
 from cista.sanic_logging import log_ws_close, log_ws_open
 
@@ -101,3 +102,101 @@ def websocket_wrapper(handler):
             log_ws_close(ws_id, close_code, duration, extra=close_extra)
 
     return wrapper
+
+
+class StopError(Exception):
+    """Used internally to end a watch websocket's task group cleanly."""
+
+
+async def get_watch_user_info(request):
+    """Return the current user info for a watch websocket, re-validating auth.
+
+    Handles all three auth modes:
+      - Paskia/SSO: re-validates with the auth backend (cache-friendly)
+      - Built-in: re-reads the local session cookie from the live store
+      - Public: returns None when no session is present
+
+    Raises Unauthorized/Forbidden in non-public mode when the session is gone.
+    """
+    # Long-lived API/share tokens are validated once at handshake; re-checking
+    # them on every message would add unnecessary backend calls.
+    if getattr(request.ctx, "auth_token", None) is not None:
+        return None
+
+    if sso.paskia_enabled():
+        try:
+            await sso.validate_sso_request(request, renew=False)
+        except SanicException:
+            if config.config.public:
+                return None
+            raise
+        sso_user = getattr(request.ctx, "sso_user", None) or {}
+        if sso_user:
+            ctx = sso_user.get("ctx", {})
+            perms = ctx.get("permissions", [])
+            return {
+                "username": ctx.get("user", {}).get("display_name", ""),
+                "privileged": "cista:admin" in perms,
+            }
+        return None
+
+    s = session.get(request)
+    if s:
+        user = config.config.users.get(s.get("username"))
+        if user:
+            return {"username": s["username"], "privileged": user.privileged}
+
+    if config.config.public:
+        return None
+
+    raise Unauthorized("Login required", "cookie", quiet=True)
+
+
+async def _check_watch_auth_or_stop(request, ws) -> None:
+    """Re-validate watch auth; on failure send an error and raise StopError."""
+    try:
+        await get_watch_user_info(request)
+    except SanicException as exc:
+        # Match the error format used by websocket_wrapper
+        message = f"⚠️ {str(exc) or 'Authentication error'}"
+        await asend(
+            ws,
+            ErrorMsg(
+                {"code": exc.status_code, "message": message, **(exc.context or {})}
+            ),
+        )
+        raise StopError from None
+
+
+async def run_auth_checked_watch(request, ws, queue, share_token) -> None:
+    """Run the watch websocket loop with per-message and periodic auth checks.
+
+    Messages are forwarded from *queue* to *ws*. Auth is re-checked before each
+    message (hitting the SSO cache in the common case) and every 10 seconds when
+    idle, so a session invalidated on the backend does not stay open forever.
+    """
+
+    async def consume() -> None:
+        while True:
+            item = await queue.get()
+            await _check_watch_auth_or_stop(request, ws)
+            if share_token is None or (
+                isinstance(item, str) and item.startswith('{"space"')
+            ):
+                await ws.send(item)
+            else:
+                await ws.send(
+                    watching.format_root(sharefs.build_virtual_root(share_token))
+                )
+
+    async def idle_checker() -> None:
+        while True:
+            await asyncio.sleep(10)
+            await _check_watch_auth_or_stop(request, ws)
+
+    try:
+        async with asyncio.TaskGroup() as tg:
+            tg.create_task(consume())
+            tg.create_task(idle_checker())
+    except* StopError:
+        pass

frontend/src/components/HeaderMain.vue

@@ -28,9 +28,9 @@
     </template>
     <div v-if="!props.editorMode && showSortHints" class="sort-hints">
       <span class="sort-label">Order</span>
-      <span class="keycap">1</span>
-      <span class="keycap">2</span>
-      <span class="keycap">3</span>
+      <button type="button" class="keycap" aria-label="Alphabetical order" @click="store.sort('name')">1</button>
+      <button type="button" class="keycap" aria-label="Newest first" @click="store.sort('modified')">2</button>
+      <button type="button" class="keycap" aria-label="Largest first" @click="store.sort('size')">3</button>
     </div>
     <SvgButton
       v-if="props.editorMode"
@@ -314,6 +314,14 @@ onUnmounted(() => {
   border-radius: 0.3em;
   padding: 0 0.45em;
   line-height: 1.4;
+  cursor: pointer;
+  transition: all 0.2s ease;
+}
+.keycap:hover,
+.keycap:focus {
+  background: #e6e6e6;
+  border-color: #aaa;
+  transform: scale(1.05);
 }
 @media screen and (min-width: 800px) {
   .sort-hints {

frontend/src/stores/main.ts

@@ -79,6 +79,7 @@ export const useMainStore = defineStore('main', {
     connected: false,
     authInProgress: false,
     cursor: '' as string,
+    lastSearchLoc: '' as string,
     server: {} as Record<string, any> & {
       public?: boolean
       paskia?: boolean
@@ -159,6 +160,10 @@ export const useMainStore = defineStore('main', {
       this.docVersion++
       // Sync documents to search worker
       this.syncSearchWorker()
+      // Re-run the current search against the updated file list
+      if (this.query) {
+        this.search(this.query, this.lastSearchLoc)
+      }
     },
     /** Patch aspect ratios on existing docs from a server ar update message */
     updateAr(arMap: Record<string, number>) {
@@ -268,6 +273,7 @@ export const useMainStore = defineStore('main', {
 
       // Update query immediately so watchers know we're handling this
       this.query = query
+      this.lastSearchLoc = loc
 
       // Cancel pending timers
       if (loadingTimer) {

pyproject.toml

@@ -33,7 +33,7 @@ dependencies = [
     "html5tagger>=1.3.0",
     "httpx>=0.28.0",
     "inotify>=0.2.12",
-    "mediapreview[standard]",
+    "mediapreview[standard]>=0.2.0",
     "msgspec>=0.19.0",
     "natsort>=8.4.0",
     "numpy>=2.3.2",

tests/test_sso_cache.py

@@ -0,0 +1,142 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+from sanic.exceptions import Unauthorized
+
+from cista import sso
+
+
+def _make_request(cookie: str = "", authorization: str = ""):
+    req = SimpleNamespace()
+    req.headers = {}
+    if cookie:
+        req.headers["cookie"] = cookie
+    if authorization:
+        req.headers["authorization"] = authorization
+    req.client_ip = "127.0.0.1"
+    req.host = "test.local"
+    req.scheme = "http"
+    req.ctx = SimpleNamespace()
+    return req
+
+
+@pytest.fixture(autouse=True)
+def _reset_sso_cache_and_client(monkeypatch):
+    """Clear the SSO validation cache and shared client between tests."""
+    sso._validate_cache.clear()
+    sso._client = None
+    monkeypatch.setenv("PASKIA_BACKEND_URL", "http://test-paskia.local")
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "http://test-paskia.local")
+    yield
+    sso._validate_cache.clear()
+
+
+@pytest.fixture
+def mock_client(monkeypatch):
+    client = AsyncMock()
+    client.is_closed = False
+    client.headers = {}
+    monkeypatch.setattr(sso, "_client", client)
+    return client
+
+
+@pytest.mark.asyncio
+async def test_validate_sso_request_caches_successful_responses(mock_client):
+    req = _make_request(cookie="session=abc123")
+    mock_client.post.return_value = httpx.Response(200, json={"user": "alice"})
+
+    data1 = await sso.validate_sso_request(req)
+    data2 = await sso.validate_sso_request(req)
+
+    assert data1 == {"user": "alice"}
+    assert data2 == data1
+    assert mock_client.post.call_count == 1
+    assert req.ctx.sso_user == {"user": "alice"}
+
+
+@pytest.mark.asyncio
+async def test_validate_sso_request_does_not_cache_errors(mock_client):
+    req = _make_request(cookie="session=bad")
+    mock_client.post.return_value = httpx.Response(401, json={"detail": "nope"})
+
+    with pytest.raises(Unauthorized):
+        await sso.validate_sso_request(req)
+    with pytest.raises(Unauthorized):
+        await sso.validate_sso_request(req)
+
+    assert mock_client.post.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_validate_sso_request_cache_is_per_credential(mock_client):
+    req_alice = _make_request(cookie="session=alice")
+    req_bob = _make_request(cookie="session=bob")
+    responses = {
+        "alice": httpx.Response(200, json={"user": "alice"}),
+        "bob": httpx.Response(200, json={"user": "bob"}),
+    }
+
+    def side_effect(*args, **kwargs):
+        cookie = kwargs.get("headers", {}).get("cookie", "")
+        if "alice" in cookie:
+            return responses["alice"]
+        return responses["bob"]
+
+    mock_client.post.side_effect = side_effect
+
+    assert await sso.validate_sso_request(req_alice) == {"user": "alice"}
+    assert await sso.validate_sso_request(req_bob) == {"user": "bob"}
+    assert await sso.validate_sso_request(req_alice) == {"user": "alice"}
+
+    assert mock_client.post.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_validate_sso_request_cache_is_per_permission(mock_client):
+    req = _make_request(cookie="session=abc123")
+    mock_client.post.return_value = httpx.Response(200, json={"user": "alice"})
+
+    await sso.validate_sso_request(req, perm="cista:login")
+    await sso.validate_sso_request(req, perm="cista:admin")
+
+    assert mock_client.post.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_invalidate_validation_cache_forces_backend_call(mock_client):
+    req = _make_request(cookie="session=abc123")
+    mock_client.post.return_value = httpx.Response(200, json={"user": "alice"})
+
+    await sso.validate_sso_request(req)
+    sso.invalidate_validation_cache(req)
+    await sso.validate_sso_request(req)
+
+    assert mock_client.post.call_count == 2
+
+
+@pytest.mark.asyncio
+async def test_invalidate_validation_cache_only_affects_same_credentials(mock_client):
+    alice = _make_request(cookie="session=alice")
+    bob = _make_request(cookie="session=bob")
+    responses = {
+        "alice": httpx.Response(200, json={"user": "alice"}),
+        "bob": httpx.Response(200, json={"user": "bob"}),
+    }
+
+    def side_effect(*args, **kwargs):
+        cookie = kwargs.get("headers", {}).get("cookie", "")
+        return responses["alice"] if "alice" in cookie else responses["bob"]
+
+    mock_client.post.side_effect = side_effect
+
+    await sso.validate_sso_request(alice)
+    await sso.validate_sso_request(bob)
+    sso.invalidate_validation_cache(alice)
+
+    assert await sso.validate_sso_request(alice) == {"user": "alice"}
+    assert await sso.validate_sso_request(bob) == {"user": "bob"}
+
+    # Alice is re-fetched; bob is still cached.
+    assert mock_client.post.call_count == 3

tests/test_watch_auth.py

@@ -0,0 +1,211 @@
+import asyncio
+import contextlib
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from sanic.exceptions import Unauthorized
+
+from cista import auth, config, session, sso
+from cista.util.apphelpers import (
+    get_watch_user_info,
+    run_auth_checked_watch,
+)
+
+
+def _make_request(cookie: str = "", auth_token=None):
+    req = SimpleNamespace()
+    req.headers = {}
+    req.cookies = {}
+    if cookie:
+        req.headers["cookie"] = cookie
+        for part in cookie.split(";"):
+            k, _, v = part.strip().partition("=")
+            req.cookies[k] = v
+    req.ctx = SimpleNamespace()
+    if auth_token:
+        req.ctx.auth_token = auth_token
+    return req
+
+
+@pytest.fixture(autouse=True)
+def _reset(tmp_path, monkeypatch):
+    alice = config.User()
+    auth.set_password(alice, "secret")
+    admin = config.User(privileged=True)
+    auth.set_password(admin, "admin-secret")
+    config.config = config.Config(
+        path=tmp_path,
+        listen=":0",
+        public=False,
+        users={"alice": alice, "admin": admin},
+    )
+    session._sessions.clear()
+    sso._validate_cache.clear()
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "")
+    yield
+    session._sessions.clear()
+    sso._validate_cache.clear()
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_builtin_valid_session():
+    token = "valid-token"
+    session.put(token, "alice")
+    req = _make_request(cookie=f"cista={token}")
+
+    info = await get_watch_user_info(req)
+
+    assert info == {"username": "alice", "privileged": False}
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_builtin_admin():
+    token = "admin-token"
+    session.put(token, "admin")
+    req = _make_request(cookie=f"cista={token}")
+
+    info = await get_watch_user_info(req)
+
+    assert info == {"username": "admin", "privileged": True}
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_builtin_invalid_session_raises():
+    req = _make_request(cookie="cista=bad-token")
+
+    with pytest.raises(Unauthorized):
+        await get_watch_user_info(req)
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_builtin_public_no_session():
+    config.config.public = True
+    req = _make_request()
+
+    info = await get_watch_user_info(req)
+
+    assert info is None
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_sso_valid(monkeypatch):
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "http://test-paskia.local")
+
+    async def mock_validate(request, *, renew=True):
+        request.ctx.sso_user = {
+            "ctx": {
+                "user": {"display_name": "alice"},
+                "permissions": ["cista:login", "cista:admin"],
+            }
+        }
+
+    monkeypatch.setattr(sso, "validate_sso_request", mock_validate)
+    req = _make_request(cookie="session=abc")
+
+    info = await get_watch_user_info(req)
+
+    assert info == {"username": "alice", "privileged": True}
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_sso_nonpublic_invalid_raises(monkeypatch):
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "http://test-paskia.local")
+
+    async def mock_validate(request, *, renew=True):
+        raise Unauthorized("Session expired", quiet=True)
+
+    monkeypatch.setattr(sso, "validate_sso_request", mock_validate)
+    req = _make_request(cookie="session=abc")
+
+    with pytest.raises(Unauthorized):
+        await get_watch_user_info(req)
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_sso_public_invalid_returns_none(monkeypatch):
+    monkeypatch.setattr(sso, "PASKIA_BACKEND_URL", "http://test-paskia.local")
+    config.config.public = True
+
+    async def mock_validate(request, *, renew=True):
+        raise Unauthorized("Session expired", quiet=True)
+
+    monkeypatch.setattr(sso, "validate_sso_request", mock_validate)
+    req = _make_request(cookie="session=abc")
+
+    info = await get_watch_user_info(req)
+
+    assert info is None
+
+
+@pytest.mark.asyncio
+async def test_run_auth_checked_watch_forwards_messages_while_valid():
+    token = "valid-token"
+    session.put(token, "alice")
+    req = _make_request(cookie=f"cista={token}")
+    ws = AsyncMock()
+    q = asyncio.Queue()
+
+    async def producer():
+        await q.put('{"space":{}}')
+        await q.put('{"update":[]}')
+        # Keep consumer alive briefly, then invalidate.
+        await asyncio.sleep(0.05)
+        session._sessions.pop(token, None)
+        await q.put('{"update":[]}')
+
+    await asyncio.wait_for(
+        asyncio.gather(producer(), run_auth_checked_watch(req, ws, q, None)),
+        timeout=1.0,
+    )
+
+    calls = [c.args[0] for c in ws.send.call_args_list]
+    assert calls[0] == '{"space":{}}'
+    assert calls[1] == '{"update":[]}'
+    assert '"error"' in calls[2]
+    assert len(calls) == 3
+
+
+@pytest.mark.asyncio
+async def test_get_watch_user_info_token_auth_skips_revalidation():
+    """Token-based auth is considered valid without re-checking the token."""
+    token_id = "api-token"
+    config.config.tokens[token_id] = config.Token(
+        key=token_id, username="alice", kind="api", mode="rw"
+    )
+    req = _make_request(auth_token=config.config.tokens[token_id])
+
+    info = await get_watch_user_info(req)
+
+    assert info is None
+
+
+@pytest.mark.asyncio
+async def test_run_auth_checked_watch_token_auth_does_not_send_errors():
+    """Token-based sockets keep forwarding messages without re-validating."""
+    token_id = "api-token"
+    config.config.tokens[token_id] = config.Token(
+        key=token_id, username="alice", kind="api", mode="rw"
+    )
+    req = _make_request(auth_token=config.config.tokens[token_id])
+    ws = AsyncMock()
+    q = asyncio.Queue()
+
+    async def producer():
+        await q.put('{"space":{}}')
+        await q.put('{"update":[]}')
+        # Deleting the token should not affect the already-open websocket.
+        await asyncio.sleep(0.05)
+        del config.config.tokens[token_id]
+
+    runner = asyncio.create_task(run_auth_checked_watch(req, ws, q, None))
+    await asyncio.wait_for(producer(), timeout=1.0)
+    await asyncio.sleep(0.05)
+    runner.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await runner
+
+    calls = [c.args[0] for c in ws.send.call_args_list]
+    assert calls[0] == '{"space":{}}'
+    assert calls[1] == '{"update":[]}'
+    assert not any('"error"' in c for c in calls)