API server rate limiting and DoS protection

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #50007 feat(api-server): configurable concurrent-run cap to prevent DoS (#7483)

  merged to main

  files: gateway/platforms/api_server.py, hermes_cli/config.py, tests/gateway/test_api_server.py

  Show the diff

  • gateway/platforms/api_server.py

    @@ -782,6 +782,15 @@ def __init__(self, config: PlatformConfig):
             # in-flight run by run_id.
             self._run_approval_sessions: Dict[str, str] = {}
             self._session_db: Optional[Any] = None  # Lazy-init SessionDB for session continuity
    +        # Concurrency cap shared across all agent-serving endpoints
    +        # (/v1/chat/completions, /v1/responses, /v1/runs). Read from
    +        # config.yaml gateway.api_server.max_concurrent_runs; 0 disables
    +        # the cap. Bounds CPU / memory / upstream-LLM-quota exhaustion
    +        # from a request flood (#7483).
    +        self._max_concurrent_runs: int = self._resolve_max_concurrent_runs()
    +        # Number of in-flight runs on the non-streaming chat/responses paths
    +        # (the /v1/runs path tracks its own in-flight set via _run_streams).
    +        self._inflight_agent_runs: int = 0
     
         @staticmethod
         def _parse_cors_origins(value: Any) -> tuple[str, ...]:
    @@ -798,6 +807,30 @@ def _parse_cors_origins(value: Any) -> tuple[str, ...]:
     
             return tuple(str(item).strip() for item in items if str(item).strip())
     
    +    @staticmethod
    +    def _resolve_max_concurrent_runs() -> int:
    +        """Read the concurrent-run cap from config.yaml (0 disables).
    +
    +        gateway.api_server.max_concurrent_runs. Falls back to the historical
    +        default of 10 when unset or malformed. Negative values are clamped
    +        to 0 (disabled).
    +        """
    +        default = 10
    +        try:
    +            from hermes_cli.config import cfg_get, load_config
    +
    +            raw = cfg_get(
    +                load_config(),
    +                "gateway",
    +                "api_server",
    +                "max_concurrent_runs",
    +                default=default,
    +            )
    +            value = int(raw)
    +        except Exception:
    +            return default
    +        return max(0, value)
    +
         @staticmethod
         def _resolve_model_name(explicit: str) -> str:
             """Derive the advertised model name for /v1/models.
    @@ -1748,6 +1781,11 @@ async def _handle_chat_completions(self, request: "web.Request") -> "web.Respons
             if auth_err:
                 return auth_err
     
    +        # Bound total in-flight agent runs (configurable; #7483).
    +        limited = self._concurrency_limited_response()
    +        if limited is not None:
    +            return limited
    +
             # Parse request body
             try:
                 body = await request.json()
    @@ -2817,6 +2855,11 @@ async def _handle_responses(self, request: "web.Request") -> "web.Response":
             if auth_err:
                 return auth_err
     
    +        # Bound total in-flight agent runs (configurable; #7483).
    +        limited = self._concurrency_limited_response()
    +        if limited is not None:
    +            return limited
    +
             # Long-term memory scope header (see chat_completions for details).
             gateway_session_key, key_err = self._parse_session_key_header(request)
             if key_err is not None:
    @@ -3568,6 +3611,31 @@ def _extract_output_items(result: Dict[str, Any], start_index: int = 0) -> List[
         # Agent execution
         # ------------------------------------------------------------------
     
    +    def _concurrency_limited_response(self) -> Optional["web.Response"]:
    +        """Return a 429 response if the concurrent-run cap is reached, else None.
    +
    +        The cap bounds total in-flight agent activity across every
    +        agent-serving endpoint: the non-streaming chat/responses paths
    +        (tracked by ``_inflight_agent_runs``) plus the ``/v1/runs`` streaming
    +        path (tracked by ``_run_streams``). A configured value of 0 disables
    +        the cap entirely.
    +        """
    +        limit = self._max_concurrent_runs
    +        if limit <= 0:
    +            return None
    +        inflight = self._inflight_agent_runs + len(self._run_streams)
    +        if inflight >= limit:
    +            return web.json_response(
    +                _openai_error(
    +                    f"Too many concurrent runs (max {limit})",
    +                    err_type="rate_limit_error",
    +                    code="rate_limit_exceeded",
    +                ),
    +                status=429,
    +                headers={"Retry-After": "1"},
    +            )
    +        return None
    +
         async def _run_agent(
             self,
             user_message: str,
    @@ -3636,13 +3704,16 @@ def _run():
                 finally:
                     clear_session_vars(tokens)
     
    -        return await loop.run_in_executor(None, _run)
    +        self._inflight_agent_runs += 1
    +        try:
    +            return await loop.run_in_executor(None, _run)
    +        finally:
    +            self._inflight_agent_runs -= 1
     
         # ------------------------------------------------------------------
         # /v1/runs — structured event streaming
         # ------------------------------------------------------------------
     
    -    _MAX_CONCURRENT_RUNS = 10  # Prevent unbounded resource allocation
         _RUN_STREAM_TTL = 300  # seconds before orphaned runs are swept
         _RUN_STATUS_TTL = 3600  # seconds to retain terminal run status for polling
     
    @@ -3718,12 +3789,11 @@ async def _handle_runs(self, request: "web.Request") -> "web.Response":
             if key_err is not None:
                 return key_err
     
    -        # Enforce concurrency limit
    -        if len(self._run_streams) >= self._MAX_CONCURRENT_RUNS:
    -            return web.json_response(
    -                _openai_error(f"Too many concurrent runs (max {self._MAX_CONCURRENT_RUNS})", code="rate_limit_exceeded"),
    -                status=429,
    -            )
    +        # Enforce concurrency limit (shared across all agent-serving
    +        # endpoints; configurable via gateway.api_server.max_concurrent_runs).
    +        limited = self._concurrency_limited_response()
    +        if limited is not None:
    +            return limited
     
             try:
                 body = await request.json()

  • hermes_cli/config.py

    @@ -2511,6 +2511,18 @@ def _ensure_hermes_home_managed(home: Path):
             # multi-tool agent turn. Bridged to HERMES_MEDIA_TRUST_RECENT_SECONDS.
             # Only consulted when ``strict`` is true.
             "trust_recent_files_seconds": 600,
    +
    +        # OpenAI-compatible API server platform
    +        # (gateway/platforms/api_server.py).
    +        "api_server": {
    +            # Maximum number of agent runs the API server will service
    +            # concurrently. Requests to /v1/chat/completions, /v1/responses,
    +            # and /v1/runs that arrive while this many runs are already
    +            # in flight are rejected with HTTP 429 + a Retry-After header,
    +            # bounding CPU / memory / upstream-LLM-quota exhaustion from a
    +            # request flood. Set to 0 to disable the cap entirely.
    +            "max_concurrent_runs": 10,
    +        },
         },
     
         # Real-time token streaming to messaging platforms (Telegram, Discord,

  • tests/gateway/test_api_server.py

    @@ -420,6 +420,63 @@ def test_malformed_auth_header_returns_401(self):
             assert result.status == 401
     
     
    +# ---------------------------------------------------------------------------
    +# Concurrency cap (gateway.api_server.max_concurrent_runs) — #7483
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestConcurrencyCap:
    +    def test_resolve_defaults_to_10_when_unset(self):
    +        with patch("hermes_cli.config.load_config", return_value={}):
    +            assert APIServerAdapter._resolve_max_concurrent_runs() == 10
    +
    +    def test_resolve_reads_config_value(self):
    +        cfg = {"gateway": {"api_server": {"max_concurrent_runs": 3}}}
    +        with patch("hermes_cli.config.load_config", return_value=cfg):
    +            assert APIServerAdapter._resolve_max_concurrent_runs() == 3
    +
    +    def test_resolve_clamps_negative_to_zero(self):
    +        cfg = {"gateway": {"api_server": {"max_concurrent_runs": -5}}}
    +        with patch("hermes_cli.config.load_config", return_value=cfg):
    +            assert APIServerAdapter._resolve_max_concurrent_runs() == 0
    +
    +    def test_resolve_malformed_falls_back_to_default(self):
    +        cfg = {"gateway": {"api_server": {"max_concurrent_runs": "not-an-int"}}}
    +        with patch("hermes_cli.config.load_config", return_value=cfg):
    +            assert APIServerAdapter._resolve_max_concurrent_runs() == 10
    +
    +    def test_under_cap_returns_none(self):
    +        adapter = _make_adapter()
    +        adapter._max_concurrent_runs = 5
    +        adapter._inflight_agent_runs = 2
    +        assert adapter._concurrency_limited_response() is None
    +
    +    def test_at_cap_returns_429_with_retry_after(self):
    +        adapter = _make_adapter()
    +        adapter._max_concurrent_runs = 3
    +        adapter._inflight_agent_runs = 3
    +        resp = adapter._concurrency_limited_response()
    +        assert resp is not None
    +        assert resp.status == 429
    +        assert resp.headers.get("Retry-After")
    +
    +    def test_cap_counts_both_buckets(self):
    +        # /v1/runs (tracked by _run_streams) + chat/responses (inflight)
    +        adapter = _make_adapter()
    +        adapter._max_concurrent_runs = 4
    +        adapter._inflight_agent_runs = 2
    +        adapter._run_streams = {"r1": object(), "r2": object()}
    +        resp = adapter._concurrency_limited_response()
    +        assert resp is not None
    +        assert resp.status == 429
    +
    +    def test_zero_disables_cap(self):
    +        adapter = _make_adapter()
    +        adapter._max_concurrent_runs = 0
    +        adapter._inflight_agent_runs = 9999
    +        assert adapter._concurrency_limited_response() is None
    +
    +
     # ---------------------------------------------------------------------------
     # Helpers for HTTP tests
     # ---------------------------------------------------------------------------