diff --git a/CHANGELOG.md b/CHANGELOG.md
index 52998b4aa..8cf93ff3a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -81,6 +81,21 @@ All notable changes to this project are documented in this file. The format is b
 
 ### Fixed
 
+- Users password creation, forgot-password fingerprints, reset verification and
+  password updates offload hashing to worker threads, keeping unrelated async
+  requests responsive during registration and bulk invites. Password policy,
+  single-use reset tokens and atomic credential revocation are preserved.
+- Admin user state totals and role options use a 30-second per-worker cache.
+  A committed user, role or membership write evicts this worker's entry; the
+  admin service's write paths (create, invite, delete, disable, enable, mark
+  verified, set roles) also publish `users.admin_summary` once after commit so
+  other workers evict theirs. Other writes (logins, lockout counters, password
+  changes) publish nothing and rely on the TTL.
+  List DTOs skip redundant validation of trusted database values; role options
+  select only the required columns, and managers share one password helper.
+- Users: `on_commit` callbacks queued before a fastapi-users adapter commit
+  (user create, update, delete through the manager) now run after that commit;
+  they used to be discarded when the request session was finalized.
 - Include module `ui/` sources in wheel-installed Tailwind scanning and the asset manifest; retain legacy `components/` support.
 - Background tasks: the executions list (when empty) and the Workers page no
   longer wait out the broker's ~1 s inspect timeout before rendering. The
diff --git a/docs/modules/users.md b/docs/modules/users.md
index 360f187d1..089575985 100644
--- a/docs/modules/users.md
+++ b/docs/modules/users.md
@@ -243,6 +243,28 @@ Everything else is DB-backed (initial values are pydantic defaults; edit under U
 | `Profile` | `/users/me` | `user` | `USER_DROPDOWN` | — | `990` | _logged-in_ |
 | `Logout` | `/users/logout` (POST) | `log-out` | `USER_DROPDOWN` | — | `999` | _logged-in_ |
 
+## Performance
+
+Password hashing and verification for local login, account creation (including
+registration and bulk invitations), password changes and reset fingerprints run
+in worker threads. Database operations, password policy and lifecycle hooks stay
+on the event loop; password resets still revoke credentials atomically.
+
+Admin user state totals and role options (including membership counts) are
+cached per worker for 30 seconds. A committed transaction that wrote a user,
+role or membership through the ORM evicts this worker's entry. Other workers
+hear about a change only through an explicit `users.admin_summary` publish from
+the admin service's write paths (create, invite, delete, disable, enable, mark
+verified, set roles), sent once after the commit and never on rollback. Other
+writes (logins, failed-login counters, password changes, session revocations,
+role edits made outside the users admin service) publish nothing, so another
+worker may show old totals until the 30-second TTL expires; the same holds
+without the Redis invalidation transport or when a message is lost.
+These summaries do not control permissions. Filtered user rows and pagination
+counts remain live queries. The accepted contains-search scan is unchanged.
+See [Users performance measurements](../perf/2026-11-05-users.md) for the local
+before/after results and reproduction command.
+
 ## Password reset and credential revocation
 
 A successful password reset, including one reached through an admin-generated
