diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1e6971fc3..282a3798d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,6 +16,18 @@ All notable changes to this project are documented in this file. The format is b
 ## [Unreleased]
 
 ### Fixed
+- Users (security): disabling or enabling an account, or replacing its roles from the admin
+  screens, now bumps `session_version` in the same transaction and evicts the cached counter on
+  every worker after commit (an atomic SQL increment, so overlapping changes cannot lose one), so
+  existing cookie sessions and bearer access tokens are refused on their next request; disabling
+  also revokes the account's refresh tokens. Before, a disabled or demoted user kept the roles
+  cached in their cookie until the session expired. **Behaviour change:** a role change signs that
+  user out of their existing sessions; a remembered sign-in re-authenticates with the new roles
+  (#581).
+- File storage: **Behaviour change.** `quota_bytes` is now enforced for plain uploads as well as
+  resumable ones (stored bytes plus active reservations, serialised per tenant on Postgres); an
+  upload that would exceed it answers 413 and its object is removed. Previously only resumable
+  sessions were checked, and the setting was described as report-only (#586).
 - File storage: `DELETE /api/file_storage/files/{id}` and the bulk delete now drop the stored
   bytes only after the request transaction commits, so a failed commit no longer leaves a live
   row with no bytes. A new hourly `file_storage.reap_objects` beat task (needs
diff --git a/docs/modules/file_storage.md b/docs/modules/file_storage.md
index 7633826b1..1e3615b8a 100644
--- a/docs/modules/file_storage.md
+++ b/docs/modules/file_storage.md
@@ -286,6 +286,7 @@ DB-backed via `register_module_settings`; pydantic defaults seed at boot. Live e
 | `s3_signature_version` | `"s3v4"` | presign signature version; pinned so custom endpoints get SigV4 |
 | `s3_presign_ttl_seconds` | `300` | TTL for presigned `GET` URLs |
 | `max_file_size_bytes` | `100 * 1024 * 1024` | upload limit (100 MB) |
+| `quota_bytes` | `None` | Enforced on plain and resumable uploads: stored bytes plus unexpired active session reservations; `None` means unlimited aggregate storage. Usage follows file-listing tenant scope; platform uploads count platform-owned files separately. PostgreSQL serializes quota admissions until the recording transaction commits. Rejected plain uploads delete the uploaded object. |
 | `allowed_content_types` | `None` | optional MIME-type whitelist; `None` ⇒ any |
 
 ## Backends
diff --git a/modules/branding/branding/endpoints/api.py b/modules/branding/branding/endpoints/api.py
index e1bf9187b..12166e17e 100644
--- a/modules/branding/branding/endpoints/api.py
+++ b/modules/branding/branding/endpoints/api.py
@@ -10,7 +10,7 @@
 from branding import constants
 from branding.contracts.schemas import BrandingOut, BrandingUpdate
 from branding.deps import BrandingServiceDep
-from branding.images import validate_image
+from branding.images import store_image, validate_image
 from branding.presets import find_preset
 
 router = APIRouter()
@@ -67,7 +67,7 @@ async def upload_logo(
     storage: FileStorageService = Depends(get_file_storage_service),
 ) -> BrandingOut:
     await validate_image(file)
-    stored = await storage.upload(file, platform=True)
+    stored = await store_image(storage, file, platform=True)
     return await service.set_logo(str(stored.id))
 
 
@@ -91,7 +91,7 @@ async def upload_logo_dark(
     storage: FileStorageService = Depends(get_file_storage_service),
 ) -> BrandingOut:
     await validate_image(file)
-    stored = await storage.upload(file, platform=True)
+    stored = await store_image(storage, file, platform=True)
     return await service.set_logo_dark(str(stored.id))
 
 
@@ -107,7 +107,7 @@ async def upload_favicon(
     storage: FileStorageService = Depends(get_file_storage_service),
 ) -> BrandingOut:
     await validate_image(file)
-    stored = await storage.upload(file, platform=True)
+    stored = await store_image(storage, file, platform=True)
     return await service.set_favicon(str(stored.id))
 
 
diff --git a/modules/branding/branding/endpoints/tenant_api.py b/modules/branding/branding/endpoints/tenant_api.py
index 790891fa4..76a073586 100644
--- a/modules/branding/branding/endpoints/tenant_api.py
+++ b/modules/branding/branding/endpoints/tenant_api.py
@@ -27,7 +27,7 @@
 
 from branding.constants import PACKAGE, PATH_TENANT_ASSET, TENANT_ASSETS, TENANT_IMAGE_FIELDS
 from branding.contracts.schemas import BrandingOut
-from branding.images import validate_image
+from branding.images import store_image, validate_image
 from branding.reaper import schedule_reap
 from branding.service import to_out
 from branding.tenant_branding import merge, overrides_from
@@ -83,7 +83,7 @@ async def upload_tenant_asset(
     field = _field(asset)
     tenant_id = active_tenant(request)
     await validate_image(file)
-    stored = await storage.upload(file)
+    stored = await store_image(storage, file)
     return await _swap(request, settings, storage, tenant_id, field, str(stored.id))
 
 
diff --git a/modules/branding/branding/images.py b/modules/branding/branding/images.py
index 334758b6f..eb1323631 100644
--- a/modules/branding/branding/images.py
+++ b/modules/branding/branding/images.py
@@ -12,10 +12,14 @@
 
 from __future__ import annotations
 
-from typing import Final
+from typing import TYPE_CHECKING, Final
 
 from fastapi import HTTPException, UploadFile
 
+if TYPE_CHECKING:
+    from file_storage.contracts.schemas import StoredFileOut
+    from file_storage.service import FileStorageService
+
 # Raster + icon only. SVG is excluded on purpose: it is an XML document that can
 # carry <script>, so serving one back from our own origin would be stored XSS.
 # The asset route defends in depth (attachment + nosniff), but the narrower
@@ -96,3 +100,22 @@ async def validate_image(file: UploadFile) -> None:
                 f"{content_type!r}. {ALLOWED_IMAGE_HINT}"
             ),
         )
+
+
+async def store_image(
+    storage: FileStorageService, file: UploadFile, *, platform: bool = False
+) -> StoredFileOut:
+    """Hand a validated image to ``file_storage``, answering 413 when the quota is full.
+
+    ``file_storage`` enforces ``quota_bytes`` on every upload; its own endpoint
+    maps the refusal to 413, but branding calls the service directly, so an
+    unmapped ``QuotaExceededError`` would surface as a 500.
+    """
+    from file_storage.upload_support import QuotaExceededError
+
+    try:
+        return await storage.upload(file, platform=platform)
+    except QuotaExceededError:
+        raise HTTPException(
+            status_code=413, detail="The storage quota is full; the image was not saved."
+        ) from None
diff --git a/modules/file_storage/file_storage/endpoints/api.py b/modules/file_storage/file_storage/endpoints/api.py
index ac89e2118..34b41a273 100644
--- a/modules/file_storage/file_storage/endpoints/api.py
+++ b/modules/file_storage/file_storage/endpoints/api.py
@@ -38,6 +38,7 @@
 from file_storage.contracts.service import StorageNotFoundError
 from file_storage.deps import OwnerScope, get_event_bus, get_file_storage_service
 from file_storage.downloads import content_disposition
+from file_storage.endpoints.upload_errors import quota
 from file_storage.format import format_bytes
 from file_storage.ranges import file_body_response
 from file_storage.service import (
@@ -49,6 +50,7 @@
     StreamDownload,
 )
 from file_storage.serving import etag_for, not_found, sandbox_headers, thumbnail_response
+from file_storage.upload_support import QuotaExceededError
 
 router = APIRouter()
 
@@ -68,6 +70,8 @@ async def upload_file(
 ) -> StoredFileOut:
     try:
         out = await service.upload(file, public=public)
+    except QuotaExceededError as exc:
+        raise quota(t) from exc
     except FileTooLargeError as exc:
         # The limit belongs in the sentence: "too large" is not actionable to
         # someone holding a 40 MB file, and every client that shows this
diff --git a/modules/file_storage/file_storage/service.py b/modules/file_storage/file_storage/service.py
index 3a1d26e41..086fab052 100644
--- a/modules/file_storage/file_storage/service.py
+++ b/modules/file_storage/file_storage/service.py
@@ -31,6 +31,7 @@
 from file_storage.models import StoredFile
 from file_storage.reads import FileStorageReads
 from file_storage.scope import PLATFORM_TENANT_ID, owning_tenant, platform_scope
+from file_storage.upload_guards import StorageQuota
 from file_storage.visibility import FileStoragePublic
 
 if TYPE_CHECKING:
@@ -54,7 +55,7 @@ class StoredFileNotFoundError(Exception):
     """Raised when a StoredFile lookup misses; endpoints map this to HTTP 404."""
 
 
-class FileStorageService(FileStoragePublic, FileStorageReads, FileStorageDeletes):
+class FileStorageService(FileStoragePublic, FileStorageReads, FileStorageDeletes, StorageQuota):
     """Orchestrates validation, hashing, backend IO, and DB lifecycle."""
 
     def __init__(
@@ -121,6 +122,7 @@ async def _hashing_stream() -> AsyncIterator[bytes]:
         # the cleanup error after logging; the orphaned key can be reaped
         # by a janitor sweep.
         try:
+            await self._check_quota(size, reserve=True, platform=platform)
             row = StoredFile(
                 tenant_id=tenant_id,
                 key=key,
diff --git a/modules/file_storage/file_storage/settings.py b/modules/file_storage/file_storage/settings.py
index 2406d8cb9..e25b5c191 100644
--- a/modules/file_storage/file_storage/settings.py
+++ b/modules/file_storage/file_storage/settings.py
@@ -82,17 +82,17 @@ class FileStorageSettings(DbBackedSettings):
         default=None,
         description="Whitelist of MIME types. None = any type allowed.",
     )
-    # Reported, not enforced: the browse screen states "1.2 GB of 5 GB used"
-    # only for installs that have said what 5 GB means to them. Left unset,
-    # the screen reports usage without inventing a ceiling, and uploads are
-    # bounded by ``max_file_size_bytes`` alone as before.
+    # Enforced for plain uploads and resumable sessions: stored bytes plus
+    # unexpired active reservations. None leaves aggregate storage unlimited.
     # Positive, and no larger than a browser can hold exactly: the screen and
     # the settings editor read it as a JavaScript number (#548).
     quota_bytes: int | None = Field(
         default=None,
         gt=0,
         le=constants.MAX_JS_SAFE_INTEGER,
-        description="Total storage the browse screen measures usage against. None = unmeasured.",
+        description=(
+            "Storage ceiling for stored files and active upload reservations. None = unlimited."
+        ),
     )
 
     public_files_cors: bool = Field(
diff --git a/modules/file_storage/file_storage/upload_guards.py b/modules/file_storage/file_storage/upload_guards.py
index 655f90a18..217b1df42 100644
--- a/modules/file_storage/file_storage/upload_guards.py
+++ b/modules/file_storage/file_storage/upload_guards.py
@@ -1,4 +1,4 @@
-"""Cleanup and quota guards for upload sessions (#454)."""
+"""Cleanup and shared quota guards for uploads."""
 
 from __future__ import annotations
 
@@ -12,7 +12,7 @@
 from file_storage import constants
 from file_storage.contracts.service import StorageError
 from file_storage.models import StoredFile, UploadSession
-from file_storage.scope import owning_tenant
+from file_storage.scope import PLATFORM_TENANT_ID, owning_tenant, platform_scope
 from file_storage.upload_support import DirectUploadUnsupportedError, QuotaExceededError
 
 if TYPE_CHECKING:
@@ -28,13 +28,62 @@
 _QUOTA_LOCK_KEY = 0x4653544F << 32
 
 
-class UploadGuards:
-    """Mixed into :class:`~file_storage.uploads.UploadSessionService`."""
+class StorageQuota:
+    """Quota checks shared by plain uploads and resumable reservations."""
 
-    files: FileStorageService
     db: AsyncSession
     settings: FileStorageSettings
 
+    async def _serialize_reservations(self, *, platform: bool = False) -> None:
+        """Make concurrent quota admissions take turns on Postgres.
+
+        Each upload sums stored + reserved bytes before recording a file or
+        reservation. Under READ COMMITTED, concurrent uploads can read the same
+        sum and all pass. The transaction-scoped lock lasts until the new row
+        becomes visible at commit. SQLite serialises writers already.
+        """
+        bind = self.db.get_bind()
+        if bind.dialect.name != "postgresql":
+            return
+        tenant = owning_tenant(self.db, platform=platform)
+        await self.db.execute(
+            text("SELECT pg_advisory_xact_lock(:key)"),
+            {"key": _QUOTA_LOCK_KEY + zlib.crc32(tenant.encode())},
+        )
+
+    async def _check_quota(self, size: int, *, reserve: bool, platform: bool = False) -> None:
+        """Refuse ``size`` bytes that would push stored (+ reserved) bytes past the quota.
+
+        Read live, not from the browse screen's cached totals: a cache a minute
+        stale would let a burst of sessions each see an empty bucket. With
+        ``reserve``, sessions still open count at their declared size, so opening
+        many small-looking sessions cannot overbook the quota.
+        """
+        quota = self.settings.quota_bytes
+        if quota is None:
+            return
+        if reserve:
+            await self._serialize_reservations(platform=platform)
+        stored_query = select(func.coalesce(func.sum(StoredFile.size_bytes), 0))
+        reserved_query = select(func.coalesce(func.sum(UploadSession.total_size), 0)).where(
+            UploadSession.status == constants.UploadStatus.ACTIVE,  # ty: ignore[invalid-argument-type]
+            UploadSession.expires_at > datetime.now(UTC),  # ty: ignore[invalid-argument-type]
+        )
+        if platform:
+            stored_query = stored_query.where(StoredFile.tenant_id == PLATFORM_TENANT_ID)
+            reserved_query = reserved_query.where(UploadSession.tenant_id == PLATFORM_TENANT_ID)
+        async with platform_scope(self.db, platform):
+            stored = await self.db.scalar(stored_query)
+            reserved = await self.db.scalar(reserved_query) if reserve else 0
+        if (stored or 0) + (reserved or 0) + size > quota:
+            raise QuotaExceededError(str(quota))
+
+
+class UploadGuards(StorageQuota):
+    """Mixed into :class:`~file_storage.uploads.UploadSessionService`."""
+
+    files: FileStorageService
+
     @property
     def backend(self) -> DirectUploadBackend:
         backend = self.files.backend
@@ -73,45 +122,3 @@ async def purge_finished(self, *, owner: str | None) -> int:
             )
         )
         return result.rowcount or 0
-
-    async def _serialize_reservations(self) -> None:
-        """Make concurrent session openers take turns on Postgres.
-
-        Each opener sums stored + reserved bytes and then inserts its own
-        reservation; under READ COMMITTED, N openers all read the same sum and
-        all pass. A transaction-scoped advisory lock (held until this request
-        commits, which is when the new reservation becomes visible) closes the
-        window. SQLite serialises writers already.
-        """
-        bind = self.db.get_bind()
-        if bind.dialect.name != "postgresql":
-            return
-        await self.db.execute(
-            text("SELECT pg_advisory_xact_lock(:key)"),
-            {"key": _QUOTA_LOCK_KEY + zlib.crc32(owning_tenant(self.db).encode())},
-        )
-
-    async def _check_quota(self, size: int, *, reserve: bool) -> None:
-        """Refuse ``size`` bytes that would push stored (+ reserved) bytes past the quota.
-
-        Read live, not from the browse screen's cached totals: a cache a minute
-        stale would let a burst of sessions each see an empty bucket. With
-        ``reserve``, sessions still open count at their declared size, so opening
-        many small-looking sessions cannot overbook the quota.
-        """
-        quota = self.settings.quota_bytes
-        if quota is None:
-            return
-        if reserve:
-            await self._serialize_reservations()
-        stored = await self.db.scalar(select(func.coalesce(func.sum(StoredFile.size_bytes), 0)))
-        reserved = 0
-        if reserve:
-            reserved = await self.db.scalar(
-                select(func.coalesce(func.sum(UploadSession.total_size), 0)).where(
-                    UploadSession.status == constants.UploadStatus.ACTIVE,  # ty: ignore[invalid-argument-type]
-                    UploadSession.expires_at > datetime.now(UTC),  # ty: ignore[invalid-argument-type]
-                )
-            )
-        if (stored or 0) + (reserved or 0) + size > quota:
-            raise QuotaExceededError(str(quota))
diff --git a/modules/file_storage/tests/test_file_storage_plain_quota.py b/modules/file_storage/tests/test_file_storage_plain_quota.py
new file mode 100644
index 000000000..1a09a59a5
--- /dev/null
+++ b/modules/file_storage/tests/test_file_storage_plain_quota.py
@@ -0,0 +1,59 @@
+"""Plain uploads share stored-byte limits and resumable reservations (#586)."""
+
+from file_storage import constants
+from file_storage_uploads_helpers import API, open_session
+
+
+async def upload(client, size):
+    return await client.post(
+        f"{API}/upload", files={"file": ("small.bin", b"x" * size, "application/octet-stream")}
+    )
+
+
+async def test_plain_uploads_enforce_aggregate_quota(authenticated_client, storage):
+    storage.settings.quota_bytes = 10
+    assert (await upload(authenticated_client, 6)).status_code == 201
+    assert (await upload(authenticated_client, 4)).status_code == 201
+    rejected = await upload(authenticated_client, 1)
+    assert rejected.status_code == 413
+    assert rejected.json()["detail"]["code"] == constants.ErrorCode.QUOTA_EXCEEDED
+    listing = (await authenticated_client.get(f"{API}/files")).json()
+    assert listing["total"] == 2
+    assert len([p for p in storage.backend.root.rglob("*") if p.is_file()]) == 2
+
+
+async def test_plain_upload_honors_reservations_and_abort_releases_them(
+    authenticated_client, storage, checksum_calls
+):
+    storage.settings.quota_bytes = 10
+    response = await open_session(authenticated_client, total=8)
+    assert response.status_code == 201
+    assert (await upload(authenticated_client, 3)).status_code == 413
+    assert (await upload(authenticated_client, 2)).status_code == 201
+    await authenticated_client.delete(f"{API}/uploads/{response.json()['id']}")
+    assert (await upload(authenticated_client, 8)).status_code == 201
+
+
+async def test_platform_quota_counts_only_platform_files(db_session, tmp_path):
+    from io import BytesIO
+
+    import pytest
+    from fastapi import UploadFile
+    from file_storage.backends.filesystem import FilesystemBackend
+    from file_storage.service import FileStorageService
+    from file_storage.settings import FileStorageSettings
+    from file_storage.upload_support import QuotaExceededError
+    from simple_module_db import tenant_context
+
+    backend = FilesystemBackend(root=tmp_path)
+    service = FileStorageService(db_session, backend, FileStorageSettings(quota_bytes=3))
+
+    def file(size):
+        return UploadFile(BytesIO(b"x" * size), filename="small.bin")
+
+    with tenant_context("tenant-a"):
+        await service.upload(file(3))
+    await service.upload(file(3), platform=True)
+    with pytest.raises(QuotaExceededError):
+        await service.upload(file(1), platform=True)
+    assert len([p for p in tmp_path.rglob("*") if p.is_file()]) == 2
diff --git a/modules/file_storage/tests/test_file_storage_upload_guards.py b/modules/file_storage/tests/test_file_storage_upload_guards.py
index 5b844a686..5e0bd8bd3 100644
--- a/modules/file_storage/tests/test_file_storage_upload_guards.py
+++ b/modules/file_storage/tests/test_file_storage_upload_guards.py
@@ -233,7 +233,7 @@ async def test_reserve_takes_the_lock_before_reading_usage(
         order: list[str] = []
         guards = _Guards(db_session, backend, quota_bytes=100)
 
-        async def fake_lock() -> None:
+        async def fake_lock(*, platform: bool = False) -> None:
             order.append("lock")
 
         monkeypatch.setattr(guards, "_serialize_reservations", fake_lock)
@@ -261,9 +261,9 @@ async def test_is_a_no_op_off_postgres(self, backend, db_session):
 
     async def test_takes_a_per_tenant_advisory_lock_on_postgres(self, backend, monkeypatch):
         calls: list[tuple[str, Any]] = []
-        monkeypatch.setattr(guards_module, "owning_tenant", lambda db: "acme")
+        monkeypatch.setattr(guards_module, "owning_tenant", lambda db, platform=False: "acme")
         await _Guards(_db_on("postgresql", calls), backend)._serialize_reservations()
-        monkeypatch.setattr(guards_module, "owning_tenant", lambda db: "globex")
+        monkeypatch.setattr(guards_module, "owning_tenant", lambda db, platform=False: "globex")
         await _Guards(_db_on("postgresql", calls), backend)._serialize_reservations()
 
         assert all("pg_advisory_xact_lock" in sql for sql, _ in calls)
diff --git a/modules/file_storage/tests/test_file_storage_uploads_pg.py b/modules/file_storage/tests/test_file_storage_uploads_pg.py
index 1739b47a4..2ffd2a5a2 100644
--- a/modules/file_storage/tests/test_file_storage_uploads_pg.py
+++ b/modules/file_storage/tests/test_file_storage_uploads_pg.py
@@ -53,3 +53,24 @@ async def test_concurrent_completes_record_one_file(authenticated_client, storag
     assert len({r.json()["file"]["id"] for r in responses}) == 1
     assert (await authenticated_client.get(f"{API}/files")).json()["total"] == 1
     assert len(checksum_calls) == 1
+
+
+async def test_concurrent_plain_uploads_cannot_overbook_quota(authenticated_client, storage):
+    storage.settings.quota_bytes = 10
+    responses = await asyncio.gather(
+        *(
+            authenticated_client.post(
+                f"{API}/upload",
+                files={"file": ("small.bin", b"x" * 5, "application/octet-stream")},
+            )
+            for _ in range(6)
+        )
+    )
+    assert sorted(r.status_code for r in responses) == [201, 201, 413, 413, 413, 413]
+    assert all(
+        r.json()["detail"]["code"] == constants.ErrorCode.QUOTA_EXCEEDED
+        for r in responses
+        if r.status_code == 413
+    )
+    assert (await authenticated_client.get(f"{API}/files")).json()["total"] == 2
+    assert len([p for p in storage.backend.root.rglob("*") if p.is_file()]) == 2
diff --git a/modules/users/tests/test_session_version_atomic.py b/modules/users/tests/test_session_version_atomic.py
new file mode 100644
index 000000000..fb9b8c230
--- /dev/null
+++ b/modules/users/tests/test_session_version_atomic.py
@@ -0,0 +1,35 @@
+"""Revocations must increment the database counter rather than a stale copy."""
+
+from sqlalchemy import event
+from users.admin.queries import _UserServiceBase
+from users.models import User
+
+
+async def test_revocations_use_atomic_sql_and_refresh_counter(db_session):
+    user = User(email="atomic@example.com", hashed_password="unused", session_version=7)
+    db_session.add(user)
+    await db_session.flush()
+    statements = []
+
+    def record_sql(conn, cursor, statement, parameters, context, executemany):
+        if statement.lstrip().upper().startswith("UPDATE"):
+            statements.append(statement)
+
+    engine = db_session.bind.sync_engine
+    event.listen(engine, "before_cursor_execute", record_sql)
+    try:
+        service = _UserServiceBase(db_session, None)
+        await service._revoke_sessions(user)
+        # A plain read must work without implicit async IO after the flush.
+        assert user.session_version == 8
+        await service._revoke_sessions(user)
+        assert user.session_version == 9
+        await db_session.refresh(user)
+        assert user.session_version == 9
+    finally:
+        event.remove(engine, "before_cursor_execute", record_sql)
+
+    # Each write derives its value from the database row, so overlapping
+    # requests with the same stale value cannot overwrite an increment.
+    assert len(statements) == 2
+    assert all("session_version=(users_user.session_version +" in sql for sql in statements)
diff --git a/modules/users/tests/test_users_admin_session_revocation.py b/modules/users/tests/test_users_admin_session_revocation.py
new file mode 100644
index 000000000..fc00dde34
--- /dev/null
+++ b/modules/users/tests/test_users_admin_session_revocation.py
@@ -0,0 +1,106 @@
+"""Admin account changes strand cached browser sessions on every worker."""
+
+import pytest
+from _users_app_builders import user_row
+from simple_module_core.invalidation import InvalidationBus
+from users.session_revocation import subscribe
+from users.session_version_cache import (
+    clear_session_version_cache,
+    read_session_version,
+    store_session_version,
+)
+
+
+class RecordingTransport:
+    def __init__(self):
+        self.messages = []
+
+    async def publish(self, message):
+        self.messages.append(message)
+
+
+@pytest.fixture(autouse=True)
+def empty_cache():
+    clear_session_version_cache()
+    yield
+    clear_session_version_cache()
+
+
+@pytest.mark.parametrize("action", ["disable", "enable", "roles"])
+async def test_admin_change_revokes_cached_session(
+    users_app, anon_client, admin_client, alice, action
+):
+    promotion = await admin_client.put(
+        f"/api/users/admin/{alice.id}/roles", json={"role_names": ["admin"]}
+    )
+    assert promotion.status_code == 200
+    login = await anon_client.post(
+        "/api/users/auth/login",
+        data={"username": alice.email, "password": "UserPass1!"},
+    )
+    assert login.status_code == 204
+    user = await user_row(users_app, alice.email)
+    assert (await anon_client.get("/api/users/admin")).status_code == 200
+    assert (await anon_client.get("/api/users/admin")).status_code == 200
+    assert read_session_version(user.id) == (True, user.session_version)
+
+    tokens = await anon_client.post(
+        "/api/users/auth/token", json={"email": alice.email, "password": "UserPass1!"}
+    )
+    assert tokens.status_code == 200
+    bearer = {"Authorization": f"Bearer {tokens.json()['access_token']}"}
+    stale_cookie = anon_client.cookies.get("session")
+    transport = RecordingTransport()
+    users_app.state.sm.invalidation.set_transport(transport)
+    path = f"/api/users/admin/{user.id}"
+    if action == "roles":
+        response = await admin_client.put(f"{path}/roles", json={"role_names": ["user"]})
+    else:
+        response = await admin_client.patch(f"{path}/{action}")
+    assert response.status_code == 200, response.text
+    updated = await user_row(users_app, user.email)
+    assert updated.session_version == user.session_version + 1
+    assert read_session_version(user.id) == (False, None)
+    assert (await anon_client.get("/api/users/admin")).status_code == 401
+
+    assert (await admin_client.get("/api/users/admin", headers=bearer)).status_code == 401
+
+    # Simulate a remote worker with its own stale counter and bus.
+    anon_client.cookies.clear()
+    anon_client.cookies.set("session", stale_cookie)
+    remote = InvalidationBus()
+    subscribe(remote)
+    store_session_version(user.id, user.session_version)
+    assert transport.messages
+    for message in transport.messages:
+        await remote.deliver(message)
+    assert read_session_version(user.id) == (False, None)
+    assert (await anon_client.get("/api/users/admin")).status_code == 401
+    if action == "roles":
+        login = await anon_client.post(
+            "/api/users/auth/login",
+            data={"username": alice.email, "password": "UserPass1!"},
+        )
+        assert login.status_code == 204
+        assert (await anon_client.get("/api/users/admin")).status_code == 403
+
+
+async def test_rollback_does_not_publish_revocation(users_app, alice):
+    from unittest.mock import Mock
+
+    from users.admin.service import UserService
+
+    transport = RecordingTransport()
+    users_app.state.sm.invalidation.set_transport(transport)
+    store_session_version(alice.id, alice.session_version)
+    async with users_app.state.sm.db.session_factory() as session:
+        service = UserService(session, Mock(), users_app)
+        changed = await service.disable(alice.id)
+        assert changed.session_version == alice.session_version + 1
+        assert read_session_version(alice.id) == (True, alice.session_version)
+        assert transport.messages == []
+        await session.rollback()
+    unchanged = await user_row(users_app, alice.email)
+    assert unchanged.is_active
+    assert unchanged.session_version == alice.session_version
+    assert transport.messages == []
diff --git a/modules/users/users/admin/queries.py b/modules/users/users/admin/queries.py
index 1dc494244..f8e7b30fb 100644
--- a/modules/users/users/admin/queries.py
+++ b/modules/users/users/admin/queries.py
@@ -6,6 +6,7 @@
 from datetime import UTC, datetime, timedelta
 from typing import Any, cast
 
+from fastapi import FastAPI
 from simple_module_db import LIKE_ESCAPE_CHAR, like_contains_pattern
 from simple_module_db.pagination import PageParams
 from sqlalchemy import func, or_, select
@@ -17,6 +18,8 @@
 from users.exceptions import UserNotFoundError
 from users.manager import UserManager
 from users.models import Role, User, UserRole
+from users.session_revocation import bump_session_version, publish_revocation
+from users.session_version_cache import forget_session_version
 
 
 def _live_lock(locked_until: datetime | None) -> datetime | None:
@@ -35,9 +38,21 @@ def __init__(
         self,
         db: AsyncSession,
         user_manager: UserManager,
+        app: FastAPI | None = None,
     ) -> None:
         self._db = db
         self._manager = user_manager
+        self._app = app
+
+    async def _revoke_sessions(self, user: User) -> None:
+        """Persist the counter with the write; evict only after it commits."""
+        await bump_session_version(self._db, user)
+        user_id = user.id
+        app = self._app
+        if app is None:
+            self._db.on_commit(lambda: forget_session_version(user_id))
+        else:
+            self._db.on_commit(lambda: publish_revocation(app, user_id))
 
     # ── Helpers ─────────────────────────────────────────────────
 
diff --git a/modules/users/users/admin/service.py b/modules/users/users/admin/service.py
index 991caf6a3..d0ee7c8b0 100644
--- a/modules/users/users/admin/service.py
+++ b/modules/users/users/admin/service.py
@@ -6,7 +6,7 @@
 import uuid
 from datetime import UTC, datetime
 
-from sqlalchemy import delete, func, select
+from sqlalchemy import delete, func, select, update
 from sqlalchemy.orm import noload
 
 from users.admin.queries import _UserServiceBase
@@ -210,6 +210,15 @@ async def disable(self, user_id: uuid.UUID) -> User:
         user = await self._require_user(user_id)
         user.disabled_at = datetime.now(UTC)
         user.is_active = False
+        await self._revoke_sessions(user)
+        # Refresh tokens are spent too, so re-enabling the account later does not
+        # revive credentials issued before it was disabled (as the self-service
+        # "sign out everywhere" does).
+        await self._db.execute(
+            update(RefreshToken)
+            .where(RefreshToken.user_id == user.id, RefreshToken.revoked_at.is_(None))  # ty: ignore[invalid-argument-type]
+            .values(revoked_at=datetime.now(UTC))
+        )
         await self._db.flush()
         return user
 
@@ -217,6 +226,7 @@ async def enable(self, user_id: uuid.UUID) -> User:
         user = await self._require_user(user_id)
         user.disabled_at = None
         user.is_active = True
+        await self._revoke_sessions(user)
         await self._db.flush()
         return user
 
@@ -259,6 +269,7 @@ async def set_roles(
                 )
             )
 
+        await self._revoke_sessions(user)
         await self._db.flush()
         await self._db.refresh(user, attribute_names=["roles"])
         return user
diff --git a/modules/users/users/auth_local/self_account.py b/modules/users/users/auth_local/self_account.py
index da4597d13..c236fd31b 100644
--- a/modules/users/users/auth_local/self_account.py
+++ b/modules/users/users/auth_local/self_account.py
@@ -29,7 +29,7 @@
 from users.deps import fastapi_users, get_user_manager
 from users.manager import UserManager
 from users.models import RefreshToken, UserAccessToken
-from users.session_revocation import publish_revocation
+from users.session_revocation import bump_session_version, publish_revocation
 
 router = APIRouter()
 
@@ -93,10 +93,7 @@ async def change_my_password(
     except fu_exceptions.InvalidPasswordException as exc:
         raise HTTPException(status_code=400, detail=exc.reason) from exc
 
-    # ``user`` is attached to this request's session (``get_user_db`` takes the
-    # same ``get_db``), so assigning is enough to mark it written — a Core
-    # UPDATE would be rolled back as a read-only request.
-    user.session_version = int(user.session_version or 0) + 1
+    await bump_session_version(db, user)
     request.session[SESSION_VERSION_KEY] = user.session_version
     # Every worker caches the counter; without this they would keep honouring
     # the sessions this change was meant to strand. Hung on the commit rather
@@ -168,12 +165,7 @@ async def revoke_all_my_sessions(
     session that ordered the revocation as the exception to it.
     """
     now = datetime.now(UTC)
-    # Bumped through the ORM rather than as a bulk UPDATE: a Core statement
-    # does not mark the request session as written, so ``get_db`` would treat
-    # this handler as read-only and roll the whole revocation back. ``user``
-    # is already attached to this session — ``get_user_db`` takes the same
-    # ``get_db`` dependency — so the assignment is enough.
-    user.session_version = int(user.session_version or 0) + 1
+    await bump_session_version(db, user)
     # See the password change: the cached counter has to go with the bump, or
     # the workers keep letting the sessions this just revoked back in — and it
     # has to go *after* the commit, or a rollback restores the old counter
diff --git a/modules/users/users/deps.py b/modules/users/users/deps.py
index 564992652..cec16fa49 100644
--- a/modules/users/users/deps.py
+++ b/modules/users/users/deps.py
@@ -59,12 +59,13 @@ async def get_event_bus(request: Request) -> EventBus:
 
 
 async def get_user_service(
+    request: Request,
     db: AsyncSession = Depends(get_db),
     user_manager: UserManager = Depends(get_user_manager),
 ) -> UserService:
     from users.admin.service import UserService
 
-    return UserService(db, user_manager)
+    return UserService(db, user_manager, request.app)
 
 
 __all__ = [
diff --git a/modules/users/users/session_revocation.py b/modules/users/users/session_revocation.py
index faf3a2b8c..a898f1f75 100644
--- a/modules/users/users/session_revocation.py
+++ b/modules/users/users/session_revocation.py
@@ -20,20 +20,41 @@
 import uuid
 from typing import TYPE_CHECKING
 
+from users.models import User
 from users.session_version_cache import clear_session_version_cache, forget_session_version
 
 if TYPE_CHECKING:
     from fastapi import FastAPI
     from simple_module_core.invalidation import Invalidation, InvalidationBus
+    from sqlalchemy.ext.asyncio import AsyncSession
 
 logger = logging.getLogger(__name__)
 
-__all__ = ["CHANNEL", "apply_invalidation", "publish_revocation", "subscribe"]
+__all__ = [
+    "CHANNEL",
+    "apply_invalidation",
+    "bump_session_version",
+    "publish_revocation",
+    "subscribe",
+]
 
 CHANNEL = "users.session_version"
 """Carries one account's user id, or ``None`` for "forget every account"."""
 
 
+async def bump_session_version(db: AsyncSession, user: User) -> None:
+    """Increment in SQL, then load the counter for session/token minting.
+
+    The ORM assignment marks the request as written. A Python increment would
+    lose a revocation when two requests loaded the same old counter.
+    """
+    user.session_version = User.session_version + 1
+    await db.flush()
+    # SQL expressions expire the attribute on flush; async callers must not
+    # trigger implicit IO when they read the new counter later in the request.
+    await db.refresh(user, attribute_names=["session_version"])
+
+
 def _cache_key(key: str):
     """Turn a wire key back into what the cache is keyed by.