diff --git a/backend/app/gateway/routers/auth.py b/backend/app/gateway/routers/auth.py
index 701e1a83b..2b953f1d0 100644
--- a/backend/app/gateway/routers/auth.py
+++ b/backend/app/gateway/routers/auth.py
@@ -164,15 +164,47 @@ def _set_session_cookie(response: Response, token: str, request: Request, *, rem
 #
 # **Limitation**: with multi-worker deployments (e.g., gunicorn -w N), each
 # worker maintains its own lockout table, so an attacker effectively gets
-# N × _MAX_LOGIN_ATTEMPTS guesses before being locked out everywhere. For
+# N × max_login_attempts guesses before being locked out everywhere. For
 # production multi-worker setups, replace this with a shared store (Redis,
 # database-backed counter) to enforce a true per-IP limit.
+#
+# The policy values are operator-configurable via auth.local.max_login_attempts /
+# auth.local.lockout_seconds (read live per call, matching _local_registration_enabled,
+# so a config reload applies to the next login without a Gateway restart). The
+# no-config.yaml fallback is the LocalAuthConfig model defaults — a single source
+# of truth, not a second copy of the numbers.
 
-_MAX_LOGIN_ATTEMPTS = 5
-_LOCKOUT_SECONDS = 300  # 5 minutes
+# ip → (fail_count, locked_at, locked_duration). The stored duration always
+# matches the policy the lock was last evaluated under (its creation counts
+# as an evaluation, and every check that leaves the lock active commits the
+# then-current duration, decreases included): a lowered lockout_seconds
+# releases an active lock early, a raised one extends it — and a sentence
+# that already served the last-evaluated duration is never resurrected.
+_login_attempts: dict[str, tuple[int, float, float]] = {}
 
-# ip → (fail_count, lock_until_timestamp)
-_login_attempts: dict[str, tuple[int, float]] = {}
+
+def _login_throttle_policy() -> tuple[int, float]:
+    """(max_login_attempts, lockout_seconds) from auth.local config, read live.
+
+    Only ``FileNotFoundError`` falls back to the model defaults, matching
+    ``_local_registration_enabled``: ``config.yaml`` is absent in bare-app
+    contexts that never load it (tests build the gateway without one), and the
+    throttle must keep its pre-config-era behavior there. A malformed config
+    propagates instead — like every other config consumer, and so an operator
+    who tightened the policy never silently gets the more permissive defaults.
+
+    Callers on request paths resolve this at most once per helper invocation;
+    ``get_app_config`` re-hashes the config file on every call, and the login
+    endpoint is unauthenticated.
+    """
+    from deerflow.config.app_config import get_app_config
+    from deerflow.config.auth_config import LocalAuthConfig
+
+    try:
+        local = get_app_config().auth.local
+    except FileNotFoundError:
+        local = LocalAuthConfig()
+    return local.max_login_attempts, local.lockout_seconds
 
 
 def _trusted_proxies() -> list:
@@ -236,46 +268,127 @@ def _get_client_ip(request: Request) -> str:
     return peer_host or "unknown"
 
 
-def _check_rate_limit(ip: str) -> None:
-    """Raise 429 if the IP is currently locked out."""
+async def _check_rate_limit(ip: str) -> None:
+    """Raise 429 if the IP is currently locked out.
+
+    The record lookup comes before policy resolution on purpose: a clean IP
+    (no failed attempts recorded — the overwhelming majority of logins) must
+    not pay a config read, and ``get_app_config`` re-hashes config.yaml on
+    every call while this endpoint is unauthenticated. When a record exists
+    the policy is resolved off the event loop via ``asyncio.to_thread``:
+    every request from a recorded IP — including an already-locked attacker
+    flooding the endpoint — pays that read on the way to its answer, and the
+    stat + hash must not block the loop.
+    """
     record = _login_attempts.get(ip)
     if record is None:
         return
-    fail_count, lock_until = record
-    if fail_count >= _MAX_LOGIN_ATTEMPTS:
-        if time.time() < lock_until:
-            raise HTTPException(
-                status_code=429,
-                detail="Too many login attempts. Try again later.",
-            )
+    max_attempts, lockout_seconds = await asyncio.to_thread(_login_throttle_policy)
+    # The await above is a yield point: while this coroutine was suspended,
+    # another request for the same IP may have deleted or replaced the record
+    # (the pre-async version was atomic on the loop). The pre-read served only
+    # as the cheap clean-IP skip; decide on a fresh snapshot from here on —
+    # everything below is synchronous, and every mutation is guarded by
+    # re-comparing against that snapshot so a record replaced mid-flight
+    # (e.g. a successful login followed by a new failure) is never clobbered.
+    record = _login_attempts.get(ip)
+    if record is None:
+        return
+    fail_count, locked_at, locked_duration = record
+    if fail_count < max_attempts:
+        return
+    if locked_at == 0.0:
+        # Over the *current* threshold but the lock never started under the
+        # threshold these failures accumulated under (the operator tightened
+        # max_login_attempts mid-count). Keep the record: the next failure
+        # starts the lock and a successful login clears it — deleting here
+        # would hand the IP a fresh budget under a stricter policy.
+        return
+    now = time.time()
+    if now >= locked_at + locked_duration:
+        # The lock served the full sentence of the duration in force when it
+        # started — a later duration increase must not resurrect it.
+        if _login_attempts.get(ip) == record:
+            del _login_attempts[ip]
+        return
+    if now < locked_at + lockout_seconds:
+        # Still locked. The sentence now follows the current duration, and
+        # that evaluation is committed — including decreases — so the stored
+        # sentence always matches the policy the lock was last evaluated
+        # under; a later raise can never resurrect time the lock already
+        # served under a shorter policy.
+        if lockout_seconds != locked_duration and _login_attempts.get(ip) == record:
+            _login_attempts[ip] = (fail_count, locked_at, lockout_seconds)
+        raise HTTPException(
+            status_code=429,
+            detail="Too many login attempts. Try again later.",
+        )
+    # Original sentence still running, but the current (lowered) duration has
+    # already elapsed — release early.
+    if _login_attempts.get(ip) == record:
         del _login_attempts[ip]
 
 
 _MAX_TRACKED_IPS = 10000
 
 
-def _record_login_failure(ip: str) -> None:
-    """Record a failed login attempt for the given IP."""
-    # Evict expired lockouts when dict grows too large
+def _record_failure_under_policy(ip: str, max_attempts: int, lockout_seconds: float) -> None:
+    """Apply one failed login to the counter under an explicit policy."""
+    # Evict expired lockouts when dict grows too large. Expiry is a property
+    # of each record's own committed sentence — `t > 0 and now >= t + d` —
+    # independent of the live threshold: a record locked under an old, lower
+    # threshold must still be swept once its sentence is served, even if the
+    # current max has moved past its count. Gating on the current threshold
+    # here would retain expired records while the capacity fallback below
+    # evicts live counters (they sort first), granting active offenders
+    # fresh budgets.
     if len(_login_attempts) >= _MAX_TRACKED_IPS:
         now = time.time()
-        expired = [k for k, (c, t) in _login_attempts.items() if c >= _MAX_LOGIN_ATTEMPTS and now >= t]
+        expired = [k for k, (c, t, d) in _login_attempts.items() if t > 0.0 and now >= t + d]
         for k in expired:
             del _login_attempts[k]
-        # If still too large, evict cheapest-to-lose half: below-threshold
-        # IPs (lock_until=0.0) sort first, then earliest-expiring lockouts.
+        # If still too large, evict cheapest-to-lose half ordered by each
+        # record's own expiry: never-locked counters (t + d == 0.0) first,
+        # then locked records whose committed sentence expires earliest.
         if len(_login_attempts) >= _MAX_TRACKED_IPS:
-            by_time = sorted(_login_attempts.items(), key=lambda kv: kv[1][1])
+            by_time = sorted(_login_attempts.items(), key=lambda kv: kv[1][1] + kv[1][2])
             for k, _ in by_time[: len(by_time) // 2]:
                 del _login_attempts[k]
 
     record = _login_attempts.get(ip)
     if record is None:
-        _login_attempts[ip] = (1, 0.0)
+        _login_attempts[ip] = (1, 0.0, 0.0)
     else:
         new_count = record[0] + 1
-        lock_until = time.time() + _LOCKOUT_SECONDS if new_count >= _MAX_LOGIN_ATTEMPTS else 0.0
-        _login_attempts[ip] = (new_count, lock_until)
+        if new_count >= max_attempts:
+            _login_attempts[ip] = (new_count, time.time(), lockout_seconds)
+        else:
+            _login_attempts[ip] = (new_count, 0.0, 0.0)
+
+
+async def _record_login_failure(ip: str) -> None:
+    """Record a failed login attempt for the given IP.
+
+    Policy resolution runs off the event loop (see ``_check_rate_limit``):
+    this is the first config read for a previously clean IP, and the login
+    endpoint is unauthenticated.
+    """
+    try:
+        max_attempts, lockout_seconds = await asyncio.to_thread(_login_throttle_policy)
+    except Exception:
+        # A malformed config keeps failing loudly, but dropping the failure
+        # here would leave the IP clean — and a clean IP skips the config
+        # read in _check_rate_limit, so every subsequent wrong password would
+        # reach authenticate() again: unlimited password verification for as
+        # long as the file stays broken. Count under the model defaults so
+        # the throttle fails closed (the next check reads the broken config
+        # before authenticate), then re-raise.
+        from deerflow.config.auth_config import LocalAuthConfig
+
+        fallback = LocalAuthConfig()
+        _record_failure_under_policy(ip, fallback.max_login_attempts, fallback.lockout_seconds)
+        raise
+    _record_failure_under_policy(ip, max_attempts, lockout_seconds)
 
 
 def _record_login_success(ip: str) -> None:
@@ -295,12 +408,12 @@ async def login_local(
 ):
     """Local email/password login."""
     client_ip = _get_client_ip(request)
-    _check_rate_limit(client_ip)
+    await _check_rate_limit(client_ip)
 
     user = await get_local_provider().authenticate({"email": form_data.username, "password": form_data.password})
 
     if user is None:
-        _record_login_failure(client_ip)
+        await _record_login_failure(client_ip)
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="Incorrect email or password").model_dump(),
diff --git a/backend/docs/AUTH_DESIGN.md b/backend/docs/AUTH_DESIGN.md
index 565365456..fb85c9f0a 100644
--- a/backend/docs/AUTH_DESIGN.md
+++ b/backend/docs/AUTH_DESIGN.md
@@ -110,7 +110,7 @@ enum UserScope:
 - 成功后签发 JWT，放入 `access_token` HttpOnly cookie。
 - 响应体只返回 `expires_in` 和 `needs_setup`，不返回 token。
 
-登录失败会按客户端 IP 计数。IP 解析只在 TCP peer 属于 `AUTH_TRUSTED_PROXIES` 时信任 `X-Real-IP`，不使用 `X-Forwarded-For`。
+登录失败会按客户端 IP 计数。IP 解析只在 TCP peer 属于 `AUTH_TRUSTED_PROXIES` 时信任 `X-Real-IP`，不使用 `X-Forwarded-For`。阈值与锁定时长可通过 `auth.local.max_login_attempts`（默认 5）和 `auth.local.lockout_seconds`（默认 300 秒）配置，按次实时读取，改配置后下一次登录即生效，无需重启 Gateway（`max_login_attempts` 最小为 2：单次失败不得锁定 IP。时长热改按方向生效：下调可提前释放进行中的锁定、收紧阈值会保留已计数的失败；上调只延长仍在锁定期内的锁定，不会复活已服满原时长的锁定）。
 
 ### 注册
 
diff --git a/backend/packages/harness/deerflow/config/auth_config.py b/backend/packages/harness/deerflow/config/auth_config.py
index 09f2b2380..0ff017eb9 100644
--- a/backend/packages/harness/deerflow/config/auth_config.py
+++ b/backend/packages/harness/deerflow/config/auth_config.py
@@ -79,6 +79,26 @@ class LocalAuthConfig(BaseModel):
             "does not apply to local registration."
         ),
     )
