diff --git a/backend/app/gateway/auth/errors.py b/backend/app/gateway/auth/errors.py
index f4a68ae101..5e588eeef9 100644
--- a/backend/app/gateway/auth/errors.py
+++ b/backend/app/gateway/auth/errors.py
@@ -22,6 +22,7 @@ class AuthErrorCode(StrEnum):
     NOT_AUTHENTICATED = "not_authenticated"
     SYSTEM_ALREADY_INITIALIZED = "system_already_initialized"
     REGISTRATION_DISABLED = "registration_disabled"
+    ACCOUNT_DISABLED = "account_disabled"
 
 
 class TokenError(StrEnum):
diff --git a/backend/app/gateway/auth/local_provider.py b/backend/app/gateway/auth/local_provider.py
index be1f6c6897..ab85a4afd2 100644
--- a/backend/app/gateway/auth/local_provider.py
+++ b/backend/app/gateway/auth/local_provider.py
@@ -40,6 +40,14 @@ class LocalAuthProvider(AuthProvider):
         if user is None:
             return None
 
+        # Operator-disabled account (#3462 gap 3): no password is even
+        # compared — the failure is indistinguishable from unknown-credentials
+        # to the caller, so nothing about the account state leaks through the
+        # login path (the distinct ACCOUNT_DISABLED code is surfaced only by
+        # the session dependency once a token exists).
+        if getattr(user, "disabled", False):
+            return None
+
         if user.password_hash is None:
             # OAuth user without local password
             return None
diff --git a/backend/app/gateway/auth/models.py b/backend/app/gateway/auth/models.py
index 008f6acef7..d935fb44d5 100644
--- a/backend/app/gateway/auth/models.py
+++ b/backend/app/gateway/auth/models.py
@@ -34,6 +34,10 @@ class User(BaseModel):
     # Auth lifecycle
     needs_setup: bool = Field(default=False, description="True when a reset account must complete setup")
     token_version: int = Field(default=0, description="Incremented on password change to invalidate old JWTs")
+    # Account lifecycle (RFC #4063 / #3462 gap 3): a disabled account is
+    # rejected at every authentication surface; the row (credentials, role,
+    # OAuth linkage) is retained so re-enabling restores what was suspended.
+    disabled: bool = Field(default=False)
 
 
 class UserResponse(BaseModel):
@@ -42,6 +46,7 @@ class UserResponse(BaseModel):
     id: str
     email: str
     system_role: str = Field(..., min_length=1)
