fix(frontend): evict async tokenizer executors when the tokenizer is retired

components/src/dynamo/frontend/prepost.py

@@ -5,6 +5,7 @@
 
 import json
 import os
+import weakref
 from collections.abc import Awaitable, Callable, Sequence
 from concurrent.futures import ThreadPoolExecutor
 from copy import deepcopy
@@ -55,7 +56,22 @@ class PreprocessResult:
     uses_dynamo_json_tool_call_fallback: bool = False
 
 
-_ASYNC_TOKENIZER_POOL: dict[int, Callable[..., Awaitable[Any]]] = {}
+# One executor per live tokenizer. Keyed weakly by the tokenizer object
+# itself (not id(), whose reuse could alias a stale entry): when the owning
+# tokenizer is garbage-collected the entry drops out and the finalizer shuts
+# the executor down. The executor must not be captured by a cached wrapper
+# that also captures the tokenizer (vLLM's make_async closes over both) —
+# that value->key strong reference would pin the key forever.
+_ASYNC_TOKENIZER_EXECUTORS: weakref.WeakKeyDictionary[
+    TokenizerLike, ThreadPoolExecutor
+] = weakref.WeakKeyDictionary()
+# Fallback for tokenizers that do not support weak references; retains
+# entries for the process lifetime (the previous behavior for all tokenizers).
+# The tokenizer is stored alongside its executor so its id() cannot be
+# recycled by a different tokenizer while the entry lives.
+_STRONG_ASYNC_TOKENIZER_EXECUTORS: dict[
+    int, tuple[TokenizerLike, ThreadPoolExecutor]
+] = {}
 SKIP_REQUEST_VALIDATION = os.getenv("DYN_VLLM_SKIP_REQUEST_VALIDATION", "1") == "1"
 
 
