diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx
index c245b364..97e5718f 100644
--- a/docs/advanced/configuration.mdx
+++ b/docs/advanced/configuration.mdx
@@ -42,6 +42,21 @@ Configure Strix using environment variables or a config file.
failing with `LLM CONNECTION FAILED`. Scan requests use `LLM_TIMEOUT`.
+
+ Maximum number of LLM requests in flight at once, across every agent in the
+ process. `0` leaves concurrency unlimited, which is the default. Set a
+ positive value when a provider rate-limits the account and parallel agents
+ would otherwise open too many requests together. The cap covers agent turns,
+ retries, context compaction, and finding deduplication.
+
+
+
+ Minimum seconds between the starts of successive LLM requests. `0` adds no
+ delay. A positive value spaces calls out after a concurrency slot is
+ acquired, so a waiting request cannot start inside another request's gap.
+ The delay is global, not per agent.
+
+
Maximum number of retries for LLM API calls on transient failures.
diff --git a/strix/config/settings.py b/strix/config/settings.py
index 5aa6da06..f598af55 100644
--- a/strix/config/settings.py
+++ b/strix/config/settings.py
@@ -88,6 +88,21 @@ class LlmSettings(BaseSettings):
ge=0,
alias="LLM_MAX_TOOL_CALLS_PER_TURN",
)
+ # 0 means unlimited: every in-flight request starts immediately, which is
+ # the historical behavior. A positive value is a process-wide cap shared by
+ # every agent, including retries and the dedupe judge.
+ max_concurrent_requests: int = Field(
+ default=0,
+ ge=0,
+ alias="LLM_MAX_CONCURRENT_REQUESTS",
+ )
+ # 0 means no spacing. A positive value is the minimum gap between the starts
+ # of successive LLM requests, measured after a concurrency slot is acquired.
+ request_delay: float = Field(
+ default=0.0,
+ ge=0,
+ alias="LLM_REQUEST_DELAY",
+ )
@field_validator("api_type", "reasoning_effort", mode="before")
@classmethod
diff --git a/strix/llm/request_log.py b/strix/llm/request_log.py
index eda0d222..fda67b6f 100644
--- a/strix/llm/request_log.py
+++ b/strix/llm/request_log.py
@@ -30,7 +30,7 @@ import logging
import re
import time
import uuid
-from collections.abc import AsyncIterator, Callable, Mapping, Sequence
+from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping, Sequence
from contextvars import ContextVar, Token
from dataclasses import asdict, dataclass, replace
from datetime import UTC, date, datetime
@@ -888,6 +888,84 @@ def _dispatch(
_litellm_logger: Any | None = None
+class LlmRequestGate:
+ """Process-wide limit on in-flight LLM requests, plus an optional start gap.
+
+ The defaults (unlimited concurrency, no delay) acquire nothing and sleep
+ nothing, so an unconfigured scan behaves exactly as before. A positive
+ concurrency cap is one semaphore for every route — agent turns, retries,
+ compaction, and the dedupe judge — because provider rate limits are
+ account-wide, not per agent. The delay is measured from the previous
+ request's start and only begins once a concurrency slot is held, so a
+ waiting caller cannot start inside another caller's gap.
+ """
+
+ def __init__(self, *, max_concurrent: int = 0, delay: float = 0.0) -> None:
+ self.max_concurrent = max_concurrent
+ self.delay = delay
+ self._semaphore = asyncio.Semaphore(max_concurrent) if max_concurrent > 0 else None
+ self._spacing = asyncio.Lock()
+ self._next_start = 0.0
+
+ @contextlib.asynccontextmanager
+ async def slot(self) -> AsyncGenerator[None]:
+ if self._semaphore is None:
+ async with self._spacing_delay():
+ yield
+ return
+ async with self._semaphore, self._spacing_delay():
+ yield
+
+ @contextlib.asynccontextmanager
+ async def _spacing_delay(self) -> AsyncGenerator[None]:
+ if self.delay <= 0:
+ yield
+ return
+ async with self._spacing:
+ remaining = self._next_start - time.monotonic()
+ if remaining > 0:
+ await asyncio.sleep(remaining)
+ self._next_start = time.monotonic() + self.delay
+ yield
+
+
+_request_gate: LlmRequestGate | None = None
+
+
+def request_gate() -> LlmRequestGate:
+ """The gate currently applied to model calls.
+
+ Built from settings on first use. :func:`configure_request_pacing` replaces
+ it when a scan starts, so a config reload is picked up without restarting
+ the process.
+ """
+ if _request_gate is None:
+ configure_request_pacing()
+ assert _request_gate is not None
+ return _request_gate
+
+
+def configure_request_pacing(
+ *, max_concurrent: int | None = None, delay: float | None = None
+) -> LlmRequestGate:
+ """Install the process-wide request gate from explicit values or settings.
+
+ Callers already waiting on a previous gate keep that gate; only requests
+ that have not yet acquired a slot observe the replacement.
+ """
+ global _request_gate # noqa: PLW0603
+ if max_concurrent is None or delay is None:
+ from strix.config.loader import load_settings
+
+ llm = load_settings().llm
+ if max_concurrent is None:
+ max_concurrent = llm.max_concurrent_requests
+ if delay is None:
+ delay = llm.request_delay
+ _request_gate = LlmRequestGate(max_concurrent=max_concurrent, delay=delay)
+ return _request_gate
+
+
def install() -> None:
"""Attach the LiteLLM capture (idempotent), the native-route reply capture
and the default log-line sink."""
@@ -895,6 +973,7 @@ def install() -> None:
if _log_line_sink not in _sinks:
_sinks.insert(0, _log_line_sink)
_observe_sdk_shared_http_client()
+ configure_request_pacing()
if _litellm_logger is not None:
return
import litellm
@@ -1194,18 +1273,19 @@ class RequestLoggingModel(Model):
reply = HttpReply()
token = _http_reply.set(reply)
try:
- response = await self._inner.get_response(
- system_instructions,
- input,
- model_settings,
- tools,
- output_schema,
- handoffs,
- tracing,
- previous_response_id=previous_response_id,
- conversation_id=conversation_id,
- prompt=prompt,
- )
+ async with request_gate().slot():
+ response = await self._inner.get_response(
+ system_instructions,
+ input,
+ model_settings,
+ tools,
+ output_schema,
+ handoffs,
+ tracing,
+ previous_response_id=previous_response_id,
+ conversation_id=conversation_id,
+ prompt=prompt,
+ )
except BaseException as exc:
if self._should_emit(exc):
emit(
@@ -1236,6 +1316,42 @@ class RequestLoggingModel(Model):
)
return response
+ async def _paced_stream(
+ self,
+ system_instructions: str | None,
+ input: str | list[TResponseInputItem], # noqa: A002
+ model_settings: ModelSettings,
+ tools: list[Tool],
+ output_schema: AgentOutputSchemaBase | None,
+ handoffs: list[Handoff],
+ tracing: ModelTracing,
+ *,
+ previous_response_id: str | None,
+ conversation_id: str | None,
+ prompt: ResponsePromptParam | None,
+ ) -> AsyncIterator[TResponseStreamEvent]:
+ """Yield the inner stream while holding one global request slot.
+
+ The slot lives in this generator, not the caller's ``async for``.
+ Closing or cancelling the outer generator closes this one too, so the
+ slot is released when the stream ends, fails, or is abandoned — not
+ when the first event is handed back.
+ """
+ async with request_gate().slot():
+ async for event in self._inner.stream_response(
+ system_instructions,
+ input,
+ model_settings,
+ tools,
+ output_schema,
+ handoffs,
+ tracing,
+ previous_response_id=previous_response_id,
+ conversation_id=conversation_id,
+ prompt=prompt,
+ ):
+ yield event
+
async def stream_response(
self,
system_instructions: str | None,
@@ -1270,7 +1386,7 @@ class RequestLoggingModel(Model):
reply = HttpReply()
token = _http_reply.set(reply)
try:
- async for event in self._inner.stream_response(
+ async for event in self._paced_stream(
system_instructions,
input,
model_settings,
@@ -1431,12 +1547,14 @@ __all__ = [
"ERROR_MESSAGE_MAX_CHARS",
"LlmCallContext",
"LlmRequestEvent",
+ "LlmRequestGate",
"LlmRequestSink",
"RequestLoggingModel",
"api_host",
"bind_call_context",
"bound_details",
"clean_error_message",
+ "configure_request_pacing",
"current_call_context",
"emit",
"event_from_litellm",
@@ -1447,6 +1565,7 @@ __all__ = [
"json_text",
"merge_details",
"register_sink",
+ "request_gate",
"request_id_from_headers",
"request_id_from_reply",
"request_id_from_text",
diff --git a/tests/test_llm_request_pacing.py b/tests/test_llm_request_pacing.py
new file mode 100644
index 00000000..79e28542
--- /dev/null
+++ b/tests/test_llm_request_pacing.py
@@ -0,0 +1,218 @@
+"""Tests for the global LLM request cap and inter-request delay.
+
+Both settings default to off. A scan that does not set them must start every
+request immediately, including requests that overlap. A positive cap is one
+semaphore for every route; a positive delay is the minimum gap between the
+starts of successive requests.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import TYPE_CHECKING, Any
+
+import pytest
+from agents.items import ModelResponse
+from agents.model_settings import ModelSettings
+from agents.models.interface import Model, ModelTracing
+from agents.usage import Usage
+from pydantic import ValidationError
+
+from strix.config import loader
+from strix.config.loader import load_settings
+from strix.llm import request_log
+
+
+if TYPE_CHECKING:
+ from collections.abc import AsyncIterator, Iterator
+
+ from agents.items import TResponseStreamEvent
+
+
+class _ProbeModel(Model):
+ """Records when each call holds the provider, and can stay there on demand.
+
+ The Model methods match the SDK signature and ignore its arguments: pacing
+ depends on when the call is in flight, not on what it asks the provider.
+ """
+
+ def __init__(self) -> None:
+ self.started: list[float] = []
+ self.in_flight = 0
+ self.peak = 0
+ self.release = asyncio.Event()
+ self.entered = asyncio.Event()
+ self._active = 0
+
+ async def _hold(self) -> ModelResponse:
+ self._active += 1
+ self.in_flight = self._active
+ self.peak = max(self.peak, self._active)
+ self.started.append(time.monotonic())
+ self.entered.set()
+ try:
+ await self.release.wait()
+ finally:
+ self._active -= 1
+ self.in_flight = self._active
+ return ModelResponse(output=[], usage=Usage(), response_id="resp-probe")
+
+ async def get_response(self, *_args: Any, **_kwargs: Any) -> ModelResponse:
+ return await self._hold()
+
+ async def stream_response(
+ self, *_args: Any, **_kwargs: Any
+ ) -> AsyncIterator[TResponseStreamEvent]:
+ await self._hold()
+ if False: # pragma: no cover - keeps this an async generator
+ yield None
+
+
+def _logging(inner: Model) -> request_log.RequestLoggingModel:
+ return request_log.RequestLoggingModel(
+ inner,
+ model_name="probe",
+ provider="openai",
+ base_url=None,
+ )
+
+
+def _call(model: Model) -> Any:
+ return model.get_response(
+ None,
+ "go",
+ ModelSettings(),
+ [],
+ None,
+ [],
+ ModelTracing.DISABLED,
+ previous_response_id=None,
+ conversation_id=None,
+ prompt=None,
+ )
+
+
+def _stream(model: Model) -> AsyncIterator[Any]:
+ return model.stream_response(
+ None,
+ "go",
+ ModelSettings(),
+ [],
+ None,
+ [],
+ ModelTracing.DISABLED,
+ previous_response_id=None,
+ conversation_id=None,
+ prompt=None,
+ )
+
+
+@pytest.fixture
+def _reset_gate(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+ for key in ("LLM_MAX_CONCURRENT_REQUESTS", "LLM_REQUEST_DELAY"):
+ monkeypatch.delenv(key, raising=False)
+ monkeypatch.setattr(loader, "_cached", None)
+ monkeypatch.setattr(loader, "_override", None)
+ monkeypatch.setattr(request_log, "_request_gate", None)
+ yield
+ monkeypatch.setattr(request_log, "_request_gate", None)
+
+
+def test_defaults_leave_requests_unpaced(_reset_gate: None) -> None:
+ llm = load_settings().llm
+
+ gate = request_log.configure_request_pacing()
+
+ assert llm.max_concurrent_requests == 0
+ assert llm.request_delay == 0
+ assert gate.max_concurrent == 0
+ assert gate.delay == 0
+ assert gate._semaphore is None
+
+
+def test_settings_install_the_gate(monkeypatch: pytest.MonkeyPatch, _reset_gate: None) -> None:
+ monkeypatch.setenv("LLM_MAX_CONCURRENT_REQUESTS", "3")
+ monkeypatch.setenv("LLM_REQUEST_DELAY", "1.5")
+ monkeypatch.setattr(loader, "_cached", None)
+
+ gate = request_log.configure_request_pacing()
+
+ assert gate.max_concurrent == 3
+ assert gate.delay == 1.5
+ assert gate._semaphore is not None
+
+
+async def test_default_gate_lets_requests_overlap(_reset_gate: None) -> None:
+ request_log.configure_request_pacing()
+ probe = _ProbeModel()
+ model = _logging(probe)
+
+ calls = [asyncio.create_task(_call(model)) for _ in range(4)]
+ await asyncio.wait_for(probe.entered.wait(), timeout=1)
+ await asyncio.sleep(0.05)
+ assert probe.peak == 4
+ probe.release.set()
+
+ await asyncio.wait_for(asyncio.gather(*calls), timeout=1)
+
+
+async def test_concurrency_cap_is_global(_reset_gate: None) -> None:
+ request_log.configure_request_pacing(max_concurrent=2, delay=0)
+ first = _ProbeModel()
+ second = _ProbeModel()
+ models = (_logging(first), _logging(second))
+
+ calls = [asyncio.create_task(_call(models[i % 2])) for i in range(4)]
+ await asyncio.wait_for(first.entered.wait(), timeout=1)
+ await asyncio.sleep(0.05)
+ assert first.peak + second.peak == 2
+ assert first.in_flight + second.in_flight == 2
+
+ first.release.set()
+ second.release.set()
+ await asyncio.wait_for(asyncio.gather(*calls), timeout=1)
+ assert first.peak + second.peak == 2
+
+
+async def test_delay_spaces_request_starts(_reset_gate: None) -> None:
+ delay = 0.08
+ request_log.configure_request_pacing(max_concurrent=0, delay=delay)
+ probe = _ProbeModel()
+ probe.release.set()
+ model = _logging(probe)
+
+ await asyncio.wait_for(asyncio.gather(*[_call(model) for _ in range(3)]), timeout=1)
+
+ gaps = [probe.started[i + 1] - probe.started[i] for i in range(len(probe.started) - 1)]
+ assert len(gaps) == 2
+ assert all(gap >= delay * 0.85 for gap in gaps)
+
+
+async def test_open_stream_holds_its_slot_until_closed(_reset_gate: None) -> None:
+ request_log.configure_request_pacing(max_concurrent=1, delay=0)
+ streaming = _ProbeModel()
+ waiting = _ProbeModel()
+ waiting.release.set()
+
+ stream = _stream(_logging(streaming))
+ reader = asyncio.create_task(stream.__anext__())
+ await asyncio.wait_for(streaming.entered.wait(), timeout=1)
+
+ follower = asyncio.create_task(_call(_logging(waiting)))
+ await asyncio.sleep(0.05)
+ assert waiting.started == []
+
+ streaming.release.set()
+ with pytest.raises(StopAsyncIteration):
+ await asyncio.wait_for(reader, timeout=1)
+ await stream.aclose()
+ await asyncio.wait_for(follower, timeout=1)
+ assert len(waiting.started) == 1
+
+
+def test_negative_values_are_rejected() -> None:
+ with pytest.raises(ValidationError, match="greater than or equal to 0"):
+ load_settings().llm.__class__.model_validate(
+ {"LLM_MAX_CONCURRENT_REQUESTS": -1, "LLM_REQUEST_DELAY": -0.1}
+ )