+    disabled: bool = False
     needs_setup: bool = False
     oauth_provider: str | None = Field(None, description="OAuth/SSO provider ID if the user logged in via SSO (e.g. 'keycloak')")
     permissions: list[str] | None = Field(
diff --git a/backend/app/gateway/auth/pat.py b/backend/app/gateway/auth/pat.py
index 8ae44314d4..4c71a69501 100644
--- a/backend/app/gateway/auth/pat.py
+++ b/backend/app/gateway/auth/pat.py
@@ -226,6 +226,11 @@ async def authenticate_pat(app: Any, authorization: str | None) -> tuple[Any, fr
         # dead even though its row survives (deleting a user revokes their
         # PATs, without needing a FK cascade).
         raise HTTPException(status_code=401, detail="Invalid token")
+    if getattr(user, "disabled", False):
+        # Operator-disabled account (#3462 gap 3): the token stays valid on
+        # disk but resolves nothing while the account is suspended; the same
+        # generic 401 keeps the verdict from serving as an oracle.
+        raise HTTPException(status_code=401, detail="Invalid token")
     await pat_repo.touch_last_used(str(record["id"]))
     return user, frozenset(record.get("scopes") or ())
 
diff --git a/backend/app/gateway/auth/repositories/base.py b/backend/app/gateway/auth/repositories/base.py
index b2924b1404..c849971347 100644
--- a/backend/app/gateway/auth/repositories/base.py
+++ b/backend/app/gateway/auth/repositories/base.py
@@ -14,6 +14,16 @@ class LastAdminRemainsError(RuntimeError):
     """
 
 
+class LastActiveAdminError(RuntimeError):
+    """Raised when an account write would disable the last active admin.
+
+    "Active" = an enabled admin row: disabled admins cannot authenticate, so
+    they cannot restore the deployment's management access. Raised inside the
+    same serialized transaction that would perform the write (see
+    ``LastAdminRemainsError`` for the race this closes).
+    """
+
+
 class UserNotFoundError(LookupError):
     """Raised when a user repository operation targets a non-existent row.
 
@@ -90,6 +100,17 @@ class UserRepository(ABC):
         raise NotImplementedError
 
     @abstractmethod
+    async def set_disabled(self, user_id: str, disabled: bool) -> User:
+        """Enable/disable an account as a single-column, serialized write.
+
+        Origin: admin user-management surface (RFC #4063 / #3462 gap 3).
+        Field-scoped like :meth:`update_system_role` — account-lifecycle
+        state has exactly one writer, and credential/role writers never
+        touch it. Disabling the last remaining ACTIVE admin raises
+        :class:`LastActiveAdminError` (re-enabling is always allowed).
+        """
+        raise NotImplementedError
+
     async def update_system_role(self, user_id: str, system_role: str) -> User:
         """Assign a user's system role as a single-column, serialized write.
 
diff --git a/backend/app/gateway/auth/repositories/sqlite.py b/backend/app/gateway/auth/repositories/sqlite.py
index f4e3c5a5b9..f0f6e7640b 100644
--- a/backend/app/gateway/auth/repositories/sqlite.py
+++ b/backend/app/gateway/auth/repositories/sqlite.py
@@ -21,7 +21,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from app.gateway.auth.models import User
-from app.gateway.auth.repositories.base import LastAdminRemainsError, UserNotFoundError, UserRepository
+from app.gateway.auth.repositories.base import LastActiveAdminError, LastAdminRemainsError, UserNotFoundError, UserRepository
 from deerflow.persistence.user.model import OAUTH_IDENTITY_INDEX_NAME, UserRow
 
 # ``email`` is ``mapped_column(unique=True, index=True)``, which SQLAlchemy
@@ -161,6 +161,7 @@ class SQLiteUserRepository(UserRepository):
             oauth_id=row.oauth_id,
             needs_setup=row.needs_setup,
             token_version=row.token_version,
+            disabled=row.disabled,
         )
 
     @staticmethod
@@ -175,6 +176,7 @@ class SQLiteUserRepository(UserRepository):
             oauth_id=user.oauth_id,
             needs_setup=user.needs_setup,
             token_version=user.token_version,
+            disabled=user.disabled,
         )
 
     # ── CRUD ──────────────────────────────────────────────────────────
@@ -315,13 +317,34 @@ class SQLiteUserRepository(UserRepository):
             if row is None:
                 raise UserNotFoundError(f"User {user_id} no longer exists")
             if row.system_role == "admin" and system_role != "admin":
-                admin_count = await session.scalar(select(func.count()).select_from(UserRow).where(UserRow.system_role == "admin"))
-                if admin_count is None or admin_count <= 1:
+                # Count ACTIVE admins only: a disabled admin cannot
+                # authenticate, so counting them would let "disable B, then
+                # demote self" (or a combined role+disable request) strand
+                # the deployment with zero usable management credentials.
+                active_admins = await session.scalar(select(func.count()).select_from(UserRow).where(UserRow.system_role == "admin", UserRow.disabled.is_(False)))
+                if active_admins is None or active_admins <= 1:
                     raise LastAdminRemainsError("cannot demote the last remaining admin")
             row.system_role = system_role
             await session.commit()
             return self._row_to_user(row)
 
+    async def set_disabled(self, user_id: str, disabled: bool) -> User:
+        """Serialized single-column account enable/disable (see base contract)."""
+        async with self._sf() as session:
+            # Same serialization idiom as update_system_role: the
+            # active-admin count and the column write share one transaction.
+            await self._serialize_account_role_mutation(session)
+            row = await session.get(UserRow, user_id)
+            if row is None:
+                raise UserNotFoundError(f"User {user_id} no longer exists")
+            if disabled and not row.disabled and row.system_role == "admin":
+                active_admins = await session.scalar(select(func.count()).select_from(UserRow).where(UserRow.system_role == "admin", UserRow.disabled.is_(False)))
+                if active_admins is None or active_admins <= 1:
+                    raise LastActiveAdminError("cannot disable the last remaining active admin")
+            row.disabled = disabled
+            await session.commit()
+            return self._row_to_user(row)
+
     @staticmethod
     async def _serialize_account_role_mutation(session: AsyncSession) -> None:
         """Serialize role mutations before the admin count is read.
@@ -375,6 +398,9 @@ class SQLiteUserRepository(UserRepository):
             # Role changes go through update_system_role, which likewise
             # never touches credentials.
             user.system_role = row.system_role
+            # disabled is account-lifecycle state: only set_disabled writes
+            # it (same field-scoping as the role column).
+            user.disabled = row.disabled
             row.oauth_provider = user.oauth_provider
             row.oauth_id = user.oauth_id
             row.needs_setup = user.needs_setup
diff --git a/backend/app/gateway/auth/user_provisioning.py b/backend/app/gateway/auth/user_provisioning.py
index 228856c5bd..bbeda0beed 100644
--- a/backend/app/gateway/auth/user_provisioning.py
+++ b/backend/app/gateway/auth/user_provisioning.py
@@ -38,6 +38,14 @@ async def get_or_provision_oidc_user(
     # 1. Existing OAuth link
     existing = await local_provider.get_user_by_oauth(provider_id, identity.subject)
     if existing:
+        if getattr(existing, "disabled", False):
+            # Operator-disabled account (#3462 gap 3): the IdP authenticated
+            # the identity, but the account is suspended — the row stays
+            # linked so re-enabling restores access unchanged.
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="This account has been disabled. Please contact your administrator.",
+            )
         return {"user": existing, "created": False}
 
     # 2. Verified email requirement
diff --git a/backend/app/gateway/auth_middleware.py b/backend/app/gateway/auth_middleware.py
index e20cd0d07a..b2c01e40b0 100644
--- a/backend/app/gateway/auth_middleware.py
+++ b/backend/app/gateway/auth_middleware.py
@@ -9,6 +9,7 @@ owner filtering works automatically via the sentinel pattern.
 Fine-grained permission checks remain in authz.py decorators.
 """
 
+import logging
 from collections.abc import Callable
 
 from fastapi import HTTPException, Request, Response
@@ -30,6 +31,8 @@ from app.gateway.internal_auth import INTERNAL_AUTH_HEADER_NAME, get_internal_us
 from app.gateway.request_path import get_request_route_path
 from deerflow.runtime.user_context import reset_current_user, set_current_user
 
+logger = logging.getLogger(__name__)
+
 # Paths that never require authentication.
 _PUBLIC_PATH_PREFIXES: tuple[str, ...] = (
     "/health",
@@ -104,6 +107,35 @@ class AuthMiddleware(BaseHTTPMiddleware):
             if owner_user_id:
                 owner_user_id = owner_user_id.strip()
             internal_user = get_internal_user(owner_user_id=owner_user_id or None)
+            if owner_user_id and not is_auth_disabled():
+                # The internal principal is synthesized from trusted headers
+                # without a users-row lookup, so suspension has no other chance
+                # to bite on this surface (#3462 gap 3): enforce it here, the
+                # one gate every owner-bound internal HTTP call (IM channel
+                # dispatch above all) passes through. Auth-disabled mode is
+                # exempt — its owner header carries the synthetic
+                # auth-disabled identity and no suspendable account exists.
+                # Owner ids without a users row stay allowed (a nonexistent
+                # account cannot be suspended), and a store that cannot be
+                # read cannot hold a suspension verdict either; run admission
+                # re-asserts the check when the row is readable.
+                from app.gateway.deps import get_local_provider
+
+                try:
+                    owner = await get_local_provider().get_user(owner_user_id)
+                except Exception:
+                    logger.warning("Internal-owner suspension check skipped: users lookup failed", exc_info=True)
+                    owner = None
+                if owner is not None and getattr(owner, "disabled", False):
+                    return JSONResponse(
+                        status_code=401,
+                        content={
+                            "detail": AuthErrorResponse(
+                                code=AuthErrorCode.ACCOUNT_DISABLED,
+                                message="Account disabled",
+                            ).model_dump()
+                        },
+                    )
 
         auth_source = AUTH_SOURCE_SESSION
         access_token = request.cookies.get("access_token")
diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py
index 7f77e4b946..0ae9fd18f9 100644
--- a/backend/app/gateway/deps.py
+++ b/backend/app/gateway/deps.py
@@ -28,6 +28,7 @@ from typing import TYPE_CHECKING, Any, TypeVar, cast
 from fastapi import FastAPI, HTTPException, Request
 from langgraph.types import Checkpointer
 
+from app.gateway.auth.errors import AuthErrorCode
 from deerflow.community.browser_automation.session import browser_multi_worker_error
 from deerflow.config.app_config import AppConfig, get_app_config
 from deerflow.config.deployment_config import multi_instance_declaration
@@ -1126,16 +1127,38 @@ async def get_current_user_from_request(request: Request):
             detail=AuthErrorResponse(code=AuthErrorCode.USER_NOT_FOUND, message="User not found").model_dump(),
         )
 
-    # Token version mismatch → password was changed, token is stale
-    if user.token_version != payload.ver:
+    # Operator-disabled account (#3462 gap 3) and stale token versions are
+    # both verdicts of the shared post-lookup validator: the password and
+    # PAT paths reject at their own surfaces; OAuth provisioning rejects at
+    # resolve time.
+    error = validate_resolved_session_user(user, payload)
+    if error is not None:
         raise HTTPException(
             status_code=401,
-            detail=AuthErrorResponse(code=AuthErrorCode.TOKEN_INVALID, message="Token revoked (password changed)").model_dump(),
+            detail=AuthErrorResponse(code=error, message="Account disabled" if error is AuthErrorCode.ACCOUNT_DISABLED else "Token revoked (password changed)").model_dump(),
         )
 
     return user
 
 
+def validate_resolved_session_user(user, payload) -> AuthErrorCode | None:
+    """Shared post-lookup session validation for EVERY JWT authenticator.
+
+    The Gateway's HTTP dependency, the WebSocket authenticator (browser
+    streaming bypasses AuthMiddleware), and the standalone LangGraph
+    ``authenticate`` callback all resolve cookie → JWT → user; this helper
+    is the one place the post-lookup verdicts live so a lifecycle change
+    (token_version bump, account suspension) lands everywhere at once.
+    Returns the failure code, or ``None`` when the session is valid —
+    callers map it to their own exception surface.
+    """
+    if user.token_version != payload.ver:
+        return AuthErrorCode.TOKEN_INVALID
+    if getattr(user, "disabled", False):
+        return AuthErrorCode.ACCOUNT_DISABLED
+    return None
+
+
 async def is_admin_user(request: Request) -> bool:
     """Return whether the authenticated caller is an admin user.
 
diff --git a/backend/app/gateway/langgraph_auth.py b/backend/app/gateway/langgraph_auth.py
index 0c035763f6..c7745d2398 100644
--- a/backend/app/gateway/langgraph_auth.py
+++ b/backend/app/gateway/langgraph_auth.py
@@ -276,10 +276,17 @@ async def authenticate(request):
             status_code=401,
             detail="User not found",
         )
