Merge 1d8c65afe8 into 87f2d783d6

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`.
 </ParamField>
 
+<ParamField path="LLM_MAX_CONCURRENT_REQUESTS" default="0" type="integer">
+  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.
+</ParamField>
+
+<ParamField path="LLM_REQUEST_DELAY" default="0" type="number">
+  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.
+</ParamField>
+
 <ParamField path="STRIX_LLM_MAX_RETRIES" default="5" type="integer">
   Maximum number of retries for LLM API calls on transient failures.
 </ParamField>

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

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",

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}
+        )