diff --git a/docs/perf/2026-11-05-users.md b/docs/perf/2026-11-05-users.md
new file mode 100644
index 000000000..3640783fb
--- /dev/null
+++ b/docs/perf/2026-11-05-users.md
@@ -0,0 +1,45 @@
+# Users performance measurements
+
+Measured locally on 2026-11-05, using production password costs
+(`SM_TEST_REAL_PASSWORD_HASHING=1`) and the installed fastapi-users 15.0.5.
+The baseline is the working tree before the password offload change.
+
+A concurrent heartbeat sleeps for 5 ms and records the excess scheduling delay
+while the real manager/service performs the operation. The bulk case calls
+`UserService.invite` sequentially for 100 distinct accounts; it does not include
+HTTP parsing or email delivery. Fixtures use SQLite, so these measurements do
+not reproduce the review's Postgres/10k-user workload.
+
+| Operation | Before: worst loop lag | After: worst loop lag | Before / after elapsed |
+|---|---:|---:|---:|
+| 100 sequential invitations | 254.1 ms | 5.1 ms | 27.273 / 27.387 s |
+| Forgot-password fingerprint | 237.2 ms | 4.4 ms | 0.267 / 0.256 s |
+
+Offloading keeps the event loop responsive; it does not reduce the underlying
+password hashing work. These are single-run observations, not request latency
+percentiles. This sandbox did not reliably wake its event loop after cross-thread
+completions (even a standalone aiosqlite connection hung). Both runs used the
+same temporary test-runner shim that capped selector waits at 1 ms; application
+code and hashing costs were unchanged by the shim.
+
+Run the opt-in measurements on a normal local environment:
+
+```sh
+SM_TEST_REAL_PASSWORD_HASHING=1 uv run pytest \
+  modules/users/tests/test_users_password_offload.py \
+  -k real_hashing -q -s -p no:cacheprovider
+```
+
+The ordinary targeted regressions check crypto thread identity independently
+of timing, reset-token reuse rejection, and that database execution stays on
+the event-loop thread. Existing credential-revocation and password-policy tests
+cover the reset's transaction semantics.
+
+For admin summaries, regressions record SQL execution and require **zero new
+queries** on repeated state-count and role-option reads within the TTL. They
+also cover expiry, local eviction when a user/role/membership write commits,
+one explicit `users.admin_summary` publish per admin write method after commit
+(none on rollback, none for login or lockout writes), remote invalidation,
+invalidation racing a cache fill, and role projections that do not instantiate
+ORM rows. DTO construction is checked without invoking SQLModel input
+validation.
diff --git a/modules/users/tests/test_users_admin_summary_cache.py b/modules/users/tests/test_users_admin_summary_cache.py
new file mode 100644
index 000000000..c30b06451
--- /dev/null
+++ b/modules/users/tests/test_users_admin_summary_cache.py
@@ -0,0 +1,264 @@
+"""Admin summaries avoid repeated scans, forget committed writes and publish explicitly."""
+
+import uuid
+from datetime import UTC, datetime
+from unittest.mock import patch
+
+import pytest
+from fastapi_users.password import PasswordHelper
+from simple_module_core.invalidation import Invalidation
+from simple_module_db.callbacks import register_on_commit
+from simple_module_db.transaction import finalize_session, rollback_session
+from sqlalchemy import event, select
+from users.admin import summary_cache
+from users.admin.service import UserService
+from users.contracts.schemas import UserCreate, UserListItem
+from users.db_adapter import UserDatabaseWithRoles
+from users.manager import UserManager
+from users.models import Role, User, UserRole
+from users.roles_cache import refresh_roles_cache
+
+
+def service(db, app):
+    manager = UserManager(
+        UserDatabaseWithRoles(db, User), app.state.users.mailer, app.state.users.settings
+    )
+    return UserService(db, manager, app)
+
+
+def _record(app):
+    messages = []
+    app.state.sm.invalidation.subscribe(summary_cache.CHANNEL, messages.append)
+    return messages
+
+
+async def _make_user(app, email="target@example.com", **fields) -> uuid.UUID:
+    async with app.state.sm.db.session_factory() as db:
+        user = User(email=email, hashed_password=PasswordHelper().hash("SecurePass1!"), **fields)
+        db.add(user)
+        await finalize_session(db)
+        return user.id
+
+
+@pytest.mark.parametrize("method", ["count_user_states", "list_roles"])
+async def test_summary_cache_hit_and_expiry(users_app, monkeypatch, method):
+    queries = []
+    event.listen(
+        users_app.state.sm.db.engine.sync_engine,
+        "before_cursor_execute",
+        lambda *args: queries.append(args[2]),
+    )
+    async with users_app.state.sm.db.session_factory() as db:
+        get = getattr(service(db, users_app), method)
+        first = await get()
+        queries.clear()
+        assert await get() == first
+        assert queries == []
+        # Values handed to callers must not let them corrupt the cached result.
+        first.clear()
+        assert await get()
+        monkeypatch.setattr(summary_cache.time, "monotonic", lambda: 10**12)
+        await get()
+        assert queries
+
+
+@pytest.mark.parametrize(
+    "write", ["create", "verify", "invite", "disable", "delete", "role", "membership"]
+)
+async def test_local_orm_writes_clear_this_workers_cache_on_commit(users_app, write):
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        user = User(
+            email="cached@example.com", hashed_password=PasswordHelper().hash("SecurePass1!")
+        )
+        db.add(user)
+        await finalize_session(db)
+        before = await svc.count_user_states()
+        roles_before = await svc.list_roles()
+        if write == "create":
+            db.add(User(email="added@example.com", hashed_password="unused"))
+        elif write == "verify":
+            user.is_verified = True
+        elif write == "invite":
+            user.invited_at = datetime.now(UTC)
+        elif write == "disable":
+            user.is_active = False
+        elif write == "delete":
+            await db.delete(user)
+        elif write == "role":
+            db.add(Role(name="reviewer"))
+        else:
+            role = (await db.execute(select(Role).where(Role.name == "user"))).scalar_one()
+            db.add(UserRole(user_id=user.id, role_id=role.id))
+        await finalize_session(db)
+        if write in ("role", "membership"):
+            assert await svc.list_roles() != roles_before
+        else:
+            assert await svc.count_user_states() != before
+
+
+async def test_rollback_and_remote_invalidation(users_app):
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        before = await svc.count_user_states()
+        db.add(User(email="rolledback@example.com", hashed_password="unused"))
+        await db.flush()
+        await rollback_session(db)
+        # A rollback must preserve a warm cache.
+        with patch.object(summary_cache, "load_counts", side_effect=AssertionError("cache miss")):
+            assert await svc.count_user_states() == before
+        await users_app.state.sm.invalidation.deliver(
+            Invalidation(summary_cache.CHANNEL, origin="other-worker").to_wire()
+        )
+        # Invalidate remotely even when no local write occurred.
+        with (
+            patch.object(summary_cache, "load_counts", side_effect=RuntimeError("reloaded")),
+            pytest.raises(RuntimeError, match="reloaded"),
+        ):
+            await svc.count_user_states()
+
+
+async def _call(svc, method, user_id):
+    if method == "create_user":
+        await svc.create_user("new@example.com", "SecurePass1!", None, ["user"], created_by=None)
+    elif method == "invite":
+        await svc.invite("invited@example.com", None, ["user"])
+    elif method == "set_roles":
+        await svc.set_roles(user_id, [])
+    else:
+        await getattr(svc, method)(user_id)
+
+
+_WRITES = ["create_user", "invite", "delete_user", "disable", "enable", "mark_verified"]
+
+
+@pytest.mark.parametrize("method", [*_WRITES, "set_roles"])
+async def test_admin_writes_publish_once_after_commit(users_app, method):
+    user_id = await _make_user(users_app, is_active=method != "enable")
+    messages = _record(users_app)
+    async with users_app.state.sm.db.session_factory() as db:
+        await _call(service(db, users_app), method, user_id)
+        assert messages == []
+        await finalize_session(db)
+    assert len(messages) == 1
+
+
+@pytest.mark.parametrize("method", [*_WRITES, "set_roles"])
+async def test_admin_writes_publish_nothing_on_rollback(users_app, method):
+    user_id = await _make_user(users_app, is_active=method != "enable")
+    messages = _record(users_app)
+    async with users_app.state.sm.db.session_factory() as db:
+        await _call(service(db, users_app), method, user_id)
+        await rollback_session(db)
+    assert messages == []
+
+
+async def test_login_and_lockout_writes_publish_nothing(users_app, anon_client):
+    messages = _record(users_app)
+    for _ in range(3):
+        resp = await anon_client.post(
+            "/api/users/auth/login",
+            data={"username": "admin@example.com", "password": "wrong-password"},
+        )
+        assert resp.status_code == 400, resp.text
+    resp = await anon_client.post(
+        "/api/users/auth/login",
+        data={"username": "admin@example.com", "password": "AdminPass1!"},
+    )
+    assert resp.status_code == 204, resp.text
+    assert messages == []
+
+
+@pytest.mark.parametrize("operation", ["create", "update", "delete"])
+async def test_adapter_commits_run_queued_on_commit_callbacks(users_app, operation):
+    ran = []
+    async with users_app.state.sm.db.session_factory() as db:
+        manager = service(db, users_app)._manager
+        register_on_commit(db, lambda: ran.append(operation))
+        if operation == "create":
+            await manager.create(
+                UserCreate(email="adapter@example.com", password="SecurePass1!", is_verified=True)
+            )
+        else:
+            user = await manager.get_by_email("admin@example.com")
+            if operation == "update":
+                await manager._update(user, {"is_active": False})
+            else:
+                await manager.delete(user)
+        assert ran == [operation]
+
+
+async def test_list_items_use_trusted_construction(users_app, monkeypatch):
+    def reject_validation(*args, **kwargs):
+        raise AssertionError("DB rows should not be validated again")
+
+    monkeypatch.setattr(UserListItem, "__init__", reject_validation)
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        items, total = await svc.list_users()
+        assert total and items[0].model_dump(mode="json")["email"]
+        user = await svc.get_with_roles(items[0].id)
+        assert svc.to_list_item(user) == items[0]
+
+
+async def test_inflight_invalidation_does_not_repopulate_stale_cache(users_app, monkeypatch):
+    calls = 0
+
+    async def racing_load(db):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            await users_app.state.sm.invalidation.publish(summary_cache.CHANNEL)
+        return {"active": calls, "invited": 0, "unverified": 0}
+
+    monkeypatch.setattr(summary_cache, "load_counts", racing_load)
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        assert (await svc.count_user_states())["active"] == 1
+        assert (await svc.count_user_states())["active"] == 2
+        assert (await svc.count_user_states())["active"] == 2
+
+
+async def test_role_options_do_not_hydrate_orm_rows(users_app):
+    loaded = []
+
+    def record(*args):
+        loaded.append(args[0])
+
+    event.listen(Role, "load", record)
+    try:
+        async with users_app.state.sm.db.session_factory() as db:
+            assert await service(db, users_app).list_roles()
+            assert await refresh_roles_cache(users_app)
+        assert loaded == []
+    finally:
+        event.remove(Role, "load", record)
+
+
+async def test_create_user_without_roles_still_publishes(users_app):
+    messages = _record(users_app)
+    async with users_app.state.sm.db.session_factory() as db:
+        await service(db, users_app).create_user(
+            "noroles@example.com", "SecurePass1!", None, [], created_by=None
+        )
+        # The adapter committed; the request would roll back and drop callbacks.
+        await rollback_session(db)
+    assert len(messages) == 1
+
+
+async def test_login_bookkeeping_keeps_cache_warm_but_counted_columns_clear_it(users_app):
+    async with users_app.state.sm.db.session_factory() as db:
+        svc = service(db, users_app)
+        user = User(email="login@example.com", hashed_password="unused")
+        db.add(user)
+        await finalize_session(db)
+        before = await svc.count_user_states()
+        user.last_login_at = datetime.now(UTC)
+        user.failed_login_count = 3
+        user.session_version = (user.session_version or 0) + 1
+        await finalize_session(db)
+        with patch.object(summary_cache, "load_counts", side_effect=AssertionError("cache miss")):
+            assert await svc.count_user_states() == before
+        user.is_active = False
+        await finalize_session(db)
+        assert await svc.count_user_states() != before
diff --git a/modules/users/tests/test_users_password_offload.py b/modules/users/tests/test_users_password_offload.py
new file mode 100644
index 000000000..1f3b46882
--- /dev/null
+++ b/modules/users/tests/test_users_password_offload.py
@@ -0,0 +1,122 @@
+"""Crypto must run off-loop while the manager's DB operations stay on it."""
+
+import asyncio
+import os
+import threading
+import time
+from unittest.mock import AsyncMock
+
+import pytest
+from fastapi_users import exceptions
+from users.admin.service import UserService
+from users.contracts.schemas import UserCreate
+from users.db_adapter import UserDatabaseWithRoles
+from users.manager import UserManager
+from users.models import User
+
+
+def build_manager(db, app):
+    return UserManager(
+        UserDatabaseWithRoles(db, User), app.state.users.mailer, app.state.users.settings
+    )
+
+
+@pytest.mark.parametrize("operation", ["create", "forgot", "reset", "update"])
+async def test_crypto_runs_in_worker_thread(users_app, monkeypatch, operation):
+    async with users_app.state.sm.db.session_factory() as db:
+        manager = build_manager(db, users_app)
+        user = await manager.get_by_email("admin@example.com")
+        token = await manager.generate_reset_password_token(user)
+        loop_thread = threading.get_ident()
+        crypto_threads = []
+        for name in ("hash", "verify_and_update"):
+            original = getattr(manager.password_helper, name)
+
+            def record(*args, _original=original):
+                crypto_threads.append(threading.get_ident())
+                return _original(*args)
+
+            monkeypatch.setattr(manager.password_helper, name, record)
+        original_execute = db.execute
+
+        async def execute(*args, **kwargs):
+            assert threading.get_ident() == loop_thread
+            return await original_execute(*args, **kwargs)
+
+        monkeypatch.setattr(db, "execute", execute)
+        if operation == "create":
+            created = await manager.create(
+                UserCreate(email="new@example.com", password="SecurePass1!", is_verified=True)
+            )
+            assert created.email == "new@example.com"
+        elif operation == "forgot":
+            manager.mailer.send_password_reset = AsyncMock()
+            await manager.forgot_password(user)
+            manager.mailer.send_password_reset.assert_awaited_once()
+        elif operation == "reset":
+            await manager.reset_password(token, "SecurePass1!")
+            with pytest.raises(exceptions.InvalidResetPasswordToken):
+                await manager.reset_password(token, "AnotherPass1!")
+        else:
+            await manager._update(user, {"password": "SecurePass1!"})
+        assert crypto_threads
+        assert loop_thread not in crypto_threads
+
+
+async def test_managers_share_password_helper(users_app):
+    async with users_app.state.sm.db.session_factory() as db:
+        assert (
+            build_manager(db, users_app).password_helper
+            is build_manager(db, users_app).password_helper
+        )
+
+
+@pytest.mark.skipif(
+    not os.getenv("SM_TEST_REAL_PASSWORD_HASHING"), reason="opt-in production-cost measurement"
+)
+@pytest.mark.parametrize("operation", ["bulk_invite", "forgot"])
+async def test_real_hashing_loop_lag(users_app, operation):
+    """A 5 ms heartbeat records worst scheduler delay across real service work."""
+    delays = []
+    running = True
+
+    async def heartbeat():
+        while running:
+            started = time.perf_counter()
+            await asyncio.sleep(0.005)  # intentional-sleep: heartbeat measuring loop lag
+            delays.append(max(0, time.perf_counter() - started - 0.005))
+
+    ticker = asyncio.create_task(heartbeat())
+    await asyncio.sleep(0)
+    started = time.perf_counter()
+    try:
+        async with users_app.state.sm.db.session_factory() as db:
+            manager = build_manager(db, users_app)
+            if operation == "bulk_invite":
+                service = UserService(db, manager, users_app)
+                for i in range(100):
+                    await service.invite(f"lag-{i}@example.com", None, [])
+            else:
+                await manager.forgot_password(await manager.get_by_email("admin@example.com"))
+    finally:
+        running = False
+        await ticker
+    print(
+        f"{operation}: elapsed={time.perf_counter() - started:.3f}s "
+        f"max_loop_lag={max(delays) * 1000:.1f}ms"
+    )
+    assert max(delays) < 0.05
+
+
+async def test_bearer_login_reuses_shared_helper(anon_client, monkeypatch):
+    from fastapi_users.password import PasswordHelper
+
+    def reject_construction(*args, **kwargs):
+        raise AssertionError("request constructed another PasswordHelper")
+
+    monkeypatch.setattr(PasswordHelper, "__init__", reject_construction)
+    response = await anon_client.post(
+        "/api/users/auth/token",
+        json={"email": "admin@example.com", "password": "AdminPass1!"},
+    )
+    assert response.status_code == 200, response.text
diff --git a/modules/users/users/admin/queries.py b/modules/users/users/admin/queries.py
index a478c3ac4..580661f9d 100644
--- a/modules/users/users/admin/queries.py
+++ b/modules/users/users/admin/queries.py
@@ -16,6 +16,7 @@
 from sqlalchemy.orm import selectinload
 from sqlmodel import col
 
+from users.admin.summary_cache import get_role_summary, get_state_counts
 from users.admin.user_state import STATUS_CONDITIONS, user_state
 from users.contracts.schemas import RoleListItem, UserListItem
 from users.exceptions import EmailAlreadyExistsError, UserNotFoundError
@@ -100,7 +101,7 @@ def _invite_expiry(self, invited_at: datetime | None) -> datetime | None:
 
     def to_list_item(self, user: User) -> UserListItem:
         """Build the DTO from a User with roles already eager-loaded."""
-        return UserListItem(
+        return UserListItem.model_construct(
             id=user.id,
             email=user.email,
             full_name=user.full_name,
@@ -118,22 +119,7 @@ def to_list_item(self, user: User) -> UserListItem:
         )
 
     async def list_roles(self) -> list[RoleListItem]:
-        stmt = (
-            select(Role, func.count(UserRole.user_id))  # ty: ignore[invalid-argument-type]
-            .outerjoin(UserRole, col(UserRole.role_id) == Role.id)
-            .group_by(Role.id)  # ty: ignore[invalid-argument-type]
-            .order_by(Role.name)
-        )
-        result = await self._db.execute(stmt)
-        return [
-            RoleListItem(
-                id=role.id,
-                name=role.name,
-                description=role.description,
-                user_count=user_count,
-            )
-            for role, user_count in result.all()
-        ]
+        return await get_role_summary(self._app, self._db)
 
     async def _get_user_with_roles(self, user_id: uuid.UUID) -> User | None:
         # populate_existing: ``roles`` is ``noload``, so an instance already in
@@ -259,7 +245,7 @@ async def list_users(
         # Selected columns == UserListItem fields (minus the derived ones and
         # roles): unpack directly.
         items = [
-            UserListItem(
+            UserListItem.model_construct(
                 **cast(
                     "dict[str, Any]", {**r._mapping, "locked_until": _live_lock(r.locked_until)}
                 ),
@@ -281,15 +267,10 @@ async def count_user_states(self) -> dict[str, int]:
         self-registered account that never clicked its verification mail is
         nobody's to chase.
 
-        One scan with ``COUNT(*) FILTER (WHERE …)`` per state rather than a
-        round-trip each (Postgres and SQLite >= 3.30 both support it).
+        Cached for 30 seconds per worker (``summary_cache.TTL_SECONDS``);
+        admin writes evict it on commit.
         """
-        keys = ("active", "unverified", "invited")
-        stmt = select(*(func.count().filter(STATUS_CONDITIONS[k]()) for k in keys)).select_from(
-            User
-        )
-        row = (await self._db.execute(stmt)).one()
-        return {key: int(value) for key, value in zip(keys, row, strict=True)}
+        return await get_state_counts(self._app, self._db)
 
     async def get_with_roles(self, user_id: uuid.UUID) -> User | None:
         return await self._get_user_with_roles(user_id)
diff --git a/modules/users/users/admin/service.py b/modules/users/users/admin/service.py
index 1abe19d15..4237ea77a 100644
--- a/modules/users/users/admin/service.py
+++ b/modules/users/users/admin/service.py
@@ -11,6 +11,7 @@
 from sqlmodel import col
 
 from users.admin.safety import _UserAdminSafety
+from users.admin.summary_cache import publish_after_commit, publish_now
 from users.constants import ADMIN_ROLE_NAME
 from users.contracts.schemas import UserCreate
 from users.exceptions import (
@@ -57,17 +58,18 @@ async def create_user(
             is_verified=True,
             is_superuser=False,
         )
-        user = await self._manager.create(user_create, safe=False)  # ty: ignore[invalid-argument-type]
+        user = await self._manager.create(user_create, safe=False)
 
         roles = await self._resolve_roles(role_names)
+        if roles:
+            # Queued after the adapter's own commit, so it waits for the roles.
+            publish_after_commit(self._app, self._db)
+        else:
+            # The adapter already committed and nothing else will be written, so
+            # the request would roll back and drop an on-commit callback.
+            await publish_now(self._app)
         for role in roles:
-            self._db.add(
-                UserRole(
-                    user_id=user.id,
-                    role_id=role.id,
-                    assigned_by=created_by,
-                )
-            )
+            self._db.add(UserRole(user_id=user.id, role_id=role.id, assigned_by=created_by))
         if roles:
             await self._db.flush()
             # User.roles is lazy="noload": selectinload only populates a fresh
@@ -134,6 +136,7 @@ async def delete_user(self, user_id: uuid.UUID) -> None:
             await self._db.execute(delete(model).where(col(model.user_id) == user_id))
         await self._db.delete(user)
         await self._db.flush()
+        publish_after_commit(self._app, self._db)
 
     async def invite(
         self,
@@ -159,20 +162,14 @@ async def invite(
             is_active=True,
             is_verified=False,
         )
-        user = await self._manager.create(user_create, safe=False)  # ty: ignore[invalid-argument-type]
+        user = await self._manager.create(user_create, safe=False)
         user.invited_at = datetime.now(UTC)
+        publish_after_commit(self._app, self._db)
 
-        # Assign roles
         roles = await self._resolve_roles(role_names)
         invited_by_str = str(invited_by.id) if invited_by else None
         for role in roles:
-            self._db.add(
-                UserRole(
-                    user_id=user.id,
-                    role_id=role.id,
-                    assigned_by=invited_by_str,
-                )
-            )
+            self._db.add(UserRole(user_id=user.id, role_id=role.id, assigned_by=invited_by_str))
         await self._db.flush()
         if roles:
             await self._db.refresh(user, attribute_names=["roles"])
@@ -224,6 +221,7 @@ async def disable(self, user_id: uuid.UUID) -> User:
             .values(revoked_at=datetime.now(UTC))
         )
         await self._db.flush()
+        publish_after_commit(self._app, self._db)
         return user
 
     async def enable(self, user_id: uuid.UUID) -> User:
@@ -232,6 +230,7 @@ async def enable(self, user_id: uuid.UUID) -> User:
         user.is_active = True
         await self._revoke_sessions(user)
         await self._db.flush()
+        publish_after_commit(self._app, self._db)
         return user
 
     async def unlock(self, user_id: uuid.UUID) -> User:
@@ -248,6 +247,7 @@ async def mark_verified(self, user_id: uuid.UUID) -> User:
         if not user.is_verified:
             user.is_verified = True
             await self._db.flush()
+            publish_after_commit(self._app, self._db)
         return user
 
     async def set_roles(
@@ -262,22 +262,16 @@ async def set_roles(
         if ADMIN_ROLE_NAME in {r.name for r in user.roles} and ADMIN_ROLE_NAME not in role_names:
             await self._guard_admin_removal(user)
 
-        # Delete all existing role assignments for this user
         await self._db.execute(delete(UserRole).where(col(UserRole.user_id) == user_id))
 
-        # Insert new role assignments
         roles = await self._resolve_roles(role_names)
         for role in roles:
-            self._db.add(
-                UserRole(
-                    user_id=user_id,
-                    role_id=role.id,
-                    assigned_by=assigned_by,
-                )
-            )
+            self._db.add(UserRole(user_id=user_id, role_id=role.id, assigned_by=assigned_by))
 
         await self._revoke_sessions(user)
         await self._db.flush()
+        # The bulk DELETE above is Core DML, which the local flush marker misses.
+        publish_after_commit(self._app, self._db)
         return await self._require_user(user_id)
 
     async def generate_reset_link(self, user_id: uuid.UUID, base_url: str) -> str:
diff --git a/modules/users/users/admin/summary_cache.py b/modules/users/users/admin/summary_cache.py
new file mode 100644
index 000000000..078dc5507
--- /dev/null
+++ b/modules/users/users/admin/summary_cache.py
@@ -0,0 +1,185 @@
+"""Short-lived, app-local admin summaries (state totals and role options).
+
+This worker forgets its entries when a transaction that wrote a ``User``,
+``Role`` or ``UserRole`` through the ORM commits. Other workers learn of a
+change only through :func:`publish_after_commit`, which the admin service's
+write paths call explicitly; other writes (logins, lockout counters, password
+changes) publish nothing. Without a cross-process transport, or if a message
+is lost, another worker shows old totals until :data:`TTL_SECONDS` expires;
+these figures are informational and never gate access.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from fastapi import FastAPI
+from simple_module_core.invalidation import InvalidationBus
+from simple_module_db.callbacks import register_on_commit
+from sqlalchemy import event, func, select
+from sqlalchemy import inspect as sa_inspect
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session, SessionTransaction
+from sqlmodel import col
+
+from users.admin.user_state import STATUS_CONDITIONS
+from users.contracts.schemas import RoleListItem
+from users.models import Role, User, UserRole
+
+logger = logging.getLogger(__name__)
+
+CHANNEL = "users.admin_summary"
+TTL_SECONDS = 30.0
+_COUNTS, _ROLES = "counts", "roles"
+
+
+async def load_counts(db: AsyncSession) -> dict[str, int]:
+    """One scan with ``COUNT(*) FILTER (WHERE …)`` per state rather than a
+    round-trip each (Postgres and SQLite >= 3.30 both support it)."""
+    keys = ("active", "unverified", "invited")
+    stmt = select(*(func.count().filter(STATUS_CONDITIONS[k]()) for k in keys)).select_from(User)
+    row = (await db.execute(stmt)).one()
+    return {key: int(value) for key, value in zip(keys, row, strict=True)}
+
+
+async def load_roles(db: AsyncSession) -> list[RoleListItem]:
+    stmt = (
+        select(
+            Role.id,
+            Role.name,
+            Role.description,
+            func.count(col(UserRole.user_id)).label("user_count"),
+        )
+        .outerjoin(UserRole, col(UserRole.role_id) == Role.id)
+        .group_by(Role.id)
+        .order_by(Role.name)
+    )
+    return [RoleListItem.model_construct(**row._mapping) for row in (await db.execute(stmt)).all()]
+
+
+class SummaryCache:
+    def __init__(self) -> None:
+        self.slots: dict[str, tuple[Any, float]] = {}
+        self.generation = 0
+        self.wired = False
+
+    def forget(self) -> None:
+        self.slots.clear()
+        self.generation += 1
+
+
+def _cache(app: FastAPI) -> SummaryCache:
+    """The app's cache, created on first use.
+
+    Lazy so ``register_invalidations`` only subscribes: it must not assume an
+    app it can write state onto (the hook is also called app-less).
+    """
+    cache = getattr(app.state, "users_admin_summary", None)
+    if cache is None:
+        cache = app.state.users_admin_summary = SummaryCache()
+    return cache
+
+
+async def _cached(
+    app: FastAPI | None,
+    db: AsyncSession,
+    key: str,
+    load: Callable[[AsyncSession], Awaitable[Any]],
+) -> Any:
+    # Service-only callers and writes in progress need their transaction's own view.
+    if app is None or db.new or db.dirty or db.deleted or db.info.get(CHANNEL):
+        return await load(db)
+    cache = _cache(app)
+    slot = cache.slots.get(key)
+    now = time.monotonic()
+    if slot is not None and now < slot[1]:
+        return slot[0]
+    generation = cache.generation
+    # Never cache uncommitted rows or retain the request's transaction snapshot.
+    async with app.state.sm.db.session_factory() as committed_db:
+        value = await load(committed_db)
+    if generation == cache.generation:
+        cache.slots[key] = (value, now + TTL_SECONDS)
+    return value
+
+
+async def get_role_summary(app: FastAPI | None, db: AsyncSession) -> list[RoleListItem]:
+    """Roles with their member counts, cached for :data:`TTL_SECONDS`."""
+    value = await _cached(app, db, _ROLES, load_roles)
+    return [item.model_copy() for item in value]
+
+
+async def get_state_counts(app: FastAPI | None, db: AsyncSession) -> dict[str, int]:
+    """Active/unverified/invited totals, cached for :data:`TTL_SECONDS`."""
+    return dict(await _cached(app, db, _COUNTS, load_counts))
+
+
+async def publish_now(app: FastAPI | None) -> None:
+    """Tell every worker to forget its summaries; the write is already durable."""
+    if app is None:
+        return
+    try:
+        await app.state.sm.invalidation.publish(CHANNEL)
+    except Exception:
+        logger.exception("users.admin_summary.publish_failed")
+
+
+def publish_after_commit(app: FastAPI | None, db: AsyncSession) -> None:
+    """Tell every worker to forget its summaries once ``db`` commits."""
+    if app is None:
+        return
+
+    async def publish() -> None:
+        await app.state.sm.invalidation.publish(CHANNEL)
+
+    register_on_commit(db, publish)
+
+
+def subscribe(bus: InvalidationBus, app: FastAPI) -> None:
+    from users.roles_cache import CHANNEL as ROLES_CHANNEL
+
+    bus.subscribe(CHANNEL, lambda _: _cache(app).forget())
+    bus.subscribe(ROLES_CHANNEL, lambda _: _cache(app).forget())
+
+
+# The columns ``load_counts`` aggregates (see ``users.admin.user_state``); login
+# bookkeeping (last_login_at, lockout counters, rehash, session_version) is not one.
+_COUNTED_USER_COLUMNS = ("is_active", "is_verified", "invited_at")
+
+
+def _counted_change(user: User) -> bool:
+    attrs = sa_inspect(user).attrs
+    return any(attrs[name].history.has_changes() for name in _COUNTED_USER_COLUMNS)
+
+
+def register_writes(app: FastAPI) -> None:
+    """Forget this worker's summaries when a user/role write commits."""
+    cache = _cache(app)
+    if cache.wired:
+        return
+    cache.wired = True
+
+    def before_flush(session: Session, *_: object) -> None:
+        if (
+            any(isinstance(obj, User | Role | UserRole) for obj in (*session.new, *session.deleted))
+            or any(isinstance(obj, Role | UserRole) for obj in session.dirty)
+            or any(_counted_change(obj) for obj in session.dirty if isinstance(obj, User))
+        ):
+            session.info[CHANNEL] = True
+
+    def after_commit(session: Session) -> None:
+        # Also covers fastapi-users adapters that commit before request finalization.
+        if not session.in_nested_transaction() and session.info.pop(CHANNEL, False):
+            cache.forget()
+
+    def after_rollback(session: Session, transaction: SessionTransaction) -> None:
+        if transaction.parent is None:
+            session.info.pop(CHANNEL, None)
+
+    session_class = app.state.sm.db.sync_session_class
+    event.listen(session_class, "before_flush", before_flush)
+    event.listen(session_class, "after_commit", after_commit)
+    event.listen(session_class, "after_soft_rollback", after_rollback)
diff --git a/modules/users/users/auth_local/token_api.py b/modules/users/users/auth_local/token_api.py
index c1086b068..a23a37c1d 100644
--- a/modules/users/users/auth_local/token_api.py
+++ b/modules/users/users/auth_local/token_api.py
@@ -26,6 +26,7 @@
     record_success,
     release,
 )
+from users.core.password_hashing import PASSWORD_HELPER
 from users.demo.guard import DEMO_READ_ONLY_DETAIL, is_read_only_demo
 from users.models import User
 from users.models.refresh_token import RefreshToken
@@ -68,9 +69,6 @@ async def token_login(
     db: AsyncSession = Depends(get_db),
 ):
     """Exchange email + password for an access/refresh token pair."""
-    from fastapi_users.password import PasswordHelper
-
-    helper = PasswordHelper()
     # Same lockout as the session login (GH #530): a slot is taken before the
     # password is checked, and a success gives it back.
     attempt = login_attempt(request, body.email)
@@ -94,13 +92,13 @@ async def token_login(
         # Constant-time: run bcrypt on a dummy hash to prevent timing-based
         # email enumeration (existing user + wrong password takes ~50ms for
         # bcrypt; missing user would be instant without this).
-        await asyncio.to_thread(helper.verify_and_update, body.password, _DUMMY_HASH)
+        await asyncio.to_thread(PASSWORD_HELPER.verify_and_update, body.password, _DUMMY_HASH)
         await rollback_session(db)
         locked = await record_failure(db, attempt, admission)
         raise locked or HTTPException(status_code=401, detail="Invalid credentials")
 
     verified, updated_hash = await asyncio.to_thread(
-        helper.verify_and_update, body.password, user.hashed_password
+        PASSWORD_HELPER.verify_and_update, body.password, user.hashed_password
     )
     if not verified:
         await rollback_session(db)
diff --git a/modules/users/users/core/__init__.py b/modules/users/users/core/__init__.py
new file mode 100644
index 000000000..38e57919f
--- /dev/null
+++ b/modules/users/users/core/__init__.py
@@ -0,0 +1 @@
+"""Users domain core: password hashing boundaries."""
diff --git a/modules/users/users/core/password_hashing.py b/modules/users/users/core/password_hashing.py
new file mode 100644
index 000000000..8c0f9c018
--- /dev/null
+++ b/modules/users/users/core/password_hashing.py
@@ -0,0 +1,73 @@
+"""Async crypto boundaries for fastapi-users 15's synchronous password helper."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import jwt
+from fastapi import Request
+from fastapi_users import exceptions
+from fastapi_users.jwt import decode_jwt
+from fastapi_users.password import PasswordHelper
+
+from users.contracts.schemas import UserCreate
+from users.models import User
+
+PASSWORD_HELPER = PasswordHelper()
+
+
+class PasswordHashingMixin:
+    """Keep validation, database operations and lifecycle hooks on the event loop."""
+
+    async def create(
+        self,
+        user_create: UserCreate,
+        safe: bool = False,
+        request: Request | None = None,
+    ) -> User:
+        await self.validate_password(user_create.password, user_create)
+        if await self.user_db.get_by_email(user_create.email) is not None:
+            raise exceptions.UserAlreadyExists()
+        data = (
+            user_create.create_update_dict() if safe else user_create.create_update_dict_superuser()
+        )
+        password = data.pop("password")
+        data["hashed_password"] = await asyncio.to_thread(self.password_helper.hash, password)
+        user = await self.user_db.create(data)
+        await self.on_after_register(user, request)
+        return user
+
+    async def _update(self, user: User, update_dict: dict[str, Any]) -> User:
+        # Let upstream handle email uniqueness and every non-password field.
+        data = dict(update_dict)
+        if data.get("password") is not None:
+            password = data.pop("password")
+            await self.validate_password(password, user)
+            data["hashed_password"] = await asyncio.to_thread(self.password_helper.hash, password)
+        return await super()._update(user, data)
+
+    async def reset_password(
+        self, token: str, password: str, request: Request | None = None
+    ) -> User:
+        # Mirrors upstream token/fingerprint checks; only crypto leaves the loop.
+        try:
+            data = decode_jwt(
+                token, self.reset_password_token_secret, [self.reset_password_token_audience]
+            )
+            user_id = data["sub"]
+            fingerprint = data["password_fgpt"]
+            parsed_id = self.parse_id(user_id)
+        except (jwt.PyJWTError, KeyError, exceptions.InvalidID):
+            raise exceptions.InvalidResetPasswordToken() from None
+        user = await self.get(parsed_id)
+        valid, _ = await asyncio.to_thread(
+            self.password_helper.verify_and_update, user.hashed_password, fingerprint
+        )
+        if not valid:
+            raise exceptions.InvalidResetPasswordToken()
+        if not user.is_active:
+            raise exceptions.UserInactive()
+        updated = await self._update(user, {"password": password})
+        await self.on_after_reset_password(updated, request)
+        return updated
diff --git a/modules/users/users/db_adapter.py b/modules/users/users/db_adapter.py
index 5b8f5099a..b9e9939dd 100644
--- a/modules/users/users/db_adapter.py
+++ b/modules/users/users/db_adapter.py
@@ -9,6 +9,7 @@
 from fastapi_users.exceptions import UserAlreadyExists
 from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase
 from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase
+from simple_module_db.callbacks import run_on_commit_callbacks
 from simple_module_db.deps import get_db
 from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
@@ -48,9 +49,19 @@ async def create(self, create_dict):
                 raise UserAlreadyExists() from None
             raise
         await self.session.commit()
+        await run_on_commit_callbacks(self.session)
         await self.session.refresh(user)
         return user
 
+    async def update(self, user, update_dict):
+        user = await super().update(user, update_dict)
+        await run_on_commit_callbacks(self.session)
+        return user
+
+    async def delete(self, user):
+        await super().delete(user)
+        await run_on_commit_callbacks(self.session)
+
     async def get(self, id):
         stmt = (
             select(self.user_table)
diff --git a/modules/users/users/manager.py b/modules/users/users/manager.py
index 3d72c15c1..a39a344a7 100644
--- a/modules/users/users/manager.py
+++ b/modules/users/users/manager.py
@@ -23,6 +23,7 @@
     SESSION_VERSION_KEY,
 )
 from users.contracts.events import UserRegistered
+from users.core.password_hashing import PASSWORD_HELPER, PasswordHashingMixin
 from users.db_adapter import UserDatabaseWithRoles, get_user_db
 from users.exceptions import ExternalUserNoPasswordError
 from users.mailer import Mailer
@@ -36,7 +37,9 @@
 logger = logging.getLogger(__name__)
 
 
-class UserManager(ResetRevocationMixin, UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
+class UserManager(
+    ResetRevocationMixin, PasswordHashingMixin, UUIDIDMixin, BaseUserManager[User, uuid.UUID]
+):
     """Customizes password validation, token secrets, and lifecycle emails."""
 
     # Secrets and lifetimes pulled from settings at construction time so
@@ -47,7 +50,7 @@ def __init__(
         mailer: Mailer,
         settings: UsersSettings,
     ) -> None:
-        super().__init__(user_db)
+        super().__init__(user_db, PASSWORD_HELPER)
         self.mailer = mailer
         self.reset_password_token_secret = settings.reset_password_token_secret
         self.verification_token_secret = settings.verification_token_secret
@@ -113,7 +116,10 @@ async def forgot_password(self, user: User, request: Request | None = None) -> N
         """
         if user.is_external or user.hashed_password is None:
             return
-        await super().forgot_password(user, request)
+        if not user.is_active:
+            raise exceptions.UserInactive()
+        token = await self.generate_reset_password_token(user)
+        await self.on_after_forgot_password(user, token, request)
 
     # ── Lifecycle hooks ──────────────────────────────────────
 
@@ -260,17 +266,9 @@ def mint_invite_token(self, user: User, invited_by: str | None = None) -> str:
     async def generate_reset_password_token(self, user: User) -> str:
         """Mint a reset-audience JWT without firing on_after_forgot_password.
 
-        Shape matches ``BaseUserManager.forgot_password``: the payload includes
-        ``password_fgpt`` (a bcrypt of the current hashed_password) so the token
-        invalidates when the password changes. Used by the admin
-        reset-password-link endpoint, where the admin copies the link instead
-        of triggering an email.
-
-        The fingerprint must be bcrypt-style because the stock
-        ``reset_password`` path in fastapi-users verifies it with
-        ``password_helper.verify_and_update``. Bcrypt is CPU-bound (~100ms at
-        default rounds) so we offload it to a worker thread — otherwise a
-        single admin action would stall the event loop for other requests.
+        The payload fingerprints the current hashed password, so changing the
+        password invalidates the token. The CPU-bound fingerprint is hashed in
+        a worker thread for both forgot-password mail and admin reset links.
         """
         if user.is_external or user.hashed_password is None:
             # No local password to fingerprint — hashing None would crash.
diff --git a/modules/users/users/module.py b/modules/users/users/module.py
index c19e23de3..d5cbdcff5 100644
--- a/modules/users/users/module.py
+++ b/modules/users/users/module.py
@@ -126,12 +126,14 @@ def register_invalidations(self, bus: InvalidationBus, app: FastAPI) -> None:
         sessions for the rest of their TTL (GH #318).
         """
         from users import roles_cache
+        from users.admin import summary_cache
         from users.session_revocation import subscribe
 
         subscribe(bus)
         # The admin user screens' cached role list: the setup wizard can add
         # the admin role after every worker cached the seeded ``user`` role.
         roles_cache.subscribe(bus, app)
+        summary_cache.subscribe(bus, app)
 
     def register_exception_handlers(self, app: FastAPI) -> None:
         from users.errors import install_exception_handlers
diff --git a/modules/users/users/roles_cache.py b/modules/users/users/roles_cache.py
index 5fbb00ee2..843b3c41b 100644
--- a/modules/users/users/roles_cache.py
+++ b/modules/users/users/roles_cache.py
@@ -52,8 +52,8 @@ class RoleSummary:
 async def refresh_roles_cache(app: FastAPI) -> list[RoleSummary]:
     """Reload the roles list from the DB into ``app.state.users.roles_cache``."""
     async with app.state.sm.db.session_factory() as db:
-        result = await db.execute(select(Role).order_by(Role.name))
-        cached = [RoleSummary(id=str(r.id), name=r.name) for r in result.scalars().all()]
+        result = await db.execute(select(Role.id, Role.name).order_by(Role.name))
+        cached = [RoleSummary(id=str(r.id), name=r.name) for r in result.all()]
     app.state.users.roles_cache = cached
     app.state.users.roles_cache_loaded_at = time.monotonic()
     return cached
diff --git a/modules/users/users/startup.py b/modules/users/users/startup.py
index 0b00a4748..dc6bb45fe 100644
--- a/modules/users/users/startup.py
+++ b/modules/users/users/startup.py
@@ -44,6 +44,7 @@
 
 async def run_startup(module: UsersModule, app: FastAPI) -> None:
     """Populate ``app.state.users`` and apply settings-derived configuration."""
+    from users.admin.summary_cache import register_writes
     from users.backend import reconfigure_cookie_transport
     from users.bootstrap import bootstrap_admin_from_env, ensure_user_role
     from users.demo import ensure_demo_users
@@ -52,6 +53,7 @@ async def run_startup(module: UsersModule, app: FastAPI) -> None:
     from users.session_version_cache import configure_session_version_cache
     from users.settings import DEFAULT_LOGIN_REDIRECT_URL
 
+    register_writes(app)
     state = app.state.users
     s = state.settings