Bound and offload grammar validation for response_format

docs/en/advance/structed_output.md

@@ -106,3 +106,16 @@ response = client.chat.completions.create(
     top_p=0.8)
 print(response)
 ```
+
+## Limits
+
+To protect the server's availability, grammar sources supplied through `response_format` are
+validated before compilation, and requests exceeding the following limits are rejected with
+HTTP 400:
+
+- the serialized grammar source (JSON schema, regex, or structural tag) may not exceed 16 KiB;
+- a JSON schema may not exceed a nesting depth of 128 raw JSON levels (a nested object spans
+  two levels through its `properties` wrapper), since grammar compile cost grows exponentially
+  with depth;
+- validation runs off the event loop with a 5-second budget; a schema that still takes longer
+  is rejected.

docs/zh_cn/advance/structed_output.md

@@ -108,3 +108,13 @@ print(response)
 ```
 
 输出结果是一个 json 格式的回答。
+
+## 限制
+
+为保障服务的可用性，通过 `response_format` 提供的语法源在编译前会先做校验，超出以下限制的请求
+将以 HTTP 400 拒绝：
+
+- 序列化后的语法源（JSON schema、正则或 structural tag）不超过 16 KiB；
+- JSON schema 的原始 JSON 嵌套深度不超过 128 层（嵌套对象经由 `properties` 包装，每层占两级），
+  因为语法编译开销随深度呈指数增长；
+- 校验在事件循环之外的工作线程中执行，预算为 5 秒，超时的 schema 同样会被拒绝。

lmdeploy/_guided_decoding.py

@@ -1,11 +1,73 @@
 # Copyright (c) OpenMMLab. All rights reserved.
 """Backend-neutral guided-decoding grammar compilation."""
+
 from __future__ import annotations
 
+import asyncio
 import json
+from concurrent.futures import ThreadPoolExecutor
 from functools import lru_cache
 from typing import Any
 
+# Bounds for client-supplied grammar sources. The compile cost of a nested
+# JSON schema grows exponentially with its depth and XGrammar exposes no
+# depth or time limit on that path (as of 0.2.x), so unbounded schemas must be
+# rejected before compilation.
+MAX_GRAMMAR_SOURCE_BYTES = 16 * 1024
+# Raw JSON nesting depth of a grammar source (a nested object spans two JSON
+# levels per schema level through its "properties" wrapper). XGrammar's
+# compile cost grows exponentially with this depth, so keep the worst
+# accepted compile well under a second. Realistic structured-output schemas
+# stay far below the limit.
+MAX_JSON_NESTING_DEPTH = 128
+# Hard budget for one grammar validation. The work itself runs in a worker
+# thread, so this only bounds how long a request may wait on it.
+GRAMMAR_COMPILE_TIMEOUT = 5.0
+
+
+def _max_nesting_depth(value: Any) -> int:
+    """Return the nesting depth of a JSON-like structure.
+
+    Raises ValueError when the depth exceeds MAX_JSON_NESTING_DEPTH or the
+    structure contains a cycle. Every container type json.dumps serializes
+    (dict, list, tuple) counts toward the depth, and scalars do not.
+
+    Recursive within the cap on purpose: the walk stops at
+    MAX_JSON_NESTING_DEPTH + 1 frames, far below Python's recursion limit,
+    and at the first cycle instead of spinning.
+    """
+    active: set[int] = set()
+
+    def visit(node: Any, depth: int) -> int:
+        if not isinstance(node, (dict, list, tuple)):
+            return depth - 1
+        if depth > MAX_JSON_NESTING_DEPTH:
+            raise ValueError(
+                f'json_schema exceeds the maximum nesting depth of {MAX_JSON_NESTING_DEPTH}.')
+        if id(node) in active:
+            raise ValueError('circular reference detected in grammar source')
+        active.add(id(node))
+        try:
+            children = node.values() if isinstance(node, dict) else node
+            return max((visit(child, depth + 1) for child in children), default=depth)
+        finally:
+            active.discard(id(node))
+
+    return visit(value, 1)
+
+
+def _check_source_size(source: str) -> None:
+    # Count UTF-8 bytes, not code points: a serialized source keeps non-ASCII
+    # characters (json.dumps with ensure_ascii=False), so code points can
+    # understate the real size by up to 4x.
+    size = len(source.encode('utf-8'))
+    if size > MAX_GRAMMAR_SOURCE_BYTES:
+        raise ValueError(f'grammar source exceeds the maximum size of {MAX_GRAMMAR_SOURCE_BYTES} bytes.')
+
+
+def _check_schema_depth(schema: Any) -> None:
+    _max_nesting_depth(schema)
+
 
 def _json_schema_from_response_format(response_format: dict[str, Any]) -> str:
     schema: Any = response_format['json_schema']
@@ -15,29 +77,47 @@ def _json_schema_from_response_format(response_format: dict[str, Any]) -> str:
                 schema = schema[key]
                 break
     if isinstance(schema, (dict, bool)):
+        _check_schema_depth(schema)
         return json.dumps(schema, ensure_ascii=False)
     if isinstance(schema, str):