-    if user.token_version != payload.ver:
+    # Same post-lookup verdicts as every other JWT surface — including
+    # account suspension, so a disabled user's still-valid cookie is
+    # rejected here exactly as the Gateway's HTTP dependency rejects it.
+    from app.gateway.auth.errors import AuthErrorCode
+    from app.gateway.deps import validate_resolved_session_user
+
+    failure = validate_resolved_session_user(user, payload)
+    if failure is not None:
         raise Auth.exceptions.HTTPException(
             status_code=401,
-            detail="Token revoked (password changed)",
+            detail="Token revoked (password changed)" if failure is AuthErrorCode.TOKEN_INVALID else "Account disabled",
         )
 
     return payload.sub
diff --git a/backend/app/gateway/routers/admin_users.py b/backend/app/gateway/routers/admin_users.py
index d50837ccc9..ccf99bf177 100644
--- a/backend/app/gateway/routers/admin_users.py
+++ b/backend/app/gateway/routers/admin_users.py
@@ -21,7 +21,7 @@ from uuid import UUID
 from fastapi import APIRouter, Depends, HTTPException, Request, status
 from pydantic import BaseModel, Field
 
-from app.gateway.auth.repositories.base import LastAdminRemainsError, UserNotFoundError
+from app.gateway.auth.repositories.base import LastActiveAdminError, LastAdminRemainsError, UserNotFoundError
 from app.gateway.authz import assignable_role_names
 from app.gateway.deps import get_user_repository, require_admin_user
 
@@ -33,10 +33,16 @@ from app.gateway.auth.models import UserResponse
 router = APIRouter(prefix="/api/v1/admin/users", tags=["admin-users"])
 
 
-class UserRoleUpdate(BaseModel):
-    """Assignment payload: a built-in or provider-configured role name."""
+class UserAccountUpdate(BaseModel):
+    """Partial account update: role and/or enable-disable.
 
-    system_role: str = Field(..., min_length=1, max_length=16, description="Built-in (admin/user) or a role configured in the authorization provider's roles mapping")
+    Omitted fields are unchanged; each present field routes to its
+    field-scoped writer (update_system_role / set_disabled) so role,
+    credentials, and lifecycle state can never clobber one another.
+    """
+
+    system_role: str | None = Field(None, min_length=1, max_length=16, description="Built-in (admin/user) or a role configured in the authorization provider's roles mapping")
+    disabled: bool | None = Field(None, description="Suspend (true) or restore (false) the account; every authentication surface rejects a disabled account")
 
 
 def _to_response(user) -> UserResponse:
@@ -46,6 +52,7 @@ def _to_response(user) -> UserResponse:
         system_role=str(user.system_role),
         needs_setup=bool(getattr(user, "needs_setup", False)),
         oauth_provider=getattr(user, "oauth_provider", None),
+        disabled=bool(getattr(user, "disabled", False)),
     )
 
 
