feat(models): pace shared RPM budgets before dispatch (#5432)

* feat(models): add shared RPM admission queues

* fix(models): address admission pacing review feedback

* docs: simplify request admission quick start guidance

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -110,6 +110,10 @@ That prompt is intended for coding agents. It tells the agent to clone the repo
 
 ### Configuration
 
+Optional per-model [`request_admission`](backend/docs/CONFIGURATION.md#model-request-admission)
+paces requests to help stay within provider request-per-minute limits.
+It is disabled by default; see the linked guide to enable it.
+
 1. **Clone the DeerFlow repository**
 
    ```bash

backend/docs/CONFIGURATION.md

@@ -2,6 +2,51 @@
 
 This guide explains how to configure DeerFlow for your environment.
 
+## Model request admission
+
+For request-per-minute limits, opt into pacing on each relevant `models[]`
+entry. For example, add this alongside its `name`, `use`, and `model` fields:
+
+```yaml
+request_admission:
+  requests_per_minute: 60
+  group: shared-provider-account
+  max_wait_seconds: 300
+  max_queue_size: 256
+```
+
+Calls wait in a bounded FIFO before dispatch. At 60 RPM, admissions are spaced
+at least one second apart, even after idle periods. The first call can proceed
+immediately. Async waiting is cancellable; a cancelled or expired waiter spends
+no admission. Queue overflow and wait expiry fail locally before dispatch.
+The deadline covers admission waiting only, not the provider's response time.
+
+An explicit `group` shares the budget across model profiles using the same
+provider quota. Omit it for a separate budget per configured model name.
+All profiles in a group must have identical settings. The limiter is shared by
+factory-created model instances across threads and event loops, including
+lead agents, subagents and auxiliary models using the standard LangChain
+BaseChatModel invoke/stream hooks. Restart the Gateway after changing,
+disabling or regrouping active policies; conflicting settings fail model
+construction instead of resetting a live budget.
+
+When enabled, exposed SDK `max_retries` settings are set to zero: SDK retries
+would bypass the admission hook. Agent middleware retries still work and each
+new attempt is paced. Calls outside that middleware no longer get SDK retries.
+Custom providers that bypass BaseChatModel admission hooks or perform hidden
+retries need their own integration. A caller-supplied `rate_limiter` cannot be
+combined with `request_admission`.
+
+This is **process-local RPM pacing**, not TPM accounting or a distributed quota
+service. Divide the provider allowance among Gateway workers/replicas and allow
+headroom for other applications. Provider token limits, external consumption,
+billing failures and permanent errors can still fail a task. Existing
+`llm_call.max_concurrent_calls` remains independent: when enabled, its slot is
+held while the underlying model waits for admission, so use shared groups and
+concurrency settings deliberately. Waiting contributes to model-call latency
+and remains subject to the enclosing run's timeout. This option is off by
+default and does not promise unlimited retries or eventual task completion.
+
 ## Config Versioning
 
 `config.example.yaml` contains a `config_version` field that tracks schema changes. When the example version is higher than your local `config.yaml`, the application emits a startup warning:

backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py

@@ -23,6 +23,7 @@ from langchain_core.messages import AIMessage
 from langgraph.errors import GraphBubbleUp
 
 from deerflow.config.app_config import AppConfig
+from deerflow.models.request_admission import AdmissionError
 from deerflow.utils.custom_events import aemit_custom_event, emit_custom_event
 
 logger = logging.getLogger(__name__)
@@ -504,6 +505,8 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
                 self._circuit_probe_token = None
 
     def _classify_error(self, exc: BaseException) -> tuple[bool, str]:
+        if isinstance(exc, AdmissionError):
+            return False, "admission"
         detail = _extract_error_detail(exc)
         lowered = detail.lower()
         error_code = _extract_error_code(exc)

backend/packages/harness/deerflow/config/AGENTS.md

@@ -47,6 +47,14 @@ Configuration priority:
 Config values starting with `$` are resolved as environment variables (e.g., `$OPENAI_API_KEY`).
 `ModelConfig` also declares `use_responses_api` and `output_version` so OpenAI `/v1/responses` can be enabled explicitly while still using `langchain_openai:ChatOpenAI`.
 
+`ModelConfig.request_admission` is optional and is not a provider parameter.
+Its positive RPM, finite wait deadline, queue bound and optional quota-group name
+configure process-local model pacing. Models sharing an explicit group must use
+identical policies. Restart after changing, disabling or regrouping an active
+policy; conflicting policies fail construction rather than silently resetting
+an active budget. This nested model option is enforced by its limiter registry,
+not by the top-level infrastructure reload-boundary registry.
+
 **Extensions Configuration** (`extensions_config.json`):
 
 MCP servers and skills are configured together in `extensions_config.json` in project root:

backend/packages/harness/deerflow/config/model_config.py

@@ -1,10 +1,21 @@
 from pydantic import BaseModel, ConfigDict, Field
 
 
+class RequestAdmissionConfig(BaseModel):
+    """Optional per-process pacing of model requests, shared by quota group."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+    requests_per_minute: int = Field(gt=0, strict=True)
+    group: str | None = Field(default=None, min_length=1, max_length=100, pattern=r"^[A-Za-z0-9_.-]+$")
+    max_wait_seconds: float = Field(default=300, gt=0, allow_inf_nan=False)
+    max_queue_size: int = Field(default=256, gt=0, strict=True)
+
+
 class ModelConfig(BaseModel):
     """Config section for a model"""
 
     name: str = Field(..., description="Unique name for the model")
+    request_admission: RequestAdmissionConfig | None = Field(default=None, description="Opt-in process-local RPM pacing. Changing an active group's policy requires a process restart.")
     display_name: str | None = Field(..., default_factory=lambda: None, description="Display name for the model")
     description: str | None = Field(..., default_factory=lambda: None, description="Description for the model")
     use: str = Field(

backend/packages/harness/deerflow/models/AGENTS.md

@@ -1,5 +1,10 @@
 ### Model Factory (`packages/harness/deerflow/models/factory.py`)
 
+Request-admission waits follow the next scheduled admission and configured
+interval, capped at 50 ms; the cap must not become a minimum poll interval that
+limits high-RPM throughput. Local `AdmissionError` is structurally non-retriable
+in LLM error handling regardless of its message text.
+
 - `create_chat_model(name, thinking_enabled)` instantiates LLM from config via reflection
 - Supports `thinking_enabled` flag with per-model `when_thinking_enabled` overrides
 - Supports vLLM-style thinking toggles via `when_thinking_enabled.extra_body.chat_template_kwargs.enable_thinking` for Qwen reasoning models, while normalizing legacy `thinking` configs for backward compatibility
@@ -7,6 +12,7 @@
 - Supports `supports_vision` flag for image understanding models
 - Config values starting with `$` resolved as environment variables
 - Missing provider modules surface actionable install hints from reflection resolvers (for example `uv add langchain-google-genai`)
+- Optional `models[].request_admission` attaches a process-shared `BaseRateLimiter` at the model factory. Identical explicit groups share one FIFO across model instances, threads and event loops; implicit groups use the configured model name. Policies are immutable once registered and conflicting settings fail construction. A monotonic minimum interval spaces requests without idle-time burst credit; bounded waiters poll without occupying executor threads and unregister in `finally`. The factory strips the policy from provider kwargs and sets exposed SDK `max_retries=0` so middleware retries re-enter admission. This limits model invocations, not tokens or a distributed provider account; custom providers bypassing BaseChatModel hooks are outside the contract. Tests: `test_model_request_admission.py`.
 
 ### Claude Code Credentials (`packages/harness/deerflow/models/credential_loader.py`)
 

backend/packages/harness/deerflow/models/factory.py

@@ -226,6 +226,7 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, *
             # display) — must never reach the provider client, which would
             # forward unknown kwargs into the completion request payload.
             "pricing",
+            "request_admission",
         },
     )
     # Layer per-caller sampling overrides (e.g. a custom agent's temperature /
@@ -325,6 +326,18 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, *
     # explicit profile from a caller or model_overrides is never clobbered.
     translate_context_window = bool(model_config.context_window) and "profile" not in kwargs and "profile" not in model_settings_from_config
 
+    if model_config.request_admission is not None:
+        from deerflow.models.request_admission import get_request_admission
+
+        if "rate_limiter" in kwargs or "rate_limiter" in model_settings_from_config:
+            raise ValueError("request_admission cannot be combined with a custom rate_limiter")
+        model_settings_from_config["rate_limiter"] = get_request_admission(name, model_config.request_admission)
+        # SDK-internal retries do not re-enter BaseChatModel's admission hook.
+        # Keep retries at the middleware layer where each attempt is paced.
+        if "max_retries" in model_class.model_fields:
+            kwargs.pop("max_retries", None)
+            model_settings_from_config["max_retries"] = 0
+
     _warn_unknown_model_settings(model_class, name, model_settings_from_config)
 
     model_instance = model_class(**kwargs, **model_settings_from_config)

backend/packages/harness/deerflow/models/request_admission.py

@@ -0,0 +1,113 @@
+"""Bounded FIFO admission shared by synchronous calls and independent loops.
+
+No executor worker or timer task is retained while waiting. Short polling sleeps
+allow cancellation and deadlines without storing references to caller loops.
+Requests are evenly spaced; idle time never accumulates a burst allowance.
+"""
+
+import asyncio
+import threading
+import time
+from collections import deque
+from time import monotonic
+
+from langchain_core.rate_limiters import BaseRateLimiter
+
+from deerflow.config.model_config import RequestAdmissionConfig
+
+
+class AdmissionError(RuntimeError):
+    """Local admission failed before making an upstream request."""
+
+
+class RequestAdmission(BaseRateLimiter):
+    def __init__(self, config: RequestAdmissionConfig):
+        self.config = config
+        self._interval = 60 / config.requests_per_minute
+        self._next = 0.0
+        self._lock = threading.Lock()
+        self._waiters: deque[object] = deque()
+
+    def _try(self, ticket: object | None) -> bool:
+        with self._lock:
+            now = monotonic()
+            if self._waiters and self._waiters[0] is not ticket:
+                return False
+            if now < self._next:
+                return False
+            self._next = now + self._interval
+            return True
+
+    def _enqueue(self) -> object:
+        ticket = object()
+        with self._lock:
+            if len(self._waiters) >= self.config.max_queue_size:
+                raise AdmissionError("LLM admission queue is full; reduce workload or increase queue capacity.")
+            self._waiters.append(ticket)
+        return ticket
+
+    def _remove(self, ticket: object) -> None:
+        with self._lock:
+            self._waiters.remove(ticket)
+
+    def _delay(self, deadline: float) -> float:
+        now = monotonic()
+        remaining = deadline - now
+        if remaining <= 0:
+            raise AdmissionError("LLM admission timed out before dispatch; increase max_wait_seconds or reduce workload.")
+        with self._lock:
+            until_next = self._next - now
+        # Track short admission intervals rather than imposing a 20/s ceiling.
+        # Non-head waiters still yield when the schedule is already due.
+        return min(0.05, self._interval, until_next if until_next > 0 else self._interval, remaining)
+
+    def acquire(self, *, blocking: bool = True) -> bool:
+        if self._try(None):
+            return True
+        if not blocking:
+            return False
+        deadline = monotonic() + self.config.max_wait_seconds
+        ticket = self._enqueue()
+        try:
+            while True:
+                delay = self._delay(deadline)
+                if self._try(ticket):
+                    return True
+                time.sleep(delay)
+        finally:
+            self._remove(ticket)
+
+    async def aacquire(self, *, blocking: bool = True) -> bool:
+        if self._try(None):
+            return True
+        if not blocking:
+            return False
+        deadline = monotonic() + self.config.max_wait_seconds
+        ticket = self._enqueue()
+        try:
+            while True:
+                delay = self._delay(deadline)
+                if self._try(ticket):
+                    return True
+                await asyncio.sleep(delay)
+        finally:
+            self._remove(ticket)
+
+
+_registry_lock = threading.Lock()
+_registry: dict[tuple[str, str], RequestAdmission] = {}
+
+
+def get_request_admission(model_name: str, config: RequestAdmissionConfig) -> RequestAdmission:
+    # Explicit group names never collide with implicit model names. Operator
+    # configuration determines the finite set; no user IDs or secrets are keys.
+    key = ("group", config.group) if config.group else ("model", model_name)
+    with _registry_lock:
+        existing = _registry.get(key)
+        if existing is not None:
+            if existing.config != config:
+                raise ValueError("Model request admission policy changed or conflicts within a shared group; align settings and restart the Gateway.")
+            return existing
+        limiter = RequestAdmission(config)
+        _registry[key] = limiter
+        return limiter

backend/tests/test_model_request_admission.py

@@ -0,0 +1,228 @@
+"""Offline request pacing, queue lifecycle, and model-factory integration."""
+
+import asyncio
+import threading
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+from pydantic import ValidationError
+
+from deerflow.config.model_config import RequestAdmissionConfig
+from deerflow.models import request_admission as admission
+
+
+@pytest.fixture
+def clock(monkeypatch):
+    now = [0.0]
+    monkeypatch.setattr(admission, "monotonic", lambda: now[0])
+    return now
+
+
+def test_pacing_has_no_catch_up_burst(clock):
+    limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=60))
+    assert limiter.acquire(blocking=False)
+    assert not limiter.acquire(blocking=False)
+    clock[0] = 0.999
+    assert not limiter.acquire(blocking=False)
+    clock[0] = 1
+    assert limiter.acquire(blocking=False)
+    clock[0] = 100
+    assert limiter.acquire(blocking=False)
+    assert not limiter.acquire(blocking=False)
+
+
+def test_high_rpm_wait_tracks_next_admission(clock):
+    limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=6000))
+    limiter.acquire()
+    assert limiter._delay(300) == pytest.approx(0.01)
+    clock[0] = 0.009
+    assert limiter._delay(300) == pytest.approx(0.001)
+    clock[0] = 0.02
+    # A non-head waiter must yield, rather than spin on an overdue schedule.
+    assert 0 < limiter._delay(300) <= 0.01
+
+
+@pytest.mark.asyncio
+async def test_fifo_cancellation_and_queue_capacity(clock):
+    limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=60, max_queue_size=2))
+    await limiter.aacquire()
+    first = asyncio.create_task(limiter.aacquire())
+    second = asyncio.create_task(limiter.aacquire())
+    await asyncio.sleep(0)
+    try:
+        with pytest.raises(admission.AdmissionError, match="queue is full"):
+            await limiter.aacquire()
+        assert not limiter.acquire(blocking=False)
+        first.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await first
+        clock[0] = 1
+        assert await asyncio.wait_for(second, 1)
+        assert not limiter._waiters
+    finally:
+        for task in (first, second):
+            task.cancel()
+        await asyncio.gather(first, second, return_exceptions=True)
+
+
+@pytest.mark.asyncio
+async def test_wait_deadline_does_not_spend_a_permit(clock):
+    limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=1, max_wait_seconds=1))
+    await limiter.aacquire()
+    waiter = asyncio.create_task(limiter.aacquire())
+    await asyncio.sleep(0)
+    clock[0] = 2
+    with pytest.raises(admission.AdmissionError, match="timed out"):
+        await asyncio.wait_for(waiter, 1)
+    assert not limiter._waiters
+    clock[0] = 60
+    assert limiter.acquire(blocking=False)
+
+
+def test_sync_and_foreign_loops_share_one_budget(clock):
+    limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=1))
+    barrier = threading.Barrier(8)
+
+    def attempt(index):
+        barrier.wait(timeout=5)
+        return limiter.acquire(blocking=False) if index % 2 else asyncio.run(limiter.aacquire(blocking=False))
+
+    with ThreadPoolExecutor(max_workers=8) as executor:
+        assert sum(executor.map(attempt, range(8))) == 1
+
+
+@pytest.mark.parametrize("values", [{"requests_per_minute": 0}, {"requests_per_minute": True}, {"requests_per_minute": 1, "max_wait_seconds": float("inf")}, {"requests_per_minute": 1, "max_queue_size": 0}])
+def test_invalid_configuration(values):
+    with pytest.raises(ValidationError):
+        RequestAdmissionConfig(**values)
+
+
+def test_sync_timeout_removes_waiter(clock, monkeypatch):
+    limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=1, max_wait_seconds=1))
+    limiter.acquire()
+    monkeypatch.setattr(admission.time, "sleep", lambda _: clock.__setitem__(0, 2))
+    with pytest.raises(admission.AdmissionError, match="timed out"):
+        limiter.acquire()
+    assert not limiter._waiters
+
+
+@pytest.mark.asyncio
+async def test_fifo_prevents_newcomers_overtaking(clock):
+    limiter = admission.RequestAdmission(RequestAdmissionConfig(requests_per_minute=60))
+    limiter.acquire()
+    order = []
+
+    async def wait(index):
+        await limiter.aacquire()
+        order.append(index)
+
+    first = asyncio.create_task(wait(1))
+    second = asyncio.create_task(wait(2))
+    await asyncio.sleep(0)
+    try:
+        clock[0] = 1
+        assert not limiter.acquire(blocking=False)
+        await asyncio.wait_for(first, 1)
+        assert order == [1]
+        clock[0] = 2
+        await asyncio.wait_for(second, 1)
+        assert order == [1, 2]
+    finally:
+        first.cancel()
+        second.cancel()
+        await asyncio.gather(first, second, return_exceptions=True)
+
+
+@pytest.fixture
+def registry(monkeypatch):
+    monkeypatch.setattr(admission, "_registry", {})
+
+
+def test_group_sharing_isolation_and_conflict(registry, clock):
+    config = RequestAdmissionConfig(requests_per_minute=60, group="shared")
+    a = admission.get_request_admission("a", config)
+    assert admission.get_request_admission("b", config) is a
+    assert a.acquire(blocking=False)
+    assert not admission.get_request_admission("b", config).acquire(blocking=False)
+    # An implicit model named 'shared' must not collide with that group.
+    b = admission.get_request_admission("shared", RequestAdmissionConfig(requests_per_minute=60))
+    assert b.acquire(blocking=False)
+    with pytest.raises(ValueError, match="restart"):
+        admission.get_request_admission("a", config.model_copy(update={"requests_per_minute": 30}))
+
+
+def make_model(monkeypatch, *, name="a", policy=None, provider=False):
+    from langchain_core.language_models.fake_chat_models import FakeListChatModel
+
+    from deerflow.config.app_config import AppConfig
+    from deerflow.config.model_config import ModelConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+    from deerflow.models import factory
+
+    model = ModelConfig(name=name, use="langchain_openai:ChatOpenAI", model="test", api_key="offline-test-key", request_admission=policy)
+    config = AppConfig(models=[model], sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"))
+    if not provider:
+        monkeypatch.setattr(factory, "resolve_class", lambda *args: FakeListChatModel)
+    return factory.create_chat_model(name, app_config=config, attach_tracing=False, **({} if provider else {"responses": ["ok"]}))
+
+
+@pytest.mark.asyncio
+async def test_factory_invoke_and_stream_share_budget(monkeypatch, registry, clock):
+    policy = RequestAdmissionConfig(requests_per_minute=60, group="account")
+    a = make_model(monkeypatch, policy=policy)
+    b = make_model(monkeypatch, name="b", policy=policy)
+    assert a.rate_limiter is b.rate_limiter
+    assert a.invoke("hello").content == "ok"
+    pending = asyncio.create_task(b.ainvoke("hello"))
+    # Drive the normal BaseChatModel pipeline up to admission, not just a mock.
+    for _ in range(100):
+        if a.rate_limiter._waiters:
+            break
+        await asyncio.sleep(0)
+    try:
+        assert a.rate_limiter._waiters
+        assert not pending.done()
+        clock[0] = 1
+        assert (await asyncio.wait_for(pending, 1)).content == "ok"
+        clock[0] = 2
+        assert "".join(chunk.content for chunk in a.stream("hello")) == "ok"
+        assert not a.rate_limiter.acquire(blocking=False)
+        clock[0] = 3
+        assert "".join([chunk.content async for chunk in b.astream("hello")]) == "ok"
+        assert not a.rate_limiter.acquire(blocking=False)
+    finally:
+        pending.cancel()
+        await asyncio.gather(pending, return_exceptions=True)
+
+
+def test_real_openai_factory_does_not_forward_policy_or_retry_inside_sdk(monkeypatch, registry):
+    model = make_model(monkeypatch, policy=RequestAdmissionConfig(requests_per_minute=30), provider=True)
+    assert isinstance(model.rate_limiter, admission.RequestAdmission)
+    assert model.max_retries == 0
+    assert "request_admission" not in model.model_kwargs
+    assert "request_admission" not in model._default_params
+
+
+def test_disabled_factory_preserves_default(monkeypatch, registry):
+    model = make_model(monkeypatch)
+    assert model.rate_limiter is None
+    assert not admission._registry
+
+
+def test_admission_failures_are_not_retried_as_provider_errors(monkeypatch):
+    from deerflow.agents.middlewares import llm_error_handling_middleware as errors
+    from deerflow.config.app_config import AppConfig
+    from deerflow.config.sandbox_config import SandboxConfig
+
+    monkeypatch.setattr(errors, "_PROCESS_LIMITER", None)
+    monkeypatch.setattr(errors, "_CAP_RESOLVED", False)
+    middleware = errors.LLMErrorHandlingMiddleware(app_config=AppConfig(sandbox=SandboxConfig(use="test")))
+    for message in (
+        "LLM admission queue is full; reduce workload or increase queue capacity.",
+        "LLM admission timed out before dispatch; increase max_wait_seconds or reduce workload.",
+        "rate limit exceeded locally",
+        "provider quota",
+        "server busy",
+    ):
+        retry, _ = middleware._classify_error(admission.AdmissionError(message))
+        assert retry is False

config.example.yaml

@@ -20,7 +20,17 @@
 # ============================================================================
 # Bump this number when the config schema changes.
 # Run `make config-upgrade` to merge new fields into your local config.yaml.
-config_version: 43
+config_version: 44
+
+# Optional per-model request pacing (inside a models[] entry):
+# request_admission:
+#   requests_per_minute: 60
+#   group: shared-provider-account  # optional; defaults to the model config name
+#   max_wait_seconds: 300
+#   max_queue_size: 256
+# Off unless configured. Shared groups require identical settings; restart after
+# changes. Process-local RPM only, not TPM or a cluster-wide quota. See the
+# "Model request admission" section in backend/docs/CONFIGURATION.md.
 
 # ============================================================================
 # Logging

deploy/helm/deer-flow/README.md

@@ -131,7 +131,7 @@ they resolve from the `secrets` map):
 
 ```yaml
 config: |
-  config_version: 43
+  config_version: 44
   models:
     - name: gpt-4
       use: langchain_openai:ChatOpenAI

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: 43
+  config_version: 44
   log_level: info
   recursion_limit: 100
   max_recursion_limit: 1000