+        _check_source_size(schema)
+        try:
+            parsed = json.loads(schema)
+        except RecursionError as err:
+            raise ValueError(f'json_schema exceeds the maximum nesting depth of {MAX_JSON_NESTING_DEPTH}.') from err
+        _check_schema_depth(parsed)
         return schema
     raise ValueError(
         f'Cannot parse schema {schema}. The schema must be either a dictionary '
-        'or a string that contains the JSON Schema specification')
+        'or a string that contains the JSON Schema specification'
+    )
 
 
 def _grammar_source(response_format: dict[str, Any]) -> tuple[str, str]:
     """Return the XGrammar entry point and its serialized input."""
     schema_type = response_format.get('type', 'text')
     if schema_type == 'json_schema':
-        return schema_type, _json_schema_from_response_format(response_format)
-    if schema_type == 'regex_schema':
-        return schema_type, str(response_format.get('regex_schema', ''))
-    if schema_type == 'json_object':
-        schema = json.dumps({'type': 'object', 'additionalProperties': True})
-        return 'json_schema', schema
-    if schema_type == 'structural_tag':
-        return schema_type, json.dumps(response_format, ensure_ascii=False)
-    if schema_type == 'text':
+        source = _json_schema_from_response_format(response_format)
+    elif schema_type == 'regex_schema':
+        source = str(response_format.get('regex_schema', ''))
+    elif schema_type == 'json_object':
+        source = json.dumps({'type': 'object', 'additionalProperties': True})
+        schema_type = 'json_schema'
+    elif schema_type == 'structural_tag':
+        # Built by the tool-call parsers and the gpt-oss response parser (the
+        # latter wraps client-supplied schemas), and also accepted directly by
+        # GenerationConfig. Only the size bound applies here: XGrammar guards
+        # the tag's own format tree with its recursion guard, pydantic
+        # serialization caps how deep the gpt-oss parser can wrap a schema,
+        # and an embedded schema additionally passes the EBNF parser's fixed
+        # nest limit.
+        source = json.dumps(response_format, ensure_ascii=False)
+    elif schema_type == 'text':
         return schema_type, ''
-    raise ValueError(f'unsupported format type: {schema_type}')
+    else:
+        raise ValueError(f'unsupported format type: {schema_type}')
+    _check_source_size(source)
+    return schema_type, source
 
 
 @lru_cache(maxsize=128)
@@ -57,15 +137,38 @@ def _check_response_format(serialized_format: str) -> None:
         xgr.Grammar.from_structural_tag(source)
 
 
-def ensure_response_format_compilable(response_format: dict[str, Any]) -> None:
-    """Reject response formats that XGrammar cannot compile."""
+def _ensure_response_format_compilable(response_format: dict[str, Any]) -> None:
+    """Synchronous validation body; must not run on an event loop."""
     try:
         serialized_format = json.dumps(response_format, ensure_ascii=False, separators=(',', ':'))
         _check_response_format(serialized_format)
-    except (KeyError, RuntimeError, TypeError) as err:
+    except (KeyError, RecursionError, RuntimeError, TypeError) as err:
         raise ValueError(f'Unsupported response format: {err}') from err
 
 
+VALIDATION_EXECUTOR = ThreadPoolExecutor(max_workers=2, thread_name_prefix='grammar-validate')
+
+
+async def ensure_response_format_compilable(response_format: dict[str, Any]) -> None:
+    """Reject response formats that XGrammar cannot compile.
+
+    The compile runs in a dedicated worker pool under a hard timeout:
+    XGrammar has no compile-time bound of its own, and a synchronous compile
+    on the event loop would stall every request, stream, and health check in
+    the process. A compile that overruns the timeout cannot be cancelled, but
+    it only ever occupies this small pool, never the interpreter's default
+    executor shared by other offloaded work.
+    """
+    loop = asyncio.get_running_loop()
+    try:
+        await asyncio.wait_for(
+            loop.run_in_executor(VALIDATION_EXECUTOR, _ensure_response_format_compilable, response_format),
+            timeout=GRAMMAR_COMPILE_TIMEOUT,
+        )
+    except asyncio.TimeoutError as err:
+        raise ValueError(f'Response format validation timed out after {GRAMMAR_COMPILE_TIMEOUT}s.') from err
+
+
 def compile_response_format(compiler, response_format: dict[str, Any]):
     """Compile one public or internal response format with XGrammar.
 

lmdeploy/serve/core/async_engine.py

@@ -530,8 +530,6 @@ def remove_session():
             if (messages is not None) ^ (input_ids is None):
                 raise RequestError(ErrorCode.INVALID_REQUEST,
                                    'You must specify exactly one of messages or input_ids.')
-            if gen_config is not None and gen_config.response_format is not None:
-                ensure_response_format_compilable(gen_config.response_format)
             if isinstance(session_id, Session):
                 session = session_id
             elif isinstance(session_id, int):
@@ -545,6 +543,11 @@ def remove_session():
                 raise RequestError(
                     ErrorCode.REQUEST_CONFLICT,
                     f'Session {session.session_id} already has an active request.')
+            if gen_config is not None and gen_config.response_format is not None:
+                # Resolve the session first: a rejected response format raises
+                # ValueError below, and remove_session() can only clean up a
+                # session that this method has already taken charge of.
+                await ensure_response_format_compilable(gen_config.response_format)
 
             chat_template_kwargs = chat_template_kwargs or {}
             if enable_thinking is not None: