Codec: address PR review — async dispatch, cached registry, hardened varint

Mirrors the sglang fork PR fixes (sgl-project/sglang#25544):

1. codec_frame.py: numpy-free LE-uint32 unpack now uses
   `struct.unpack('<NI', b)` (~10× faster than the per-element
   list comprehension) and rejects buffers whose length is not a
   multiple of 4 instead of silently corrupting.

2. codec_frame.py: `_decode_varint` gains bounds-check +
   shift-cap (35 bits = 5 bytes, the max uint32 varint width).
   Used by every length-delimited field decode including the
   packed `prompt_ids` loop. Malformed or malicious input fails
   fast with a clear ValueError instead of looping unbounded.

3. codec_dispatcher.py: add `dispatch_call_async`, a
   `asyncio.to_thread`-wrapping variant of `dispatch_call`. The
   sync form does a blocking `urllib.request.urlopen` POST that
   would freeze the event loop if called from an `async def`
   request handler.

4. openai/chat_completion/serving.py: cache the `ToolRegistry`
   on `OpenAIServingChat` rather than calling
   `ToolRegistry.from_env` per request. `from_env` performs
   blocking HTTP fetches against manifest URLs; the first request
   after process start pays the cost (off-loop via
   `asyncio.to_thread`) and every subsequent request reuses the
   cached registry. Same site now uses `dispatch_call_async` so
   tool dispatches don't block the worker either.

Wire format unchanged. No new dependencies.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: William Dunn <wdunn001@gmail.com>

Author: wdunn001

vllm/entrypoints/codec_dispatcher.py

@@ -197,8 +197,11 @@ def dispatch_call(
 ) -> CodecToolResult:
     """POST a CodecToolCall to ``tool.endpoint`` and decode the response.
 
-    Synchronous + stdlib-only. Engines that want async dispatch can
-    wrap this in a thread or replace with their preferred HTTP client.
+    Synchronous + stdlib-only. ASGI engines should NOT call this
+    directly from inside the async request loop — `urlopen` blocks the
+    event loop. Use `dispatch_call_async` instead (it wraps this in
+    `asyncio.to_thread`). The sync form stays for non-async callers
+    (CLI tools, batch eval drivers).
     """
     import urllib.request
     call = CodecToolCall(
@@ -218,6 +221,24 @@ def dispatch_call(
     return decode_tool_result(body)
 
 
+async def dispatch_call_async(
+    tool: RegisteredTool,
+    arguments_json: str,
+    call_id: str,
+) -> CodecToolResult:
+    """Async variant of :func:`dispatch_call`.
+
+    Runs the blocking urllib POST in a worker thread via
+    `asyncio.to_thread` so it doesn't block the event loop. ASGI engines
+    (vLLM's chat / completion serving paths) MUST use this variant —
+    calling the sync `dispatch_call` from inside an `async def` request
+    handler blocks every other in-flight request on the same uvicorn
+    worker until the tool replies.
+    """
+    import asyncio
+    return await asyncio.to_thread(dispatch_call, tool, arguments_json, call_id)
+
+
 # ── Reinjection hook ───────────────────────────────────────────────────────
 
 
@@ -244,6 +265,7 @@ def reinject_ids_into_context(context_ids: list[int], reinject_ids: list[int]) -
     "ToolRegistry",
     "decode_tool_result",
     "dispatch_call",
+    "dispatch_call_async",
     "encode_tool_call",
     "reinject_ids_into_context",
 ]

vllm/entrypoints/codec_frame.py

@@ -63,8 +63,15 @@ def _normalise_ids_to_list(ids: IdsLike) -> list[int]:
     if isinstance(ids, (bytes, bytearray, memoryview)):
         if _HAVE_NUMPY:
             return _np.frombuffer(bytes(ids), dtype="<u4").tolist()
+        # Stdlib LE-uint32 unpack as the numpy-free fallback. `struct.unpack`
+        # is ~10× faster than a Python list comprehension with
+        # `int.from_bytes` — same shape, no per-element Python loop.
         b = bytes(ids)
-        return [int.from_bytes(b[i:i + 4], "little") for i in range(0, len(b), 4)]
+        if len(b) % 4 != 0:
+            raise ValueError(
+                f"codec_frame: bytes length {len(b)} is not a multiple of 4 (uint32 LE)"
+            )
+        return list(_struct.unpack(f"<{len(b) // 4}I", b))
     # Fall back to the Sequence path.
     return list(ids)
 
@@ -121,14 +128,26 @@ def _varint(n: int) -> bytes:
 
 
 def _decode_varint(data: bytes, pos: int) -> tuple[int, int]:
+    """Decode a single protobuf varint at ``pos``.
+
+    Bounds-checked + shift-capped so malformed or malicious input fails
+    fast with a ValueError instead of looping unbounded or producing a
+    silently-wrong value. Cap is 35 bits — protobuf uint32 fits in <= 5
+    bytes (5 * 7 = 35); a 6th continuation byte means the encoded value
+    cannot represent a uint32 and we reject it.
+    """
     result = shift = 0
     while True:
+        if pos >= len(data):
+            raise ValueError("Codec: truncated varint in CodecRequest")
         b = data[pos]
         pos += 1
         result |= (b & 0x7F) << shift
         if not (b & 0x80):
             return result, pos
         shift += 7
+        if shift > 35:
+            raise ValueError("Codec: varint overflow in CodecRequest")
 
 
 def _encode_tool_call_msg(call: dict) -> bytes:

vllm/entrypoints/openai/chat_completion/serving.py

@@ -23,6 +23,7 @@
     CODEC_BOLT_ON_DISPATCH,
     ToolRegistry,
     dispatch_call,
+    dispatch_call_async,
     reinject_ids_into_context,
 )
 from vllm.entrypoints.codec_frame import encode_frame
@@ -188,6 +189,15 @@ def __init__(
         self.supports_code_interpreter = False
         self.python_tool = None
 
+        # v0.5 #87: lazy-cached Codec ToolRegistry. `ToolRegistry.from_env`
+        # performs blocking HTTP fetches against manifest URLs; doing that
+        # per-request blocks the event loop and adds startup latency to
+        # every tool-watcher stream. Resolve once per process on first use
+        # and reuse the registry thereafter. Sentinel = "not yet attempted";
+        # None after load = attempted but env not configured.
+        self._codec_dispatcher_registry: ToolRegistry | None = None
+        self._codec_dispatcher_registry_loaded: bool = False
+
     def warmup(self) -> None:
         self.renderer.warmup(
             ChatParams(
@@ -1081,8 +1091,19 @@ async def chat_completion_binary_stream_generator(
         # tool's response_ids back into the stream.
         dispatcher_registry: ToolRegistry | None = None
         if CODEC_BOLT_ON_DISPATCH and watcher is not None:
-            tokenizer_hash = getattr(self, "_codec_tokenizer_map_hash", "")
-            dispatcher_registry = ToolRegistry.from_env(tokenizer_hash)
+            # Cached on the serving instance — ToolRegistry.from_env makes
+            # blocking HTTP fetches against manifest URLs, so it must NEVER
+            # run inside the async request loop. First request after process
+            # start pays the fetch cost (off-loop via to_thread); subsequent
+            # requests reuse the cached registry.
+            if not self._codec_dispatcher_registry_loaded:
+                tokenizer_hash = getattr(self, "_codec_tokenizer_map_hash", "")
+                import asyncio as _asyncio
+                self._codec_dispatcher_registry = await _asyncio.to_thread(
+                    ToolRegistry.from_env, tokenizer_hash
+                )
+                self._codec_dispatcher_registry_loaded = True
+            dispatcher_registry = self._codec_dispatcher_registry
 
         try:
             async for res in result_generator:
@@ -1119,7 +1140,14 @@ async def chat_completion_binary_stream_generator(
                             tool = dispatcher_registry.get(ev.name)
                             if tool is not None and tool.mode == "dispatch":
                                 try:
-                                    result = dispatch_call(
+                                    # Use the async variant — sync `dispatch_call`
+                                    # does a blocking urllib POST that would
+                                    # freeze the event loop if called from this
+                                    # `async def`. `dispatch_call_async` wraps
+                                    # it in asyncio.to_thread so other
+                                    # concurrent requests on the worker keep
+                                    # flowing while the tool reply is in flight.
+                                    result = await dispatch_call_async(
                                         tool,
                                         arguments_json=ev.arguments_json,
                                         call_id=ev.id or make_call_id(next_call_seq),