@@ -337,14 +353,22 @@ def _build_assistant_guided_decoding(
 
 
 def _get_async_tokenizer(tokenizer: TokenizerLike) -> Callable[..., Awaitable[Any]]:
-    key = id(tokenizer)
-    async_tokenizer = _ASYNC_TOKENIZER_POOL.get(key)
-    if async_tokenizer is None:
-        async_tokenizer = make_async(
-            tokenizer, executor=ThreadPoolExecutor(max_workers=1)
-        )
-        _ASYNC_TOKENIZER_POOL[key] = async_tokenizer
-    return async_tokenizer
+    try:
+        executor = _ASYNC_TOKENIZER_EXECUTORS.get(tokenizer)
+        if executor is None:
+            executor = ThreadPoolExecutor(max_workers=1)
+            _ASYNC_TOKENIZER_EXECUTORS[tokenizer] = executor
+            weakref.finalize(tokenizer, executor.shutdown)
+    except TypeError:
+        # Tokenizer does not support weak references.
+        key = id(tokenizer)
+        entry = _STRONG_ASYNC_TOKENIZER_EXECUTORS.get(key)
+        if entry is None:
+            executor = ThreadPoolExecutor(max_workers=1)
+            _STRONG_ASYNC_TOKENIZER_EXECUTORS[key] = (tokenizer, executor)
+        else:
+            executor = entry[1]
+    return make_async(tokenizer, executor=executor)
 
 
 def _materialize_assistant_tool_calls(

docs/fern/pages/reference/api/python/frontend.mdx

@@ -130,7 +130,7 @@ from dynamo.frontend.prepost import PreprocessResult
 PreprocessResult(request_for_sampling: ChatCompletionRequest, tool_parser: ToolParser | None, chat_template_kwargs: dict[str, Any], engine_prompt: dict[str, Any], prompt_token_ids: list[int], guided_decoding: dict[str, Any] | None = None, uses_dynamo_json_tool_call_fallback: bool = False) -> None
 ```
 
-[`components/src/dynamo/frontend/prepost.py#L47`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L47)
+[`components/src/dynamo/frontend/prepost.py#L48`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L48)
 
 **Public methods**
 
@@ -331,7 +331,7 @@ from dynamo.frontend.prepost import StreamingPostProcessor
 StreamingPostProcessor(*, tokenizer: TokenizerLike, request_for_sampling: ChatCompletionRequest, sampling_params: SamplingParams, prompt_token_ids: Sequence[int], tool_parser: ToolParser | None, reasoning_parser_class: type[ReasoningParser] | None, chat_template_kwargs: dict[str, Any], stream_response: bool = True, uses_dynamo_json_tool_call_fallback: bool = False) -> None
 ```
 
-[`components/src/dynamo/frontend/prepost.py#L649`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L649)
+[`components/src/dynamo/frontend/prepost.py#L671`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L671)
 
 **Public methods**
 
@@ -343,7 +343,7 @@ __init__(*, tokenizer: TokenizerLike, request_for_sampling: ChatCompletionReques
 
 No summary available.
 
-[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L650)
+[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L672)
 
 <h4 id="api-dynamo-frontend-prepost-streamingpostprocessor-process-output">process_output</h4>
 
@@ -353,7 +353,7 @@ process_output(output: Any) -> dict[str, Any] | None
 
 No summary available.
 
-[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L1048)
+[source](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L1070)
 </Accordion>
 
 <Accordion id="api-dynamo-frontend-vllm-processor-vllmprocessor" title="VllmProcessor (class)">
@@ -448,7 +448,7 @@ from dynamo.frontend.prepost import build_tool_call_guided_decoding
 build_tool_call_guided_decoding(request: ChatCompletionRequest, tool_parser: ToolParser | None, *, parser_guided_decoding: dict[str, Any] | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto') -> dict[str, Any] | None
 ```
 
-[`components/src/dynamo/frontend/prepost.py#L209`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L209)
+[`components/src/dynamo/frontend/prepost.py#L223`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L223)
 </Accordion>
 
 <Accordion id="api-dynamo-frontend-sglang-prepost-build-tool-call-guided-decoding" title="build_tool_call_guided_decoding (function)">
@@ -700,7 +700,7 @@ from dynamo.frontend.prepost import preprocess_chat_request
 preprocess_chat_request(request: dict[str, Any] | ChatCompletionRequest, *, tokenizer: TokenizerLike, renderer: _Renderer, tool_parser_class: type[ToolParser] | None, exclude_tools_when_tool_choice_none: bool = True, enable_auto_tool_choice: bool = False, default_chat_template_kwargs: dict[str, Any] | None = None, default_thinking_mode: str | None = None, structural_tag_mode: str = 'off', structural_tag_scope: str = 'auto', structural_tag_schema: str = 'auto') -> PreprocessResult
 ```
 
-[`components/src/dynamo/frontend/prepost.py#L529`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L529)
+[`components/src/dynamo/frontend/prepost.py#L551`](https://github.com/ai-dynamo/dynamo/blob/main/components/src/dynamo/frontend/prepost.py#L551)
 </Accordion>
 
 <Accordion id="api-dynamo-frontend-sglang-prepost-preprocess-chat-request" title="preprocess_chat_request (function)">

tests/frontend/test_prepost_tokenizer_pool.py

@@ -0,0 +1,138 @@
+#  SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#  SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for the async tokenizer executor pool in dynamo.frontend.prepost."""
+
+# mypy seems to be running both sides of the HAS_VLLM if statement
+# mypy: ignore-errors
+
+import asyncio
+import gc
+import weakref
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+
+import pytest
+
+from .common import check_module_available
+
+HAS_VLLM = check_module_available("vllm.entrypoints.openai.chat_completion.protocol")
+if HAS_VLLM:
+    from dynamo.frontend import prepost
+
+
+pytestmark = [
+    pytest.mark.vllm,
+    pytest.mark.core,
+    pytest.mark.gpu_0,  # "Hardware"
+    pytest.mark.pre_merge,  # "Lifecyle"
+    pytest.mark.unit,  # "Test Type"
+    pytest.mark.skipif(not HAS_VLLM, reason="requires vllm"),
+]
+
+
+class _DummyTokenizer:
+    def __init__(self):
+        self.prompts = []
+
+    def __call__(self, prompt, **kwargs):
+        self.prompts.append(prompt)
+        return SimpleNamespace(input_ids=[1, 2, 3])
+
+
+class _SlotsTokenizer:
+    # No __weakref__: weakref.ref() raises TypeError.
+    __slots__ = ()
+
+    def __call__(self, prompt, **kwargs):
+        return SimpleNamespace(input_ids=[4, 5, 6])
+
+
+class _RecordingExecutor(ThreadPoolExecutor):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.shutdown_calls = 0
+
+    def shutdown(self, wait=True, *, cancel_futures=False):
+        self.shutdown_calls += 1
+        super().shutdown(wait, cancel_futures=cancel_futures)
+
+
+@pytest.fixture(autouse=True)
+def _clean_pools(monkeypatch):
+    monkeypatch.setattr(prepost, "ThreadPoolExecutor", _RecordingExecutor)
+    prepost._ASYNC_TOKENIZER_EXECUTORS.clear()
+    prepost._STRONG_ASYNC_TOKENIZER_EXECUTORS.clear()
+    yield
+    for executor in list(prepost._ASYNC_TOKENIZER_EXECUTORS.values()):
+        executor.shutdown()
+    for _, executor in list(prepost._STRONG_ASYNC_TOKENIZER_EXECUTORS.values()):
+        executor.shutdown()
+    prepost._ASYNC_TOKENIZER_EXECUTORS.clear()
+    prepost._STRONG_ASYNC_TOKENIZER_EXECUTORS.clear()
+
+
+async def _call_tokenizer(tokenizer, prompt):
+    # make_async returns a Future, so the call must be awaited inside a
+    # running loop rather than passed to asyncio.run directly.
+    return await prepost._get_async_tokenizer(tokenizer)(prompt)
+
+
+def test_same_tokenizer_reuses_one_executor():
+    tokenizer = _DummyTokenizer()
+    prepost._get_async_tokenizer(tokenizer)
+    prepost._get_async_tokenizer(tokenizer)
+
+    assert len(prepost._ASYNC_TOKENIZER_EXECUTORS) == 1
+    executor = prepost._ASYNC_TOKENIZER_EXECUTORS[tokenizer]
+    assert isinstance(executor, _RecordingExecutor)
+
+    result = asyncio.run(_call_tokenizer(tokenizer, "hello"))
+    assert result.input_ids == [1, 2, 3]
+    assert tokenizer.prompts == ["hello"]
+
+
+def test_executor_evicted_and_shut_down_when_tokenizer_collected():
+    tokenizer = _DummyTokenizer()
+    prepost._get_async_tokenizer(tokenizer)
+    executor = prepost._ASYNC_TOKENIZER_EXECUTORS[tokenizer]
+    tokenizer_ref = weakref.ref(tokenizer)
+
+    # No cached wrapper may survive this del: it would pin the tokenizer.
+    del tokenizer
+    gc.collect()
+
+    assert tokenizer_ref() is None
+    assert len(prepost._ASYNC_TOKENIZER_EXECUTORS) == 0
+    assert executor.shutdown_calls == 1
+
+
+def test_new_tokenizer_after_eviction_gets_fresh_executor():
+    first = _DummyTokenizer()
+    prepost._get_async_tokenizer(first)
+    first_executor = prepost._ASYNC_TOKENIZER_EXECUTORS[first]
+    del first
+    gc.collect()
+    assert len(prepost._ASYNC_TOKENIZER_EXECUTORS) == 0
+
+    second = _DummyTokenizer()
+    prepost._get_async_tokenizer(second)
+    assert len(prepost._ASYNC_TOKENIZER_EXECUTORS) == 1
+    assert prepost._ASYNC_TOKENIZER_EXECUTORS[second] is not first_executor
+
+
+def test_non_weakrefable_tokenizer_falls_back_to_strong_pool():
+    tokenizer = _SlotsTokenizer()
+    result = asyncio.run(_call_tokenizer(tokenizer, "hi"))
+    assert result.input_ids == [4, 5, 6]
+
+    assert len(prepost._ASYNC_TOKENIZER_EXECUTORS) == 0
+    assert len(prepost._STRONG_ASYNC_TOKENIZER_EXECUTORS) == 1
+
+    # The fallback entry must retain the tokenizer itself so its id() cannot
+    # be recycled by a different tokenizer while the entry lives.
+    retained_tokenizer, _ = prepost._STRONG_ASYNC_TOKENIZER_EXECUTORS[id(tokenizer)]
+    assert retained_tokenizer is tokenizer
+
+    prepost._get_async_tokenizer(tokenizer)
+    assert len(prepost._STRONG_ASYNC_TOKENIZER_EXECUTORS) == 1