src/dashboard_auth.py

@@ -1,5 +1,6 @@
 """Signed, short-lived single-operator sessions for the dashboard boundary."""
 
+import asyncio
 import base64
 import hashlib
 import hmac
@@ -139,8 +140,8 @@ def revoke_session(session: Session) -> None:
     _session_store().revoke(session.session_id)
 
 
-def request_session(request: Request) -> Session | None:
+async def request_session(request: Request) -> Session | None:
-    return verify_session(request.cookies.get(SESSION_COOKIE))
+    return await asyncio.to_thread(verify_session, request.cookies.get(SESSION_COOKIE))
 
 
 def cookie_path(request: Request) -> str:

src/main.py

@@ -470,14 +470,16 @@ async def require_operator_session(request: Request, call_next):
         or path.startswith("/static/")
         or (path == "/api/v1/session" and request.method == "POST")
     )
-    try:
+    session = None
-        session = dashboard_auth.request_session(request)
+    if not public:
-    except dashboard_auth.SessionStoreError:
+        try:
-        return JSONResponse(
+            session = await dashboard_auth.request_session(request)
-            {"detail": "Session registry is temporarily unavailable"},
+        except dashboard_auth.SessionStoreError:
-            status_code=503,
+            return JSONResponse(
-            headers={"Cache-Control": "no-store"},
+                {"detail": "Session registry is temporarily unavailable"},
-        )
+                status_code=503,
+                headers={"Cache-Control": "no-store"},
+            )
     if not public and session is None:
         if path.startswith("/api/"):
             return JSONResponse(
@@ -615,7 +617,7 @@ async def sign_in(payload: DashboardSignIn, request: Request, response: Response
 
 @app.get("/api/v1/session")
 async def session_status(request: Request):
-    session = dashboard_auth.request_session(request)
+    session = request.state.dashboard_session
     return {
         "authenticated": session is not None,
         "csrf_token": session.csrf if session is not None else "",

src/session_store.py

@@ -26,25 +26,35 @@ class SessionStore:
     def _digest(session_id: str) -> str:
         return hashlib.sha256(session_id.encode()).hexdigest()
 
-    def _connect(self) -> sqlite3.Connection:
+    def _connect(self, *, initialize: bool = False) -> sqlite3.Connection:
         try:
-            self.path.parent.mkdir(parents=True, exist_ok=True)
+            if initialize:
-            connection = sqlite3.connect(self.path, timeout=self.lock_timeout_seconds)
+                self.path.parent.mkdir(parents=True, exist_ok=True)
-            connection.execute(
+                connection = sqlite3.connect(
-                """
+                    self.path, timeout=self.lock_timeout_seconds
-                CREATE TABLE IF NOT EXISTS active_sessions (
+                )
-                    session_hash TEXT PRIMARY KEY,
+            else:
-                    expires_at INTEGER NOT NULL
+                connection = sqlite3.connect(
+                    f"{self.path.resolve().as_uri()}?mode=rw",
+                    timeout=self.lock_timeout_seconds,
+                    uri=True,
+                )
+            if initialize:
+                connection.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS active_sessions (
+                        session_hash TEXT PRIMARY KEY,
+                        expires_at INTEGER NOT NULL
+                    )
+                    """
                 )
-                """
-            )
             return connection
         except (OSError, sqlite3.Error) as exc:
             raise SessionStoreError("Session registry is temporarily unavailable") from exc
 
     def activate(self, session_id: str, expires_at: int) -> None:
         try:
-            with self._connect() as connection:
+            with self._connect(initialize=True) as connection:
                 connection.execute(
                     "DELETE FROM active_sessions WHERE expires_at <= ?", (int(self.clock()),)
                 )
@@ -59,7 +69,6 @@ class SessionStore:
         now = int(self.clock())
         try:
             with self._connect() as connection:
-                connection.execute("DELETE FROM active_sessions WHERE expires_at <= ?", (now,))
                 row = connection.execute(
                     "SELECT expires_at FROM active_sessions WHERE session_hash = ?",
                     (self._digest(session_id),),

tests/test_dashboard_auth.py

@@ -1,3 +1,6 @@
+import asyncio
+import time
+
 import httpx
 import pytest
 
@@ -175,6 +178,30 @@ async def test_authenticated_get_reaches_private_api(access_control, monkeypatch
     assert response.json()["user"]["login"] == "timmy"
 
 
+@pytest.mark.anyio
+async def test_session_status_reuses_the_middleware_validation(access_control, monkeypatch):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session", json={"access_token": "correct horse battery staple"}
+        )
+        original_store = main.dashboard_auth._session_store()
+        lookups = 0
+
+        class CountingStore:
+            def is_active(self, session_id, expires_at):
+                nonlocal lookups
+                lookups += 1
+                return original_store.is_active(session_id, expires_at)
+
+        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: CountingStore())
+        response = await client.get("/api/v1/session")
+
+    assert response.status_code == 200
+    assert response.json()["authenticated"] is True
+    assert lookups == 1
+
+
 @pytest.mark.anyio
 async def test_authenticated_session_status_exposes_only_csrf_proof(access_control):
     transport = httpx.ASGITransport(app=main.app)
@@ -292,6 +319,31 @@ async def test_logout_revokes_a_captured_cookie_before_gitea(access_control, mon
     assert calls == 1
 
 
+@pytest.mark.anyio
+async def test_session_registry_latency_does_not_block_the_event_loop(access_control, monkeypatch):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session", json={"access_token": "correct horse battery staple"}
+        )
+
+        class SlowStore:
+            def is_active(self, session_id, expires_at):
+                time.sleep(0.15)
+                return True
+
+        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: SlowStore())
+        private_request = asyncio.create_task(client.get("/api/v1/session"))
+        await asyncio.sleep(0)
+        started = time.perf_counter()
+        await asyncio.sleep(0.01)
+        heartbeat_elapsed = time.perf_counter() - started
+        response = await private_request
+
+    assert response.status_code == 200
+    assert heartbeat_elapsed < 0.08
+
+
 @pytest.mark.anyio
 async def test_session_registry_read_failure_fails_closed_before_gitea(
     access_control, monkeypatch
@@ -375,6 +427,27 @@ async def test_logout_registry_failure_does_not_claim_revocation(access_control,
     assert "database path" not in response.text
 
 
+@pytest.mark.anyio
+async def test_public_routes_skip_session_registry_validation(access_control, monkeypatch):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        signed_in = await client.post(
+            "/api/v1/session", json={"access_token": "correct horse battery staple"}
+        )
+
+        class BrokenStore:
+            def is_active(self, session_id, expires_at):
+                raise SessionStoreError("public routes must not read the registry")
+
+        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
+        login = await client.get("/login")
+        manifest = await client.get("/manifest.webmanifest")
+        static = await client.get("/static/session.js")
+
+    assert signed_in.status_code == 200
+    assert [login.status_code, manifest.status_code, static.status_code] == [200, 200, 200]
+
+
 @pytest.mark.anyio
 async def test_public_routes_remain_available_and_readiness_hides_identity(access_control, monkeypatch):
     async def user():

tests/test_session_store.py

@@ -1,6 +1,9 @@
 import sqlite3
 
-from src.session_store import SessionStore
+import pytest
+
+from src import session_store
+from src.session_store import SessionStore, SessionStoreError
 
 
 def test_revocation_is_durable_and_scoped_to_one_session(tmp_path):
@@ -19,7 +22,34 @@ def test_revocation_is_durable_and_scoped_to_one_session(tmp_path):
     assert b"second-session-secret" not in database.read_bytes()
 
 
-def test_expired_sessions_are_rejected_and_pruned(tmp_path):
+def test_validation_does_not_create_a_missing_registry(tmp_path):
+    database = tmp_path / "sessions.sqlite3"
+    store = SessionStore(database, clock=lambda: 1_000.0)
+
+    with pytest.raises(SessionStoreError):
+        store.is_active("unknown-session", 2_000)
+
+    assert database.exists() is False
+
+
+def test_validation_executes_only_a_read_query(tmp_path, monkeypatch):
+    store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
+    store.activate("active-session", 2_000)
+    statements = []
+    connect = sqlite3.connect
+
+    def traced_connect(*args, **kwargs):
+        connection = connect(*args, **kwargs)
+        connection.set_trace_callback(statements.append)
+        return connection
+
+    monkeypatch.setattr(session_store.sqlite3, "connect", traced_connect)
+
+    assert store.is_active("active-session", 2_000) is True
+    assert [statement.split()[0].upper() for statement in statements] == ["SELECT"]
+
+
+def test_expired_sessions_are_rejected_without_writing_during_validation(tmp_path):
     now = [1_000.0]
     store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
     store.activate("expiring-session", 1_001)
@@ -28,4 +58,4 @@ def test_expired_sessions_are_rejected_and_pruned(tmp_path):
 
     assert store.is_active("expiring-session", 1_001) is False
     with sqlite3.connect(store.path) as connection:
-        assert connection.execute("SELECT COUNT(*) FROM active_sessions").fetchone() == (0,)
+        assert connection.execute("SELECT COUNT(*) FROM active_sessions").fetchone() == (1,)