+    max_login_attempts: int = Field(
+        default=5,
+        ge=2,
+        description=(
+            "Failed login attempts allowed from one client IP before it is locked out of "
+            "POST /api/v1/auth/login/local. Defaults preserve the historical hardcoded policy. "
+            "Raise it when many users share an egress IP (corporate proxy / NAT); lower it for "
+            "a stricter posture. Minimum 2: one failed attempt must never lock an IP, or a "
+            "single typo would block everyone behind a shared egress — the strictest legal "
+            "value locks after the second failure. The counter is per-Gateway-worker "
+            "(in-process), so effective attempts in multi-worker deployments scale with "
+            "worker count."
+        ),
+    )
+    lockout_seconds: float = Field(
+        default=300.0,
+        gt=0,
+        allow_inf_nan=False,
+        description=("Seconds an IP stays locked out after reaching auth.local.max_login_attempts. Defaults preserve the historical hardcoded policy (5 minutes)."),
+    )
 
 
 class AuthAppConfig(BaseModel):
diff --git a/backend/tests/blocking_io/test_login_local_throttle.py b/backend/tests/blocking_io/test_login_local_throttle.py
new file mode 100644
index 000000000..61dc48dd3
--- /dev/null
+++ b/backend/tests/blocking_io/test_login_local_throttle.py
@@ -0,0 +1,81 @@
+"""Regression anchors: the login throttle must not block the event loop.
+
+``_login_throttle_policy`` resolves the live policy via ``get_app_config()``,
+which stats and re-hashes ``config.yaml`` on every call. ``login_local`` is an
+unauthenticated async endpoint, and every request from a recorded IP resolves
+the policy — including an already-locked attacker flooding the endpoint on
+the way to its 429. Both resolution points (the rate-limit check and the
+failure recorder) offload via ``asyncio.to_thread``; if either regresses onto
+the event loop, the strict Blockbuster gate raises ``BlockingError``.
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+from fastapi import HTTPException
+from fastapi.responses import Response
+from fastapi.security import OAuth2PasswordRequestForm
+from starlette.requests import Request
+
+from app.gateway.routers import auth as auth_router
+
+pytestmark = pytest.mark.asyncio
+
+_CLIENT_IP = "203.0.113.9"
+
+
+@pytest.fixture(autouse=True)
+def _throttle_state(monkeypatch):
+    monkeypatch.delenv("AUTH_TRUSTED_PROXIES", raising=False)
+    auth_router._login_attempts.clear()
+    yield
+    auth_router._login_attempts.clear()
+
+
+def _request() -> Request:
+    return Request(
+        {
+            "type": "http",
+            "method": "POST",
+            "path": "/api/v1/auth/login/local",
+            "headers": [],
+            "query_string": b"",
+            "client": (_CLIENT_IP, 44000),
+            "server": ("testserver", 80),
+        }
+    )
+
+
+def _form() -> OAuth2PasswordRequestForm:
+    return OAuth2PasswordRequestForm(username="user@example.com", password="wrong")
+
+
+async def test_locked_ip_policy_resolution_does_not_block_loop() -> None:
+    """A locked IP floods the endpoint: every request resolves the policy on
+    the way to 429, and that resolution must stay off the event loop."""
+    auth_router._login_attempts[_CLIENT_IP] = (9, time.time(), 3600.0)  # sentence running
+
+    with pytest.raises(HTTPException) as exc_info:
+        await auth_router.login_local(_request(), Response(), _form(), remember_me=True)
+
+    assert exc_info.value.status_code == 429
+
+
+async def test_failed_login_recording_does_not_block_loop(monkeypatch) -> None:
+    """The wrong-password path resolves the policy again inside the recorder;
+    counting must happen without blocking IO on the loop."""
+
+    class _Provider:
+        async def authenticate(self, credentials):
+            return None
+
+    monkeypatch.setattr(auth_router, "get_local_provider", lambda: _Provider())
+    auth_router._login_attempts[_CLIENT_IP] = (1, 0.0, 0.0)  # counting, not locked
+
+    with pytest.raises(HTTPException) as exc_info:
+        await auth_router.login_local(_request(), Response(), _form(), remember_me=True)
+
+    assert exc_info.value.status_code == 401
+    assert auth_router._login_attempts[_CLIENT_IP][0] == 2
diff --git a/backend/tests/test_auth.py b/backend/tests/test_auth.py
index bec069b16..4855f6210 100644
--- a/backend/tests/test_auth.py
+++ b/backend/tests/test_auth.py
@@ -882,37 +882,581 @@ def test_login_response_includes_needs_setup():
 # ── Rate Limiting ──────────────────────────────────────────────────────────
 
 
-def test_rate_limiter_allows_under_limit():
+@pytest.mark.asyncio
+async def test_rate_limiter_allows_under_limit():
     """Requests under the limit are allowed."""
     from app.gateway.routers.auth import _check_rate_limit, _login_attempts
 
     _login_attempts.clear()
-    _check_rate_limit("192.168.1.1")  # Should not raise
+    await _check_rate_limit("192.168.1.1")  # Should not raise
 
 
-def test_rate_limiter_blocks_after_max_failures():
+@pytest.mark.asyncio
+async def test_rate_limiter_blocks_after_max_failures():
     """IP is blocked after 5 consecutive failures."""
     from app.gateway.routers.auth import _check_rate_limit, _login_attempts, _record_login_failure
 
     _login_attempts.clear()
     ip = "10.0.0.1"
     for _ in range(5):
-        _record_login_failure(ip)
+        await _record_login_failure(ip)
     with pytest.raises(HTTPException) as exc_info:
-        _check_rate_limit(ip)
+        await _check_rate_limit(ip)
     assert exc_info.value.status_code == 429
 
 
-def test_rate_limiter_resets_on_success():
+@pytest.mark.asyncio
+async def test_rate_limiter_resets_on_success():
     """Successful login clears the failure counter."""
     from app.gateway.routers.auth import _check_rate_limit, _login_attempts, _record_login_failure, _record_login_success
 
     _login_attempts.clear()
     ip = "10.0.0.2"
     for _ in range(4):
-        _record_login_failure(ip)
+        await _record_login_failure(ip)
     _record_login_success(ip)
-    _check_rate_limit(ip)  # Should not raise
+    await _check_rate_limit(ip)  # Should not raise
+
+
+@pytest.mark.asyncio
+async def test_rate_limiter_honors_configured_attempts_and_lockout(monkeypatch):
+    """auth.local.max_login_attempts / lockout_seconds drive the throttle policy."""
+    from app.gateway.routers import auth as auth_router
+    from app.gateway.routers.auth import _check_rate_limit, _login_attempts, _record_login_failure
+    from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
+    from deerflow.config.auth_config import AuthAppConfig, LocalAuthConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+
+    _login_attempts.clear()
+    set_app_config(
+        AppConfig(
+            sandbox=SandboxConfig(use="test"),
+            auth=AuthAppConfig(local=LocalAuthConfig(max_login_attempts=2, lockout_seconds=60.0)),
+        )
+    )
+    try:
+        ip = "10.0.0.3"
+        await _record_login_failure(ip)
+        await _check_rate_limit(ip)  # 1 failure < 2: allowed
+        await _record_login_failure(ip)
+        with pytest.raises(HTTPException) as exc_info:
+            await _check_rate_limit(ip)
+        assert exc_info.value.status_code == 429
+        # The lockout window comes from lockout_seconds (60s), not the 300s
+        # default: at locked_at + 61 the lock must already be released.
+        _, locked_at, _ = _login_attempts[ip]
+        monkeypatch.setattr(auth_router.time, "time", lambda: locked_at + 61.0)
+        await _check_rate_limit(ip)
+        assert ip not in _login_attempts
+    finally:
+        reset_app_config()
+        _login_attempts.clear()
+
+
+def test_rate_limiter_uses_defaults_when_config_unavailable(monkeypatch):
+    """An absent config.yaml falls back to the built-in (5 attempts / 300s) policy.
+
+    Mirrors ``_local_registration_enabled``: only FileNotFoundError is caught.
+    A malformed config must NOT silently change the throttle policy — pinned
+    to propagate by the test below.
+    """
+    from app.gateway.routers import auth as auth_router
+    from deerflow.config import app_config as app_config_module
+
+    def _missing():
+        raise FileNotFoundError("no config.yaml")
+
+    monkeypatch.setattr(app_config_module, "get_app_config", _missing)
+    assert auth_router._login_throttle_policy() == (5, 300)
+
+
+def test_rate_limiter_malformed_config_propagates(monkeypatch):
+    """A malformed config.yaml fails loudly instead of fail-opening the throttle.
+
+    Every other config consumer 500s on a validation failure; silently
+    substituting the (possibly more permissive) defaults here would diverge —
+    an operator who set max_login_attempts=2 must never silently get 5.
+    """
+    from app.gateway.routers import auth as auth_router
+    from deerflow.config import app_config as app_config_module
+
+    def _malformed():
+        raise ValueError("config validation error")
+
+    monkeypatch.setattr(app_config_module, "get_app_config", _malformed)
+    with pytest.raises(ValueError, match="config validation error"):
+        auth_router._login_throttle_policy()
+
+
+@pytest.mark.asyncio
+async def test_rate_limiter_clean_ip_skips_config_read(monkeypatch):
+    """A clean IP pays zero config reads: the record-None early return must
+    come before policy resolution (get_app_config re-hashes config.yaml on
+    every call, and login_local is an unauthenticated async endpoint)."""
+    from app.gateway.routers import auth as auth_router
+    from deerflow.config import app_config as app_config_module
+
+    def _must_not_load():
+        raise AssertionError("config must not be read for a clean IP")
+
+    monkeypatch.setattr(app_config_module, "get_app_config", _must_not_load)
+    auth_router._login_attempts.clear()
+    await auth_router._check_rate_limit("192.0.2.7")  # returns quietly → no config read
+
+
+@pytest.mark.asyncio
+async def test_rate_limiter_policy_change_semantics():
+    """Pin the emergent semantics of a live-read policy under config reload.
+
+    Raising max_login_attempts mid-lockout immediately unblocks IPs whose
+    fail_count falls below the new threshold — that is the issue #5108 use
+    case (shared-egress-IP office unblocked by raising the limit, no restart).
+    Tightening the threshold keeps the accumulated count (see the dedicated
+    test below); subsequent failures lock under the new, stricter policy.
+    """
+    from app.gateway.routers.auth import _check_rate_limit, _login_attempts, _record_login_failure
+    from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
+    from deerflow.config.auth_config import AuthAppConfig, LocalAuthConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+
+    def _set_policy(max_attempts: int) -> None:
+        set_app_config(
+            AppConfig(
+                sandbox=SandboxConfig(use="test"),
+                auth=AuthAppConfig(local=LocalAuthConfig(max_login_attempts=max_attempts, lockout_seconds=60.0)),
+            )
+        )
+
+    _login_attempts.clear()
+    try:
+        ip = "10.0.0.4"
+        _set_policy(2)
+        for _ in range(2):
+            await _record_login_failure(ip)
+        with pytest.raises(HTTPException):
+            await _check_rate_limit(ip)  # locked under the old policy
+
+        _set_policy(5)  # operator raises the ceiling mid-lockout
+        await _check_rate_limit(ip)  # immediately allowed: 2 < 5, no restart needed
+
+        # max_login_attempts=1 never reaches the endpoint: config load rejects
+        # it (ge=2). A legal value of 1 previously disabled lockout entirely —
+        # the (1, 0.0) first-failure record expired immediately, and every
+        # subsequent failure re-created it, so the IP was never locked.
+        import pydantic
+
+        with pytest.raises(pydantic.ValidationError):
+            _set_policy(1)
+
+        # The strictest legal value still locks, at the second failure.
+        _login_attempts.pop(ip, None)
+        _set_policy(2)
+        await _record_login_failure(ip)
+        await _check_rate_limit(ip)  # 1 failure < 2: allowed
+        await _record_login_failure(ip)
+        with pytest.raises(HTTPException):
+            await _check_rate_limit(ip)
+    finally:
+        reset_app_config()
+        _login_attempts.clear()
+
+
+def test_local_auth_throttle_config_validation():
+    """Throttle knobs reject degenerate operator values at config load."""
+    import pydantic
+
+    from deerflow.config.auth_config import LocalAuthConfig
+
+    with pytest.raises(pydantic.ValidationError):
+        LocalAuthConfig(max_login_attempts=0)
+    with pytest.raises(pydantic.ValidationError):
+        # 1 would mean a single typo locks the whole shared egress, and before
+        # ge=2 it actually disabled lockout entirely (the (1, 0.0) record
+        # expired immediately and every failure re-created it).
+        LocalAuthConfig(max_login_attempts=1)
+    LocalAuthConfig(max_login_attempts=2)  # strictest legal value
+    with pytest.raises(pydantic.ValidationError):
+        LocalAuthConfig(lockout_seconds=0)
+    with pytest.raises(pydantic.ValidationError):
+        LocalAuthConfig(lockout_seconds=float("inf"))
+
+
+@pytest.mark.asyncio
+async def test_rate_limiter_active_lockout_honors_live_lockout_seconds_change(monkeypatch):
+    """A live lockout_seconds change applies to in-flight lockouts, in the
+    direction that serves the operator, without resurrecting served sentences.
+
+    A lock records the duration in force when it started. At check time an
+    active sentence follows the *currently* configured duration — lowering
+    60 → 1 releases early — while a lock that already served its original
+    sentence stays expired even if the duration is later raised (no
+    resurrection), and a raise while the lock is still active extends it.
+    """
+    from app.gateway.routers import auth as auth_router
+    from app.gateway.routers.auth import _check_rate_limit, _login_attempts, _record_login_failure
+    from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
+    from deerflow.config.auth_config import AuthAppConfig, LocalAuthConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+
+    def _set_policy(lockout_seconds: float) -> None:
+        set_app_config(
+            AppConfig(
+                sandbox=SandboxConfig(use="test"),
+                auth=AuthAppConfig(local=LocalAuthConfig(max_login_attempts=2, lockout_seconds=lockout_seconds)),
+            )
+        )
+
+    def _freeze_clock_at(t: float) -> None:
+        monkeypatch.setattr(auth_router.time, "time", lambda: t)
+
+    _login_attempts.clear()
+    try:
+        ip = "10.0.0.5"
+
+        # Lowering mid-lockout releases early.
+        _set_policy(60.0)
+        await _record_login_failure(ip)
+        await _record_login_failure(ip)
+        _, locked_at, _ = _login_attempts[ip]
+        _freeze_clock_at(locked_at + 2.0)
+        with pytest.raises(HTTPException):
+            await _check_rate_limit(ip)  # 2s into the 60s window: still locked
+        _set_policy(1.0)  # operator shortens the window
+        await _check_rate_limit(ip)  # 2s > 1s: unlocked on the very next login
+        assert ip not in _login_attempts
+
+        # Raising while the lock is still active extends it.
+        _set_policy(1.0)
+        await _record_login_failure(ip)
+        await _record_login_failure(ip)
+        _, locked_at, _ = _login_attempts[ip]
+        _freeze_clock_at(locked_at + 0.5)  # still inside the 1s sentence
+        _set_policy(60.0)  # operator lengthens the window mid-sentence
+        with pytest.raises(HTTPException):
+            await _check_rate_limit(ip)  # active, and 0.5 < 60: extended
+        _freeze_clock_at(locked_at + 2.0)  # past the original 1s sentence
+        with pytest.raises(HTTPException):
+            await _check_rate_limit(ip)  # extended: 2 < 60, still locked
+
+        # A sentence that already elapsed before the raise is not resurrected.
+        _set_policy(1.0)
+        await _record_login_failure(ip)
+        await _record_login_failure(ip)
+        _, locked_at, _ = _login_attempts[ip]
+        _freeze_clock_at(locked_at + 2.0)  # past the 1s sentence, no check yet
+        _set_policy(60.0)  # operator lengthens the window for *future* locks
+        await _check_rate_limit(ip)  # the served 1s sentence is not resurrected
+        assert ip not in _login_attempts
+    finally:
+        reset_app_config()
+        _login_attempts.clear()
+
+
+@pytest.mark.asyncio
+async def test_rate_limiter_lowered_then_raised_duration_not_resurrected(monkeypatch):
+    """A shortened duration observed during the sentence is committed, so a
+    later raise cannot resurrect time already served under the short policy.
+
+    60s lock, evaluated at +6s under a lowered 10s policy (still locked — the
+    sentence becomes 10s), then raised to 30s at +20s: the lock expired at
+    +10s under the last-evaluated policy, so the +20s request must be allowed.
+    """
+    from app.gateway.routers import auth as auth_router
+    from app.gateway.routers.auth import _check_rate_limit, _login_attempts, _record_login_failure
+    from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
+    from deerflow.config.auth_config import AuthAppConfig, LocalAuthConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+
+    def _set_policy(lockout_seconds: float) -> None:
+        set_app_config(
+            AppConfig(
+                sandbox=SandboxConfig(use="test"),
+                auth=AuthAppConfig(local=LocalAuthConfig(max_login_attempts=2, lockout_seconds=lockout_seconds)),
+            )
+        )
+
+    _login_attempts.clear()
+    try:
+        ip = "10.0.0.8"
+        _set_policy(60.0)
+        await _record_login_failure(ip)
+        await _record_login_failure(ip)
+        _, locked_at, _ = _login_attempts[ip]
+
+        def _freeze(t: float) -> None:
+            monkeypatch.setattr(auth_router.time, "time", lambda: t)
+
+        _freeze(locked_at + 6.0)
+        _set_policy(10.0)
+        with pytest.raises(HTTPException):
+            await _check_rate_limit(ip)  # 6 < 10: still locked, and the 10s sentence is committed
+        assert _login_attempts[ip][2] == 10.0
+
+        _freeze(locked_at + 20.0)
+        _set_policy(30.0)  # raised after the 10s sentence was served at +10s
+        await _check_rate_limit(ip)  # not resurrected: allowed
+        assert ip not in _login_attempts
+    finally:
+        reset_app_config()
+        _login_attempts.clear()
+
+
+@pytest.mark.asyncio
+async def test_concurrent_checks_on_expired_lock_are_race_free(monkeypatch):
+    """Synchronized checks of the same expired lock must all resolve cleanly.
+
+    Policy resolution yields the event loop (asyncio.to_thread), so the sync
+    version's atomicity is gone: with a pre-await snapshot only, concurrent
+    checks of an expired record raced into double ``del`` — one request
+    returned normally and the others raised KeyError (review reproduction).
+    """
+    import asyncio
+
+    from app.gateway.routers import auth as auth_router
+    from app.gateway.routers.auth import _check_rate_limit, _login_attempts
+    from deerflow.config.auth_config import LocalAuthConfig
+
+    def _defaults():
+        return LocalAuthConfig().max_login_attempts, LocalAuthConfig().lockout_seconds
+
+    monkeypatch.setattr(auth_router, "_login_throttle_policy", _defaults)
+    _login_attempts.clear()
+    try:
+        _login_attempts["10.0.0.9"] = (5, 1.0, 1.0)  # sentence long expired
+
+        results = await asyncio.gather(*[_check_rate_limit("10.0.0.9") for _ in range(8)], return_exceptions=True)
+
+        assert all(result is None for result in results), results
+        assert "10.0.0.9" not in _login_attempts
+    finally:
+        _login_attempts.clear()
+
+
+@pytest.mark.asyncio
+async def test_check_survives_record_deleted_during_policy_resolution(monkeypatch):
+    """A record removed while the checker is suspended must not KeyError.
+
+    Deterministic stand-in for the suspension-window interleaving: the policy
+    resolution itself removes the record, exactly like a concurrent success
+    login would while this coroutine sits in ``asyncio.to_thread``.
+    """
+    from app.gateway.routers import auth as auth_router
+    from app.gateway.routers.auth import _check_rate_limit, _login_attempts, _record_login_success
+
+    def _policy_that_deletes_the_record():
+        _record_login_success("10.0.0.10")
+        return 5, 300.0
+
+    monkeypatch.setattr(auth_router, "_login_throttle_policy", _policy_that_deletes_the_record)
+    _login_attempts.clear()
+    try:
+        _login_attempts["10.0.0.10"] = (5, 1.0, 1.0)  # expired
+
+        await _check_rate_limit("10.0.0.10")  # pre-fix: KeyError
+
+        assert "10.0.0.10" not in _login_attempts
+    finally:
+        _login_attempts.clear()
+
+
+@pytest.mark.asyncio
+async def test_check_never_clobbers_record_replaced_during_policy_resolution(monkeypatch):
+    """A record replaced while the checker is suspended must survive intact.
+
+    The pre-await snapshot said "locked"; while suspended, a successful login
+    plus one new failure replaced the record with a fresh counter. The checker
+    must re-read and leave the fresh record alone instead of writing its
+    stale-snapshot decision over it.
+    """
+    from app.gateway.routers import auth as auth_router
+    from app.gateway.routers.auth import _check_rate_limit, _login_attempts
+
+    def _policy_that_replaces_the_record():
+        _login_attempts["10.0.0.11"] = (1, 0.0, 0.0)
+        return 2, 60.0
+
+    monkeypatch.setattr(auth_router, "_login_throttle_policy", _policy_that_replaces_the_record)
+    _login_attempts.clear()
+    try:
+        _login_attempts["10.0.0.11"] = (2, 1.0, 1.0)  # locked, sentence running
+
+        await _check_rate_limit("10.0.0.11")  # fresh read: (1, 0, 0) < max 2 → allowed
+
+        assert _login_attempts["10.0.0.11"] == (1, 0.0, 0.0)
+    finally:
+        _login_attempts.clear()
+
+
+@pytest.mark.asyncio
+async def test_rate_limiter_eviction_expires_by_stored_sentence_not_current_threshold(monkeypatch):
+    """The capacity sweep must expire records by their own committed sentence,
+    with no gate on the current threshold.
+
+    A record locked under an old, lower threshold has a count below the live
+    max after the operator raises it; gating expiry on ``count >= max`` keeps
+    that served record resident while the capacity fallback evicts live
+    counters first (they sort earliest), handing an active offender a fresh
+    budget. Reproduction from review: cap 2, live ``(1, 0, 0)`` plus expired
+    ``(2, 10, 1)`` under max=3, clock at 100.
+    """
+    from app.gateway.routers import auth as auth_router
+    from app.gateway.routers.auth import _login_attempts, _record_login_failure
+    from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
+    from deerflow.config.auth_config import AuthAppConfig, LocalAuthConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+
+    monkeypatch.setattr(auth_router, "_MAX_TRACKED_IPS", 2)
+    monkeypatch.setattr(auth_router.time, "time", lambda: 100.0)
+    set_app_config(
+        AppConfig(
+            sandbox=SandboxConfig(use="test"),
+            auth=AuthAppConfig(local=LocalAuthConfig(max_login_attempts=3, lockout_seconds=60.0)),
+        )
+    )
+    _login_attempts.clear()
+    try:
+        _login_attempts["live-counter"] = (1, 0.0, 0.0)  # active offender, counting
+        _login_attempts["expired-lock"] = (2, 10.0, 1.0)  # locked under old max=2; served at 11.0
+
+        await _record_login_failure("fresh-ip")  # hits the capacity sweep
+
+        assert "expired-lock" not in _login_attempts  # served sentence is swept
+        assert _login_attempts["live-counter"] == (1, 0.0, 0.0)  # live counter survives
+        assert _login_attempts["fresh-ip"][0] == 1
+    finally:
+        reset_app_config()
+        _login_attempts.clear()
+
+
+@pytest.mark.asyncio
+async def test_rate_limiter_tightened_threshold_preserves_failures():
+    """Tightening max_login_attempts mid-count keeps the accumulated failures.
+
+    An IP with four failures under max_login_attempts=5 must not get a fresh
+    budget when the operator lowers the threshold to 2: the count stays, the
+    next failure starts the lock, and a successful login still clears it.
+    """
+    from app.gateway.routers.auth import _check_rate_limit, _login_attempts, _record_login_failure, _record_login_success
+    from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
+    from deerflow.config.auth_config import AuthAppConfig, LocalAuthConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+
+    def _set_policy(max_attempts: int) -> None:
+        set_app_config(
+            AppConfig(
+                sandbox=SandboxConfig(use="test"),
+                auth=AuthAppConfig(local=LocalAuthConfig(max_login_attempts=max_attempts, lockout_seconds=60.0)),
+            )
+        )
+
+    _login_attempts.clear()
+    try:
+        ip = "10.0.0.6"
+        _set_policy(5)
+        for _ in range(4):
+            await _record_login_failure(ip)  # 4 failures: counting, never locked
+
+        _set_policy(2)  # operator tightens the policy mid-count
+        await _check_rate_limit(ip)  # allowed this once — but the count survives
+        assert _login_attempts[ip][0] == 4
+
+        await _record_login_failure(ip)  # 4 + 1 >= 2: locks on the very next failure
+        with pytest.raises(HTTPException):
+            await _check_rate_limit(ip)
+
+        # A correct password still clears everything (no retroactive lockout
+        # of a legitimate user who fat-fingered the password four times).
+        _record_login_success(ip)
+        await _check_rate_limit(ip)
+        assert ip not in _login_attempts
+    finally:
+        reset_app_config()
+        _login_attempts.clear()
+
+
+@pytest.mark.asyncio
+async def test_rate_limiter_counts_failure_when_config_breaks(monkeypatch):
+    """A malformed config hot-edit must not hand out unlimited verification.
+
+    Endpoint order per failed login: _check_rate_limit (clean IPs skip the
+    config read) → authenticate() → _record_login_failure. If the record call
+    raises on the broken config *before* mutating state, the IP stays clean
+    and every subsequent wrong password reaches authenticate() again. The
+    failure must land (counted under the model defaults) before the error
+    re-raises; from then on the dirty IP's own check reads the broken config
+    and fails closed — before authenticate.
+    """
+    from app.gateway.routers.auth import _check_rate_limit, _login_attempts, _record_login_failure
+    from deerflow.config import app_config as app_config_module
+
+    def _malformed():
+        raise ValueError("config validation error")
+
+    _login_attempts.clear()
+    monkeypatch.setattr(app_config_module, "get_app_config", _malformed)
+    try:
+        ip = "10.0.0.7"
+
+        # First failed login: still allowed through to authenticate(), then
+        # the record call fails loudly — but the failure is counted first.
+        await _check_rate_limit(ip)  # clean IP: no config read, allowed
+        with pytest.raises(ValueError, match="config validation error"):
+            await _record_login_failure(ip)
+        assert _login_attempts[ip] == (1, 0.0, 0.0)
+
+        # Second login: the now-dirty IP's check reads the broken config and
+        # fails closed *before* authenticate() — no more password verification.
+        with pytest.raises(ValueError, match="config validation error"):
+            await _check_rate_limit(ip)
+    finally:
+        _login_attempts.clear()
+
+
+def test_login_local_broken_config_fails_closed_after_first_failure(monkeypatch):
+    """Route-level pin of the same sequence, through POST /login/local.
+
+    The first wrong password is verified once and counted (the 500 comes from
+    the re-raised config error, not from a skipped record); the second request
+    from the same client IP must 500 in _check_rate_limit *before*
+    authenticate() is reached — no unlimited password verification while
+    config.yaml stays malformed.
+    """
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
+
+    from app.gateway.routers import auth as auth_router
+    from deerflow.config import app_config as app_config_module
+
+    def _malformed():
+        raise ValueError("config validation error")
+
+    monkeypatch.setattr(app_config_module, "get_app_config", _malformed)
+
+    calls = {"authenticate": 0}
+
+    class _Provider:
+        async def authenticate(self, credentials):
+            calls["authenticate"] += 1
+            return None  # wrong password
+
+    monkeypatch.setattr(auth_router, "get_local_provider", lambda: _Provider())
+    monkeypatch.delenv("AUTH_TRUSTED_PROXIES", raising=False)
+    auth_router._login_attempts.clear()
+
+    app = FastAPI()
+    app.include_router(auth_router.router)
+    try:
+        with TestClient(app, raise_server_exceptions=False) as client:
+            first = client.post("/api/v1/auth/login/local", data={"username": "user@example.com", "password": "wrong"})
+            assert first.status_code == 500
+            assert calls["authenticate"] == 1  # verified once — and counted despite the raise
+            assert auth_router._login_attempts["testclient"][0] == 1
+
+            second = client.post("/api/v1/auth/login/local", data={"username": "user@example.com", "password": "wrong-again"})
+            assert second.status_code == 500
+            assert calls["authenticate"] == 1  # fail-closed: no second verification
+    finally:
+        auth_router._login_attempts.clear()
 
 
 # ── Client IP extraction ─────────────────────────────────────────────────
diff --git a/config.example.yaml b/config.example.yaml
index 6a260c23f..e2c098750 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -15,7 +15,7 @@
 # ============================================================================
 # Bump this number when the config schema changes.
 # Run `make config-upgrade` to merge new fields into your local config.yaml.
-config_version: 38
+config_version: 39
 
 # ============================================================================
 # Logging
@@ -2722,6 +2722,19 @@ authorization:
 # auth:
 #   local:
 #     allow_registration: false
+#     # Login throttling per client IP (in-process, per worker). Raise
+#     # max_login_attempts when many users share one egress IP (corporate
+#     # proxy / NAT) so shared-IP lockouts don't block the whole office;
+#     # lower it for a stricter posture (minimum 2 — one failed attempt must
+#     # never lock an IP, or a single typo would block the shared egress).
+#     # Defaults preserve the historical hardcoded policy (5 failures /
+#     # 5 minutes). Live-read per login, so a config reload applies without
+#     # a Gateway restart: lowering lockout_seconds releases an active lock
+#     # early and tightening max_login_attempts keeps already-counted
+#     # failures; a raised lockout_seconds extends a still-active lock but
+#     # never resurrects one that already expired.
+#     # max_login_attempts: 5
+#     # lockout_seconds: 300
 #
 #   oidc:
 #     enabled: true
diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md
index db4bc827f..f85ee9c44 100644
--- a/deploy/helm/deer-flow/README.md
+++ b/deploy/helm/deer-flow/README.md
@@ -131,7 +131,7 @@ they resolve from the `secrets` map):
 
 ```yaml
 config: |
-  config_version: 38
+  config_version: 39
   models:
     - name: gpt-4
       use: langchain_openai:ChatOpenAI
diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml
index 9732acf3b..446781182 100644
--- a/deploy/helm/deer-flow/values.yaml
+++ b/deploy/helm/deer-flow/values.yaml
@@ -249,7 +249,7 @@ ingress:
 # -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
 #    inline literal secret values here. The default enables provisioner sandbox.
 config: |
-  config_version: 38
+  config_version: 39
   log_level: info
 
   models: []