@@ -58,38 +65,55 @@ async def list_users(request: Request, repository: SQLiteUserRepository = Depend
 
 
 @router.patch("/{user_id}")
-async def update_user_role(
+async def update_user_account(
     request: Request,
     user_id: UUID,
-    update: UserRoleUpdate,
+    update: UserAccountUpdate,
     repository: SQLiteUserRepository = Depends(get_user_repository),
 ) -> UserResponse:
-    """Assign a user's system role (admin-only).
+    """Update a user's account: assign a role and/or enable/disable (admin-only).
 
     The role must be a built-in ("admin"/"user") or declared in the
     authorization provider's roles configuration — an unknown name would make
     the user's principal fall through to ``default_role`` on every decision,
     which is never what an operator intended. Demoting the last remaining
-    admin is rejected so a deployment cannot lock itself out of management.
+    admin, and disabling the last remaining ACTIVE admin, are both rejected
+    so a deployment cannot lock itself out of management. Each present field
+    routes to its field-scoped, serialized writer; enforcement of the
+    disabled state happens at every authentication surface (password login,
+    JWT sessions, PATs, OAuth provisioning), not here.
     """
-    await require_admin_user(request, detail="role assignment requires an admin credential")
+    await require_admin_user(request, detail="account updates require an admin credential")
 
-    assignable = assignable_role_names()
-    if update.system_role not in assignable:
-        raise HTTPException(
-            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-            detail=f"unknown role '{update.system_role}'; assignable roles: {sorted(assignable)}",
-        )
+    if update.system_role is None and update.disabled is None:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="nothing to update: provide system_role and/or disabled")
 
+    if update.system_role is not None:
+        assignable = assignable_role_names()
+        if update.system_role not in assignable:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail=f"unknown role '{update.system_role}'; assignable roles: {sorted(assignable)}",
+            )
+
+    user: object | None = None
     try:
-        # Single serialized column write: the last-admin count and the write
-        # share one transaction (two concurrent demotions cannot both pass),
-        # and credential writers can never race this — see
-        # SQLiteUserRepository.update_system_role / update_user.
-        user = await repository.update_system_role(str(user_id), update.system_role)
+        if update.system_role is not None:
+            # Single serialized column write: the last-admin count and the
+            # write share one transaction (two concurrent demotions cannot
+            # both pass), and credential writers can never race this — see
+            # SQLiteUserRepository.update_system_role / update_user.
+            user = await repository.update_system_role(str(user_id), update.system_role)
+        if update.disabled is not None:
+            # Same serialization for the active-admin guard; re-enabling is
+            # always allowed.
+            user = await repository.set_disabled(str(user_id), update.disabled)
     except UserNotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") from None
     except LastAdminRemainsError:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="cannot demote the last remaining admin") from None
+    except LastActiveAdminError:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="cannot disable the last remaining active admin") from None
 
+    assert user is not None
     return _to_response(user)
diff --git a/backend/app/gateway/routers/browser.py b/backend/app/gateway/routers/browser.py
index b290dd3af2..d8d512ebbb 100644
--- a/backend/app/gateway/routers/browser.py
+++ b/backend/app/gateway/routers/browser.py
@@ -133,8 +133,15 @@ async def _authenticate_ws(websocket: WebSocket):
         if not isinstance(payload, TokenError):
             provider = get_local_provider()
             user = await provider.get_user(payload.sub)
-            if user is not None and user.token_version == payload.ver:
-                return user
+            if user is not None:
+                # Same post-lookup verdicts as every other JWT surface —
+                # including account suspension: WebSockets bypass
+                # AuthMiddleware, so a disabled user's still-valid cookie
+                # must not open a streaming/browser session.
+                from app.gateway.deps import validate_resolved_session_user
+
+                if validate_resolved_session_user(user, payload) is None:
+                    return user
     if is_auth_disabled():
         return get_auth_disabled_user()
     return None
diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py
index 4fe39d5dea..f20613606d 100644
--- a/backend/app/gateway/services.py
+++ b/backend/app/gateway/services.py
@@ -2048,6 +2048,19 @@ async def start_run(
         internal_owner_user = await resolve_trusted_internal_owner_for_attribution(request, owner_user_id)
         if internal_owner_user is None and owner_user_id == AUTH_DISABLED_USER_ID and is_auth_disabled():
             internal_owner_user = get_auth_disabled_user()
+        if getattr(internal_owner_user, "disabled", False):
+            # The scheduler's in-process launches (and MCP task-event
+            # delivery) fabricate requests that skip AuthMiddleware, where
+            # the internal-auth suspension gate lives. Re-assert it at run
+            # admission (#3462 gap 3): a disabled owner's occurrence fails
+            # here instead of driving the agent. The scheduler records the
+            # rejection as the occurrence's launch error.
+            from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse
+
+            raise HTTPException(
+                status_code=401,
+                detail=AuthErrorResponse(code=AuthErrorCode.ACCOUNT_DISABLED, message="Account disabled").model_dump(),
+            )
         inject_authenticated_user_context(
             config,
             request,
diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0039_user_disabled.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0039_user_disabled.py
new file mode 100644
index 0000000000..5eaf5c26cb
--- /dev/null
+++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0039_user_disabled.py
@@ -0,0 +1,45 @@
+"""Account disabled flag on ``users``.
+
+Revision ID: 0039_user_disabled
+Revises: 0038_channel_thread_bindings
+
+Adds ``users.disabled`` (NOT NULL, server default ``0``): an
+operator-disabled account is rejected at every authentication surface while
+the row — credentials, role, OAuth linkage — is retained, so re-enabling
+restores exactly what was suspended (RFC #4063 / issue #3462 gap 3).
+
+Column-guarded in both directions: an upgrade over a database that already
+has the column (an interrupted earlier attempt, or an empty-database
+bootstrap whose ``create_all`` saw the model first) completes, and a
+downgrade after the column is gone still moves the version row.
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0039_user_disabled"
+down_revision: str | None = "0038_channel_thread_bindings"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_TABLE = "users"
+_COLUMN = "disabled"
+
+
+def _has_column() -> bool:
+    bind = op.get_bind()
+    if not sa.inspect(bind).has_table(_TABLE):
+        return False
+    return _COLUMN in {column["name"] for column in sa.inspect(bind).get_columns(_TABLE)}
+
+
+def upgrade() -> None:
+    if not _has_column():
+        op.add_column(_TABLE, sa.Column(_COLUMN, sa.Boolean(), nullable=False, server_default=sa.false()))
+
+
+def downgrade() -> None:
+    if _has_column():
+        op.drop_column(_TABLE, _COLUMN)
diff --git a/backend/packages/harness/deerflow/persistence/user/model.py b/backend/packages/harness/deerflow/persistence/user/model.py
index a918f1f25e..570119cd86 100644
--- a/backend/packages/harness/deerflow/persistence/user/model.py
+++ b/backend/packages/harness/deerflow/persistence/user/model.py
@@ -13,7 +13,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, Index, String, text
+from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, Index, String, false, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
@@ -68,6 +68,12 @@ class UserRow(Base):
     needs_setup: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
     token_version: Mapped[int] = mapped_column(nullable=False, default=0)
 
+    # Account lifecycle (RFC #4063 / issue #3462 gap 3): an operator-disabled
+    # account is rejected at every authentication surface (password login,
+    # JWT session resolution, PAT, OAuth) — the row is retained with its
+    # credentials and role so re-enabling restores exactly what was suspended.
+    disabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=false())
+
     __table_args__ = (
         # sqlite_where alone is a SQLAlchemy dialect-specific kwarg -- it
         # does not apply on the postgresql dialect, so a table created
diff --git a/backend/tests/test_admin_users_router.py b/backend/tests/test_admin_users_router.py
index 0115c34176..7b4c5d52e8 100644
--- a/backend/tests/test_admin_users_router.py
+++ b/backend/tests/test_admin_users_router.py
@@ -50,6 +50,19 @@ class _FakeRepo:
         self.users[str(user.id)] = user
         return user
 
+    async def set_disabled(self, user_id: str, disabled: bool) -> User:
+        from app.gateway.auth.repositories.base import LastActiveAdminError, UserNotFoundError
+
+        user = self.users.get(user_id)
+        if user is None:
+            raise UserNotFoundError(f"User {user_id} no longer exists")
+        if disabled and not getattr(user, "disabled", False) and user.system_role == "admin":
+            if sum(1 for u in self.users.values() if u.system_role == "admin" and not getattr(u, "disabled", False)) <= 1:
+                raise LastActiveAdminError("cannot disable the last remaining active admin")
+        user.disabled = disabled
+        self.updates.append((user_id, f"disabled={disabled}"))
+        return user
+
     async def update_system_role(self, user_id: str, system_role: str) -> User:
         from app.gateway.auth.repositories.base import LastAdminRemainsError, UserNotFoundError
 
@@ -57,7 +70,9 @@ class _FakeRepo:
         if user is None:
             raise UserNotFoundError(f"User {user_id} no longer exists")
         if user.system_role == "admin" and system_role != "admin":
-            if await self.count_admin_users() <= 1:
+            # Active admins only (disabled admins cannot authenticate).
+            active = sum(1 for u in self.users.values() if u.system_role == "admin" and not getattr(u, "disabled", False))
+            if active <= 1:
                 raise LastAdminRemainsError("cannot demote the last remaining admin")
         user.system_role = system_role
         self.updates.append((user_id, system_role))
@@ -303,3 +318,277 @@ def test_assignment_reaches_principal_permissions():
     principal = build_principal_from_context({"user_id": "u1", "user_role": "guest"}, default_role="user")
 
     assert principal.role == "guest"
+
+
+def test_admin_disables_and_reenables_account(monkeypatch):
+    admin = _make_user(system_role="admin")
+    target = _make_user(system_role="guest")
+    repo = _FakeRepo([admin, target])
+    client = _make_client(monkeypatch, caller=admin, repo=repo, roles={"admin", "user", "guest"})
+
+    disabled = client.patch(f"/api/v1/admin/users/{target.id}", json={"disabled": True})
+    assert disabled.status_code == 200
+    assert disabled.json()["disabled"] is True
+
+    reenabled = client.patch(f"/api/v1/admin/users/{target.id}", json={"disabled": False})
+    assert reenabled.status_code == 200
+    assert reenabled.json()["disabled"] is False
+
+
+def test_cannot_disable_last_active_admin(monkeypatch):
+    admin = _make_user(system_role="admin")
+    repo = _FakeRepo([admin])
+    client = _make_client(monkeypatch, caller=admin, repo=repo)
+
+    response = client.patch(f"/api/v1/admin/users/{admin.id}", json={"disabled": True})
+
+    assert response.status_code == 409
+    assert repo.users[str(admin.id)].disabled is False
+
+
+def test_empty_account_update_rejected(monkeypatch):
+    admin = _make_user(system_role="admin")
+    target = _make_user()
+    repo = _FakeRepo([admin, target])
+    client = _make_client(monkeypatch, caller=admin, repo=repo)
+
+    assert client.patch(f"/api/v1/admin/users/{target.id}", json={}).status_code == 422
+
+
+def test_combined_role_and_disable(monkeypatch):
+    admin = _make_user(system_role="admin")
+    other = _make_user(system_role="admin", email="b@example.com")
+    target = _make_user()
+    repo = _FakeRepo([admin, other, target])
+    client = _make_client(monkeypatch, caller=admin, repo=repo, roles={"admin", "user", "guest"})
+
+    response = client.patch(f"/api/v1/admin/users/{target.id}", json={"system_role": "guest", "disabled": True})
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["system_role"] == "guest" and body["disabled"] is True
+
+
+def test_disable_then_demote_self_strands_no_admin(monkeypatch):
+    """[P1 regression] "Disable B, then demote self" and the combined
+    {system_role, disabled} self-update must both be rejected: the role
+    demotion counts ACTIVE admins only, so with B disabled the self-demotion
+    would strand zero usable management credentials."""
+    admin = _make_user(system_role="admin")
+    other = _make_user(system_role="admin", email="b@example.com")
+    repo = _FakeRepo([admin, other])
+    client = _make_client(monkeypatch, caller=admin, repo=repo, roles={"admin", "user"})
+
+    # Disable B first.
+    assert client.patch(f"/api/v1/admin/users/{other.id}", json={"disabled": True}).status_code == 200
+
+    # Self-demotion now strands: only A remains active.
+    demote = client.patch(f"/api/v1/admin/users/{admin.id}", json={"system_role": "user"})
+    assert demote.status_code == 409
+    assert repo.users[str(admin.id)].system_role == "admin"
+
+    # Combined self demote+disable has the same outcome (role write first,
+    # active-count guard fires).
+    combined = client.patch(f"/api/v1/admin/users/{admin.id}", json={"system_role": "user", "disabled": True})
+    assert combined.status_code == 409
+    assert repo.users[str(admin.id)].system_role == "admin"
+
+
+def test_shared_session_validator_verdicts():
+    """The one post-lookup verdict helper every JWT surface routes through."""
+    from types import SimpleNamespace as NS
+
+    from app.gateway.auth.errors import AuthErrorCode
+    from app.gateway.deps import validate_resolved_session_user
+
+    user = NS(token_version=3, disabled=False)
+    assert validate_resolved_session_user(user, NS(ver=3)) is None
+
+    stale = NS(token_version=4, disabled=False)
+    assert validate_resolved_session_user(stale, NS(ver=3)) is AuthErrorCode.TOKEN_INVALID
+
+    suspended = NS(token_version=3, disabled=True)
+    assert validate_resolved_session_user(suspended, NS(ver=3)) is AuthErrorCode.ACCOUNT_DISABLED
+
+
+def test_langgraph_authenticate_rejects_suspended_session():
+    """The standalone LangGraph authenticate callback rejects a still-valid
+    cookie whose account was suspended — admission, not just login."""
+    import asyncio
+    import tempfile
+
+    import pytest
+
+    from app.gateway.auth.jwt import create_access_token
+    from deerflow.persistence.engine import close_engine
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        repo = _make_sqlite_repo(tmpdir)
+
+        import app.gateway.deps as deps_module
+        from app.gateway import langgraph_auth
+
+        # Point the cached provider at THIS test's repo (module-level
+        # caches otherwise leak a previous test's closed engine), and put
+        # the previous values back afterwards so later tests that call
+        # get_local_provider() do not see this test's closed engine.
+        saved_repo = deps_module._cached_repo
+        saved_provider = deps_module._cached_local_provider
+        deps_module._cached_repo = repo
+        deps_module._cached_local_provider = None
+
+        async def _run():
+            user = await repo.create_user(User(email="lg@example.com", system_role="user"))
+            token = create_access_token(str(user.id), token_version=user.token_version)
+            await repo.set_disabled(str(user.id), True)
+
+            request = NS_stub_request(token)
+
+            with pytest.raises(langgraph_auth.Auth.exceptions.HTTPException) as exc_info:
+                await langgraph_auth.authenticate(request)
+            assert exc_info.value.status_code == 401
+            assert "disabled" in str(exc_info.value.detail).lower()
+
+        def NS_stub_request(token_value):
+            from types import SimpleNamespace as NS
+
+            return NS(cookies={"access_token": token_value}, headers={}, method="GET", url=NS(path="/"), client=NS(host="test"))
+
+        try:
+            asyncio.run(_run())
+        finally:
+            asyncio.run(close_engine())
+            deps_module._cached_repo = saved_repo
+            deps_module._cached_local_provider = saved_provider
+
+
+def test_browser_ws_authenticator_rejects_suspended_session():
+    """The WebSocket authenticator (browser streaming bypasses
+    AuthMiddleware) rejects a suspended account's still-valid cookie."""
+    import asyncio
+    import tempfile
+
+    from app.gateway.auth.jwt import create_access_token
+    from deerflow.persistence.engine import close_engine
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        repo = _make_sqlite_repo(tmpdir)
+
+        from types import SimpleNamespace as NS
+
+        import app.gateway.deps as deps_module
+        from app.gateway.routers.browser import _authenticate_ws
+
+        # Same cache hygiene as the authenticate test above: restore the
+        # module-level caches so this test's closed engine does not leak.
+        saved_repo = deps_module._cached_repo
+        saved_provider = deps_module._cached_local_provider
+        deps_module._cached_repo = repo
+        deps_module._cached_local_provider = None
+
+        async def _run():
+            user = await repo.create_user(User(email="ws@example.com", system_role="user"))
+            token = create_access_token(str(user.id), token_version=user.token_version)
+            await repo.set_disabled(str(user.id), True)
+
+            websocket = NS(cookies={"access_token": token}, headers={})
+            assert await _authenticate_ws(websocket) is None
+
+        try:
+            asyncio.run(_run())
+        finally:
+            asyncio.run(close_engine())
+            deps_module._cached_repo = saved_repo
+            deps_module._cached_local_provider = saved_provider
+
+
+def test_repo_set_disabled_field_scoped_and_last_active_admin():
+    """Real-repo slice: set_disabled is a single-column write (role and
+    credentials untouched; the credential writer never touches lifecycle
+    state), and the last-ACTIVE-admin guard fires inside the serialized
+    write — a disabled admin does not count as active, and re-enabling is
+    always allowed."""
+    import asyncio
+    import tempfile
+
+    from app.gateway.auth.repositories.base import LastActiveAdminError
+    from deerflow.persistence.engine import close_engine
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        repo = _make_sqlite_repo(tmpdir)
+
+        async def _run():
+            first = await repo.create_user(User(email="a@example.com", system_role="admin", password_hash="h1"))
+            second = await repo.create_user(User(email="b@example.com", system_role="admin", password_hash="h2"))
+
+            disabled_first = await repo.set_disabled(str(first.id), True)
+            assert disabled_first.disabled is True
+            assert disabled_first.password_hash == "h1"
+            assert disabled_first.system_role == "admin"
+
+            # The credential writer never touches lifecycle state: a stale
+            # snapshot with disabled=False does not re-enable the account.
+            stale = User(id=first.id, email=first.email, password_hash="h3", system_role="admin")
+            after = await repo.update_user(stale)
+            assert after.password_hash == "h3"
+            assert after.disabled is True
+
+            # Only one ACTIVE admin remains: disabling it raises, row intact.
+            try:
+                await repo.set_disabled(str(second.id), True)
+            except LastActiveAdminError:
+                pass
+            else:
+                raise AssertionError("disabling the last active admin must raise")
+            assert (await repo.get_user_by_id(str(second.id))).disabled is False
+
+            restored = await repo.set_disabled(str(first.id), False)
+            assert restored.disabled is False
+
+            # The role-demotion guard counts ACTIVE admins only: disable
+            # one of two, then demoting the remaining active admin raises.
+            await repo.set_disabled(str(first.id), True)
+            from app.gateway.auth.repositories.base import LastAdminRemainsError as _LAR
+
+            try:
+                await repo.update_system_role(str(second.id), "user")
+            except _LAR:
+                pass
+            else:
+                raise AssertionError("demoting the last ACTIVE admin must raise")
+            assert (await repo.get_user_by_id(str(second.id))).system_role == "admin"
+
+        try:
+            asyncio.run(_run())
+        finally:
+            asyncio.run(close_engine())
+
+
+def test_disabled_account_rejected_at_login_and_flagged_for_resolvers():
+    """Password login never compares credentials for a disabled account,
+    and lookups return the flag so the JWT resolver can reject."""
+    import asyncio
+    import tempfile
+
+    from app.gateway.auth.local_provider import LocalAuthProvider
+    from deerflow.persistence.engine import close_engine
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        repo = _make_sqlite_repo(tmpdir)
+
+        async def _run():
+            from app.gateway.auth.password import hash_password
+
+            user = await repo.create_user(User(email="d@example.com", system_role="user", password_hash=hash_password("correct-horse")))
+            await repo.set_disabled(str(user.id), True)
+            provider = LocalAuthProvider(repository=repo)
+
+            # The CORRECT password: only the disabled gate can reject it.
+            assert await provider.authenticate({"email": "d@example.com", "password": "correct-horse"}) is None
+            fetched = await provider.get_user(str(user.id))
+            assert fetched is not None and fetched.disabled is True
+
+        try:
+            asyncio.run(_run())
+        finally:
+            asyncio.run(close_engine())
diff --git a/backend/tests/test_auth_middleware.py b/backend/tests/test_auth_middleware.py
index e1a8c371a6..6253081b8f 100644
--- a/backend/tests/test_auth_middleware.py
+++ b/backend/tests/test_auth_middleware.py
@@ -447,6 +447,113 @@ def test_protected_post_with_internal_auth_header_passes():
     assert res.status_code == 200
 
 
+def test_internal_auth_owner_header_of_suspended_account_rejected(tmp_path):
+    """(#3462 gap 3) IM-channel dispatch surface: the internal principal is
+    synthesized from trusted headers without a users-row lookup, so the
+    middleware itself rejects an owner-bound internal call whose account was
+    suspended — at the very next request, like every other surface."""
+    import asyncio
+    import json
+
+    from starlette.requests import Request as StarletteRequest
+    from starlette.responses import PlainTextResponse
+
+    from app.gateway.auth.models import User
+    from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
+    from app.gateway.auth_middleware import AuthMiddleware
+    from app.gateway.internal_auth import create_internal_auth_headers
+    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+
+    async def _dispatch(owner_user_id: str):
+        headers = create_internal_auth_headers(owner_user_id=owner_user_id)
+        scope = {
+            "type": "http",
+            "method": "POST",
+            "path": "/api/threads/abc/runs/stream",
+            "root_path": "",
+            "query_string": b"",
+            "headers": [(name.lower().encode(), value.encode()) for name, value in headers.items()],
+        }
+
+        async def call_next(_request):
+            return PlainTextResponse("route-reached")
+
+        response = await AuthMiddleware(None).dispatch(StarletteRequest(scope), call_next)
+        body = json.loads(response.body) if response.status_code != 200 else {}
+        return response.status_code, body
+
+    async def _run():
+        await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path}/users.db", sqlite_dir=str(tmp_path))
+        repo = SQLiteUserRepository(get_session_factory())
+        owner = await repo.create_user(User(email="channel-owner@example.com", system_role="user"))
+        await repo.set_disabled(str(owner.id), True)
+
+        import app.gateway.deps as deps_module
+
+        saved_repo, saved_provider = deps_module._cached_repo, deps_module._cached_local_provider
+        deps_module._cached_repo = repo
+        deps_module._cached_local_provider = None
+        try:
+            status_code, body = await _dispatch(str(owner.id))
+            assert status_code == 401
+            assert body["detail"]["code"] == "account_disabled"
+
+            # Unknown owner ids stay allowed: an account that does not exist
+            # in the users store cannot be suspended.
+            status_code, body = await _dispatch("no-such-owner")
+            assert status_code == 200
+
+            # Re-enabling restores the internal surface immediately.
+            await repo.set_disabled(str(owner.id), False)
+            status_code, body = await _dispatch(str(owner.id))
+            assert status_code == 200
+        finally:
+            deps_module._cached_repo = saved_repo
+            deps_module._cached_local_provider = saved_provider
+
+    try:
+        asyncio.run(_run())
+    finally:
+        asyncio.run(close_engine())
+
+
+def test_internal_auth_owner_gate_skipped_in_auth_disabled_mode(monkeypatch):
+    """Auth-disabled mode sends the synthetic auth-disabled identity as the
+    internal owner; deployments may have no users store at all there, so the
+    suspension gate must not attempt a lookup (it would fail the request)."""
+    import asyncio
+
+    from starlette.requests import Request as StarletteRequest
+    from starlette.responses import PlainTextResponse
+
+    from app.gateway.auth_disabled import AUTH_DISABLED_USER_ID
+    from app.gateway.auth_middleware import AuthMiddleware
+    from app.gateway.internal_auth import create_internal_auth_headers
+
+    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
+
+    async def _run():
+        headers = create_internal_auth_headers(owner_user_id=AUTH_DISABLED_USER_ID)
+        scope = {
+            "type": "http",
+            "method": "POST",
+            "path": "/api/threads/abc/runs/stream",
+            "root_path": "",
+            "query_string": b"",
+            "headers": [(name.lower().encode(), value.encode()) for name, value in headers.items()],
+        }
+
+        async def call_next(_request):
+            return PlainTextResponse("route-reached")
+
+        # No engine / users store is configured on purpose: a lookup here
+        # would raise and 500 the request.
+        response = await AuthMiddleware(None).dispatch(StarletteRequest(scope), call_next)
+        assert response.status_code == 200
+
+    asyncio.run(_run())
+
+
 def test_protected_post_with_non_ascii_internal_auth_header_returns_401(client):
     from app.gateway.internal_auth import INTERNAL_AUTH_HEADER_NAME
 
diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py
index cf7d5c7217..ae17ee9df5 100644
--- a/backend/tests/test_gateway_services.py
+++ b/backend/tests/test_gateway_services.py
@@ -24,16 +24,19 @@ def _stub_app_config():
     reset_app_config()
 
 
-def _make_start_run_request(run_manager, *, thread_store=None, auth_source=None):
+def _make_start_run_request(run_manager, *, thread_store=None, auth_source=None, state_user=None, headers=None):
     from langgraph.checkpoint.memory import InMemorySaver
     from langgraph.store.memory import InMemoryStore
 
     from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
 
     store = InMemoryStore()
+    state = SimpleNamespace(auth_source=auth_source)
+    if state_user is not None:
+        state.user = state_user
     return SimpleNamespace(
-        headers={},
-        state=SimpleNamespace(auth_source=auth_source),
+        headers=dict(headers or {}),
+        state=state,
         app=SimpleNamespace(
             state=SimpleNamespace(
                 stream_bridge=SimpleNamespace(),
@@ -912,6 +915,68 @@ async def test_system_role_rejected_before_run_admission(_stub_app_config, auth_
     assert await request.app.state.checkpointer.aget_tuple({"configurable": {"thread_id": "synthetic-rejected-thread"}}) is None
 
 
+@pytest.mark.asyncio
+async def test_start_run_rejects_suspended_internal_owner(_stub_app_config, tmp_path):
+    """(#3462 gap 3) Scheduler launches fabricate internal requests that skip
+    AuthMiddleware (where the internal-auth suspension gate lives), so run
+    admission re-asserts it: a suspended owner's occurrence fails before any
+    run record exists, and re-enabling the owner admits the run again."""
+    from unittest.mock import AsyncMock, patch
+
+    from fastapi import HTTPException
+
+    import app.gateway.deps as deps_module
+    from app.gateway.auth.models import User
+    from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
+    from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, get_internal_user
+    from app.gateway.run_models import RunCreateRequest
+    from app.gateway.services import start_run
+    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+    from deerflow.runtime import RunManager
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path}/users.db", sqlite_dir=str(tmp_path))
+    repo = SQLiteUserRepository(get_session_factory())
+    owner = await repo.create_user(User(email="sched-owner@example.com", system_role="user"))
+    await repo.set_disabled(str(owner.id), True)
+
+    saved_repo, saved_provider = deps_module._cached_repo, deps_module._cached_local_provider
+    deps_module._cached_repo = repo
+    deps_module._cached_local_provider = None
+
+    manager = RunManager(store=MemoryRunStore())
+
+    def _request():
+        return _make_start_run_request(
+            manager,
+            auth_source=AUTH_SOURCE_INTERNAL,
+            state_user=get_internal_user(),
+            headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: str(owner.id)},
+        )
+
+    try:
+        with patch("app.gateway.services.resolve_agent_factory", return_value=object()), patch("app.gateway.services.run_agent", new_callable=AsyncMock) as worker:
+            with pytest.raises(HTTPException) as error:
+                await start_run(RunCreateRequest(input={"messages": [{"role": "user", "content": "scheduled tick"}]}), "suspended-owner-thread", _request())
+            worker.assert_not_awaited()
+        assert error.value.status_code == 401
+        assert error.value.detail["code"] == "account_disabled"
+        assert await manager.list_by_thread("suspended-owner-thread", user_id=None) == []
+
+        # Re-enabling the owner re-admits internal launches (exact-restore
+        # semantics of the lifecycle: suspension is not destructive). The
+        # agent worker is scheduled as a task, so admission is proven by the
+        # created run record rather than a synchronous await.
+        await repo.set_disabled(str(owner.id), False)
+        with patch("app.gateway.services.resolve_agent_factory", return_value=object()), patch("app.gateway.services.run_agent", new_callable=AsyncMock):
+            record = await start_run(RunCreateRequest(input={"messages": [{"role": "user", "content": "scheduled tick"}]}), "suspended-owner-thread", _request())
+        assert record.run_id
+    finally:
+        await close_engine()
+        deps_module._cached_repo = saved_repo
+        deps_module._cached_local_provider = saved_provider
+
+
 @pytest.mark.asyncio
 async def test_start_run_preserves_system_messages_from_internal_auth(_stub_app_config):
     from langchain_core.messages import SystemMessage
diff --git a/backend/tests/test_migration_0027_0028_notification_outbox.py b/backend/tests/test_migration_0027_0028_notification_outbox.py
index 33b4a38678..24927cc02e 100644
--- a/backend/tests/test_migration_0027_0028_notification_outbox.py
+++ b/backend/tests/test_migration_0027_0028_notification_outbox.py
@@ -23,7 +23,7 @@ OUTBOX = "0027_notification_deliveries"
 PARKED = "0028_parked_attempts"
 SCHEDULER = "0029_scheduler_agent_tasks"
 CLAIM_TOKENS = "0030_notification_claim_tokens"
-HEAD = "0038_channel_thread_bindings"
+HEAD = "0039_user_disabled"
 PREVIOUS = "0026_mcp_task_lease_tokens"
 TABLE = "notification_deliveries"
 
diff --git a/backend/tests/test_migration_0029_scheduler_agent_tasks.py b/backend/tests/test_migration_0029_scheduler_agent_tasks.py
index 342d99f4ff..7dd5eacf61 100644
--- a/backend/tests/test_migration_0029_scheduler_agent_tasks.py
+++ b/backend/tests/test_migration_0029_scheduler_agent_tasks.py
@@ -23,7 +23,7 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
 from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
 
 REVISION = "0029_scheduler_agent_tasks"
-CURRENT_HEAD = "0038_channel_thread_bindings"
+CURRENT_HEAD = "0039_user_disabled"
 NEXT = "0030_notification_claim_tokens"
 PREVIOUS = "0028_parked_attempts"
 TASK_FIELDS = {"origin_thread_id", "goal_objective", "max_runs", "end_at", "standing_notes"}
diff --git a/backend/tests/test_migration_0032_activity_and_task_events.py b/backend/tests/test_migration_0032_activity_and_task_events.py
index 4ff73981eb..f0775dde08 100644
--- a/backend/tests/test_migration_0032_activity_and_task_events.py
+++ b/backend/tests/test_migration_0032_activity_and_task_events.py
@@ -22,7 +22,9 @@ AFTER_NEXT = "0034_run_event_seq_watermark"
 LOGIN_THROTTLE = "0035_login_throttle"
 IDEMPOTENCY = "0036_run_idempotency_request"
 DOCUMENT_SUMMARIES = "0037_project_document_summaries"
-CURRENT_HEAD = "0038_channel_thread_bindings"
+CHANNEL_BINDINGS = "0038_channel_thread_bindings"
+USER_DISABLED = "0039_user_disabled"
+CURRENT_HEAD = "0039_user_disabled"
 PREVIOUS = "0031_scheduled_streak_boundary"
 TABLES = {"thread_read_markers", "thread_read_versions", "scheduled_task_events"}
 INDEXES = {
@@ -42,7 +44,8 @@ async def test_0032_remains_in_the_single_migration_chain():
     assert script.get_revision(LOGIN_THROTTLE).down_revision == AFTER_NEXT
     assert script.get_revision(IDEMPOTENCY).down_revision == LOGIN_THROTTLE
     assert script.get_revision(DOCUMENT_SUMMARIES).down_revision == IDEMPOTENCY
-    assert script.get_revision(CURRENT_HEAD).down_revision == DOCUMENT_SUMMARIES
+    assert script.get_revision(CHANNEL_BINDINGS).down_revision == DOCUMENT_SUMMARIES
+    assert script.get_revision(USER_DISABLED).down_revision == CHANNEL_BINDINGS
     # alembic_version.version_num is VARCHAR(32).
     assert len(REVISION) <= 32
 
diff --git a/backend/tests/test_migration_0034_run_event_seq_watermark.py b/backend/tests/test_migration_0034_run_event_seq_watermark.py
index b3d8469883..2b4a39dfc5 100644
--- a/backend/tests/test_migration_0034_run_event_seq_watermark.py
+++ b/backend/tests/test_migration_0034_run_event_seq_watermark.py
@@ -22,7 +22,9 @@ PREVIOUS = "0033_batch_result_artifact"
 LOGIN_THROTTLE = "0035_login_throttle"
 IDEMPOTENCY = "0036_run_idempotency_request"
 DOCUMENT_SUMMARIES = "0037_project_document_summaries"
-CURRENT_HEAD = "0038_channel_thread_bindings"
+CHANNEL_BINDINGS = "0038_channel_thread_bindings"
+USER_DISABLED = "0039_user_disabled"
+CURRENT_HEAD = "0039_user_disabled"
 TABLE = "run_event_thread_seq"
 pytestmark = pytest.mark.asyncio
 
@@ -34,7 +36,8 @@ async def test_0034_remains_in_the_single_migration_chain():
     assert script.get_revision(LOGIN_THROTTLE).down_revision == REVISION
     assert script.get_revision(IDEMPOTENCY).down_revision == LOGIN_THROTTLE
     assert script.get_revision(DOCUMENT_SUMMARIES).down_revision == IDEMPOTENCY
-    assert script.get_revision(CURRENT_HEAD).down_revision == DOCUMENT_SUMMARIES
+    assert script.get_revision(CHANNEL_BINDINGS).down_revision == DOCUMENT_SUMMARIES
+    assert script.get_revision(USER_DISABLED).down_revision == CHANNEL_BINDINGS
     assert len(REVISION) <= 32
 
 
diff --git a/backend/tests/test_migration_0035_login_throttle.py b/backend/tests/test_migration_0035_login_throttle.py
index 46b69bdabc..e0b98fdce0 100644
--- a/backend/tests/test_migration_0035_login_throttle.py
+++ b/backend/tests/test_migration_0035_login_throttle.py
@@ -23,7 +23,9 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
 REVISION = "0035_login_throttle"
 IDEMPOTENCY = "0036_run_idempotency_request"
 DOCUMENT_SUMMARIES = "0037_project_document_summaries"
-CURRENT_HEAD = "0038_channel_thread_bindings"
+CHANNEL_BINDINGS = "0038_channel_thread_bindings"
+USER_DISABLED = "0039_user_disabled"
+CURRENT_HEAD = "0039_user_disabled"
 PREVIOUS = "0034_run_event_seq_watermark"
 TABLE = "login_throttle"
 COLUMNS = {"ip", "fail_count", "locked_at", "lock_duration_seconds", "updated_at"}
@@ -37,7 +39,8 @@ async def test_0035_remains_in_the_single_head_chain_after_0034():
     assert script.get_revision(REVISION).down_revision == PREVIOUS
     assert script.get_revision(IDEMPOTENCY).down_revision == REVISION
     assert script.get_revision(DOCUMENT_SUMMARIES).down_revision == IDEMPOTENCY
-    assert script.get_revision(CURRENT_HEAD).down_revision == DOCUMENT_SUMMARIES
+    assert script.get_revision(CHANNEL_BINDINGS).down_revision == DOCUMENT_SUMMARIES
+    assert script.get_revision(USER_DISABLED).down_revision == CHANNEL_BINDINGS
     # alembic_version.version_num is VARCHAR(32).
     assert len(REVISION) <= 32
 
diff --git a/backend/tests/test_migration_0037_project_document_summaries.py b/backend/tests/test_migration_0037_project_document_summaries.py
index 68bf35f8c0..81f8672ef3 100644
--- a/backend/tests/test_migration_0037_project_document_summaries.py
+++ b/backend/tests/test_migration_0037_project_document_summaries.py
@@ -18,14 +18,16 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
 
 REVISION = "0037_project_document_summaries"
 PREVIOUS = "0036_run_idempotency_request"
-CURRENT_HEAD = "0038_channel_thread_bindings"
+CHANNEL_BINDINGS = "0038_channel_thread_bindings"
+CURRENT_HEAD = "0039_user_disabled"
 pytestmark = pytest.mark.asyncio
 
 
 async def test_0037_chains_after_0036_and_is_single_head():
     script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR))
     assert script.get_heads() == [CURRENT_HEAD]
-    assert script.get_revision(CURRENT_HEAD).down_revision == REVISION
+    assert script.get_revision(CHANNEL_BINDINGS).down_revision == REVISION
+    assert script.get_revision(CURRENT_HEAD).down_revision == CHANNEL_BINDINGS
     assert script.get_revision(REVISION).down_revision == PREVIOUS
     # alembic_version.version_num is VARCHAR(32).
     assert len(REVISION) <= 32
diff --git a/backend/tests/test_migration_0038_channel_thread_bindings.py b/backend/tests/test_migration_0038_channel_thread_bindings.py
index 94225af37e..40f37744b5 100644
--- a/backend/tests/test_migration_0038_channel_thread_bindings.py
+++ b/backend/tests/test_migration_0038_channel_thread_bindings.py
@@ -22,15 +22,16 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
 
 REVISION = "0038_channel_thread_bindings"
 PREVIOUS = "0037_project_document_summaries"
+CURRENT_HEAD = "0039_user_disabled"
 TABLE = "channel_thread_bindings"
 COLUMNS = {"key", "channel_name", "chat_id", "topic_id", "thread_id", "user_id", "created_at", "updated_at"}
 INDEX = "ix_channel_thread_bindings_channel_chat"
 pytestmark = pytest.mark.asyncio
 
 
-async def test_0038_is_the_single_head_and_chains_after_0037():
+async def test_0038_remains_in_the_single_migration_chain():
     script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR))
-    assert script.get_heads() == [REVISION]
+    assert script.get_heads() == [CURRENT_HEAD]
     assert script.get_revision(REVISION).down_revision == PREVIOUS
     # alembic_version.version_num is VARCHAR(32).
     assert len(REVISION) <= 32
@@ -182,6 +183,8 @@ async def test_bootstrap_provisions_the_table_on_an_empty_database_and_is_idempo
         assert set(shape["columns"]) == COLUMNS
         assert INDEX in shape["indexes"]
         async with engine.connect() as conn:
-            assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == REVISION
+            # Bootstrap runs the chain to its head, which this test no longer
+            # owns once later revisions chain after 0038.
+            assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == CURRENT_HEAD
     finally:
         await engine.dispose()
diff --git a/frontend/src/core/auth/auth-disabled-user.ts b/frontend/src/core/auth/auth-disabled-user.ts
index 13a499ed2a..79191a0b19 100644
--- a/frontend/src/core/auth/auth-disabled-user.ts
+++ b/frontend/src/core/auth/auth-disabled-user.ts
@@ -5,6 +5,7 @@ export const AUTH_DISABLED_USER: User = {
   email: "default@test.local",
   system_role: "admin",
   needs_setup: false,
+  disabled: false,
   oauth_provider: null,
 };
 
diff --git a/frontend/src/core/auth/static-user.ts b/frontend/src/core/auth/static-user.ts
index be5583653b..692335d2f2 100644
--- a/frontend/src/core/auth/static-user.ts
+++ b/frontend/src/core/auth/static-user.ts
@@ -5,5 +5,6 @@ export const STATIC_WEBSITE_USER: User = {
   email: "static@example.local",
   system_role: "admin",
   needs_setup: false,
+  disabled: false,
   oauth_provider: null,
 };
diff --git a/frontend/src/core/auth/types.ts b/frontend/src/core/auth/types.ts
index a087a2d21f..bb8561ee5b 100644
--- a/frontend/src/core/auth/types.ts
+++ b/frontend/src/core/auth/types.ts
@@ -12,6 +12,10 @@ export const userSchema = z.object({
   // other value is simply non-admin.
   system_role: z.string().min(1),
   needs_setup: z.boolean().optional().default(false),
+  // Account lifecycle (RFC #4063 gap 3): true when the account is
+  // operator-suspended. Optional so pre-gap-3 backends (and test literals)
+  // parse unchanged — absent means not disabled.
+  disabled: z.boolean().optional(),
   oauth_provider: z.string().nullable().optional().default(null),
   // Effective route permissions (RFC #4063 Phase 4). Optional + nullable:
   // absent = pre-Phase-4 backend, null = credential-creation response that
diff --git a/frontend/tests/unit/core/auth/server.test.ts b/frontend/tests/unit/core/auth/server.test.ts
index 4b77745e3c..4770acf8d3 100644
--- a/frontend/tests/unit/core/auth/server.test.ts
+++ b/frontend/tests/unit/core/auth/server.test.ts
@@ -200,6 +200,7 @@ describe("userSchema system_role", () => {
         email: "g@example.com",
         system_role: "",
         needs_setup: false,
+        disabled: false,
         oauth_provider: null,
         permissions: null,
       }),