feat(multimodal): eager handler integration + serial AsyncVisionEncoder (no batcher) + e2e

PR2 of the custom vision-encoder series, stacked on PR1. Wires the contract into
the aggregated dynamo.vllm worker and proves the mixed-embeds splice path end to
end with a trivial backend and a DIRECT serial call — no micro-batcher yet.

- AsyncVisionEncoder (serial glue): off-thread preprocess pool + the A5
  all-or-nothing preprocess barrier + a single actor thread running
  build / forward_batch / close. Forwards are serialized (no cross-request
  coalescing) — that lands in PR3. The public surface (load / encode /
  get_image_placeholder_token_id / shutdown) is the stable handler boundary.
- handlers.py: resolve a VisionEncoderBackend subclass from --custom-encoder-class,
  wrap it in AsyncVisionEncoder, assemble the mixed EmbedsPrompt; requires
  --enable-prompt-embeds; image-only with a text-only fallback.
- Examples: QwenVisionEncoderBackend base + HitchhikersVisionEncoder (answers "42").
- agg_custom.sh launcher + a minimal qwen_vl.jinja template.
- e2e: the agg_custom topology profile + the "42" semantic payload.

Tests: serial-glue unit tests (encode, A5 barrier, single actor thread, NO
coalescing, fail-fast load + thread reaping, shutdown). The integrated "42" e2e is
validated on the series' integration branch.

Note: hand-set GPU utilization (memory reservation is a later milestone). Based on
the series' validated merge-base; rebase onto main before merge.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: furionw

components/src/dynamo/vllm/handlers.py

@@ -0,0 +1,159 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Serial async glue (L3) between the worker's event loop and a ``VisionEncoderBackend``.
+
+This is the **eager-milestone** glue: it proves the splice path end to end with a
+**direct call — no micro-batcher**. Per request it preprocesses the images off the
+event loop, enforces request-level atomicity, and runs the author's
+``forward_batch`` on a single dedicated **actor thread**, serialized — there is no
+cross-request coalescing. A follow-up swaps this body for a
+``ThreadedMicroBatcher`` (cross-request batching); the public surface
+(``load`` / ``encode`` / ``get_image_placeholder_token_id`` / ``shutdown``) is
+identical, so the worker integration does not change.
+
+Why a single actor thread (not ``asyncio.to_thread``): build and every
+``forward_batch`` run on the **same** thread, so an author that captures a CUDA
+graph in ``build`` can replay it from ``forward_batch`` — the affinity the batched
+version also guarantees. ``max_workers=1`` serializes forwards (FIFO), so
+concurrent ``encode`` calls run one forward at a time without interleaving.
+
+Request-level atomicity (design A5): a gather-barrier sits between preprocess and
+the forward — ``encode`` waits for *every* image's preprocess to settle and runs
+the forward only if **all** succeed; on any failure it does no GPU work and raises
+the request-level error, so a text-only LM never sees a partial result.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from concurrent.futures import ThreadPoolExecutor
+from typing import Generic, List, Optional
+
+import torch
+
+from dynamo.vllm.multimodal_utils.vision_encoder_backend import (
+    ItemT,
+    Preprocessed,
+    RawT,
+    VisionEncoderBackend,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class AsyncVisionEncoder(Generic[RawT, ItemT]):
+    """Drive a ``VisionEncoderBackend`` from the async request path, serially.
+
+    The worker calls ``load`` once at startup and ``await``s ``encode`` per
+    request; ``shutdown`` on teardown. All model knowledge lives in ``backend``;
+    this class owns the preprocess pool, the A5 barrier, and the single actor
+    thread that runs ``build`` / ``forward_batch`` / ``close``.
+
+    Args:
+        backend: The author-written ``VisionEncoderBackend``.
+        preprocess_concurrency: Worker threads for off-loop ``preprocess``.
+        name: Base name for the actor thread / preprocess pool.
+    """
+
+    def __init__(
+        self,
+        backend: VisionEncoderBackend[RawT, ItemT],
+        *,
+        preprocess_concurrency: int = 4,
+        name: str = "vision-encoder",
+    ) -> None:
+        if preprocess_concurrency < 1:
+            raise ValueError("preprocess_concurrency must be >= 1")
+        self._backend = backend
+        self._preprocess_concurrency = preprocess_concurrency
+        self._name = name
+        self._actor: Optional[ThreadPoolExecutor] = None  # build + every forward
+        self._pool: Optional[ThreadPoolExecutor] = None  # off-loop preprocess
+
+    # ---- lifecycle ---------------------------------------------------------
+
+    def load(self, model_id: str, device: str) -> None:
+        """Run ``backend.build`` on the actor thread and fail fast.
+
+        Re-raises any build error, then ``validate``s the placeholder id so a
+        misconfigured encoder errors at startup instead of on the first request.
+        Single-shot: a second ``load()`` raises rather than orphaning the first
+        actor thread and model.
+        """
+        if self._actor is not None or self._pool is not None:
+            raise RuntimeError("AsyncVisionEncoder.load() called twice")
+        try:
+            # One actor thread so build + every forward share a thread; a single
+            # worker also serializes forwards (FIFO) — no cross-request batching.
+            self._actor = ThreadPoolExecutor(
+                max_workers=1, thread_name_prefix=f"{self._name}-actor"
+            )
+            self._pool = ThreadPoolExecutor(
+                max_workers=self._preprocess_concurrency,
+                thread_name_prefix=f"{self._name}-pre",
+            )
+            self._actor.submit(self._backend.build, model_id, device).result()
+            self.validate()
+        except BaseException:
+            self.shutdown()
+            raise
+
+    def validate(self) -> None:
+        """Fail-fast checks run by ``load`` after ``build`` (resolve the
+        placeholder id once)."""
+        self._backend.get_image_placeholder_token_id()
+
+    def get_image_placeholder_token_id(self) -> int:
+        """The token id marking image positions, from the backend."""
+        return self._backend.get_image_placeholder_token_id()
+
+    # ---- request path ------------------------------------------------------
+
+    async def encode(self, raws: List[RawT]) -> List[torch.Tensor]:
+        """Preprocess (off-loop, A5 barrier) then run a single serial forward.
+
+        Returns one ``(n_visual_tokens, lm_hidden_dim)`` tensor per raw input, in
+        order. Raises if any image's preprocess fails (no GPU work) or if the
+        forward fails.
+        """
+        if self._actor is None or self._pool is None:
+            raise RuntimeError("AsyncVisionEncoder.encode() called before load()")
+        if not raws:
+            return []
+        loop = asyncio.get_running_loop()
+        # A5 barrier: preprocess all images concurrently, wait for EVERY one to
+        # settle, and run the forward only if all succeeded. return_exceptions=True
+        # makes the gather a true barrier (no short-circuit), so a failed sibling
+        # cannot leave a half-run request.
+        tasks = [
+            loop.run_in_executor(self._pool, self._backend.preprocess, raw)
+            for raw in raws
+        ]
+        settled = await asyncio.gather(*tasks, return_exceptions=True)
+        errors = [r for r in settled if isinstance(r, BaseException)]
+        if errors:
+            raise errors[0]
+        preprocessed: List[Preprocessed] = list(settled)  # type: ignore[arg-type]
+        items = [p.item for p in preprocessed]
+        # Direct, serialized forward on the actor thread (eager; target_bucket
+        # defaults to None — there is no graph ladder in this milestone).
+        return await loop.run_in_executor(
+            self._actor, self._backend.forward_batch, items
+        )
+
+    def shutdown(self) -> None:
+        """Run ``backend.close`` on the actor thread, then stop both pools. Safe
+        before ``load`` and idempotent."""
+        if self._actor is not None:
+            try:
+                self._actor.submit(self._backend.close).result(timeout=10)
+            except BaseException:  # noqa: BLE001 — teardown best-effort
+                logger.exception(
+                    "AsyncVisionEncoder(%s): backend.close raised during teardown",
+                    self._name,
+                )
+            self._actor.shutdown(wait=False)
+        if self._pool is not None:
+            self._pool.shutdown(wait=False)

components/src/dynamo/vllm/multimodal_utils/async_vision_encoder.py

@@ -0,0 +1,200 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for the serial dynamo.vllm.multimodal_utils.async_vision_encoder.
+
+Pin the serial-glue contract: build / forward / close run on one actor thread;
+encode returns one tensor per raw; the A5 preprocess barrier fails a request
+atomically (no GPU work) if any image's preprocess fails; concurrent encodes are
+**not** coalesced (one forward_batch call each — that is the batched version's
+job, added later); load fails fast on a build error or unresolvable placeholder
+and reaps its threads.
+"""
+
+import threading
+
+import pytest
+import torch
+
+from dynamo.vllm.multimodal_utils.async_vision_encoder import AsyncVisionEncoder
+from dynamo.vllm.multimodal_utils.vision_encoder_backend import (
+    Preprocessed,
+    VisionEncoderBackend,
+)
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.pre_merge,
+    pytest.mark.vllm,
+    pytest.mark.gpu_0,
+    pytest.mark.multimodal,
+]
+
+
+class _FakeBackend(VisionEncoderBackend):
+    """A CPU-only fake backend; records its threads and forward-call boundaries."""
+
+    max_batch_cost = 8
+    buckets = None
+
+    def __init__(self, *, fail_on=None, placeholder_id=151655):
+        self.fail_on = set(fail_on or ())
+        self._placeholder_id = placeholder_id
+        self.build_thread = None
+        self.close_thread = None
+        self.closed = False
+        self.model_id = None
+        self.forward_threads: list[int] = []
+        self.forward_calls: list[list] = []  # one entry per forward_batch call
+
+    def build(self, model_id, device):
+        self.build_thread = threading.get_ident()
+        self.model_id = model_id
+
+    def preprocess(self, raw):
+        if raw in self.fail_on:
+            raise ValueError(f"bad input {raw}")
+        return Preprocessed(item=raw, cost=1, bucket_key=None)
+
+    def forward_batch(self, items, target_bucket=None):
+        self.forward_threads.append(threading.get_ident())
+        self.forward_calls.append(list(items))
+        return [torch.full((2, 4), float(len(str(it)))) for it in items]
+
+    def get_image_placeholder_token_id(self):
+        return self._placeholder_id
+
+    def close(self):
+        self.close_thread = threading.get_ident()
+        self.closed = True
+
+
+async def test_encode_returns_one_tensor_per_raw():
+    enc = AsyncVisionEncoder(_FakeBackend())
+    enc.load("m", "cpu")
+    try:
+        out = await enc.encode(["a", "bb", "ccc"])
+        assert len(out) == 3
+        assert all(t.shape == (2, 4) for t in out)
+    finally:
+        enc.shutdown()
+
+
+async def test_a5_barrier_fails_atomically_with_no_gpu_work():
+    be = _FakeBackend(fail_on={"bad"})
+    enc = AsyncVisionEncoder(be)
+    enc.load("m", "cpu")
+    try:
+        with pytest.raises(ValueError, match="bad input"):
+            await enc.encode(["good", "bad"])
+        assert be.forward_calls == []  # nothing ran
+    finally:
+        enc.shutdown()
+
+
+async def test_build_and_forward_share_one_non_main_thread():
+    be = _FakeBackend()
+    enc = AsyncVisionEncoder(be)
+    enc.load("m", "cpu")
+    try:
+        await enc.encode(["x"])
+        assert be.build_thread is not None
+        assert set(be.forward_threads) == {be.build_thread}
+        assert be.build_thread != threading.get_ident()
+    finally:
+        enc.shutdown()
+
+
+async def test_concurrent_encodes_are_not_coalesced():
+    """Serial glue: each encode runs its own forward_batch — no cross-request
+    batching (that is the batched version's job, added in a later PR)."""
+    import asyncio
+
+    be = _FakeBackend()
+    enc = AsyncVisionEncoder(be)
+    enc.load("m", "cpu")
+    try:
+        await asyncio.gather(enc.encode(["a"]), enc.encode(["b"]), enc.encode(["c"]))
+        # Three encodes → three separate single-item forward calls.
+        assert len(be.forward_calls) == 3
+        assert all(len(call) == 1 for call in be.forward_calls)
+    finally:
+        enc.shutdown()
+
+
+async def test_load_resolves_placeholder_and_passes_model_id():
+    be = _FakeBackend()
+    enc = AsyncVisionEncoder(be)
+    enc.load("my-model", "cpu")
+    try:
+        assert enc.get_image_placeholder_token_id() == 151655
+        assert be.model_id == "my-model"
+    finally:
+        enc.shutdown()
+
+
+async def test_encode_empty_returns_empty():
+    enc = AsyncVisionEncoder(_FakeBackend())
+    enc.load("m", "cpu")
+    try:
+        assert await enc.encode([]) == []
+    finally:
+        enc.shutdown()
+
+
+async def test_encode_before_load_raises():
+    enc = AsyncVisionEncoder(_FakeBackend())
+    with pytest.raises(RuntimeError, match="before load"):
+        await enc.encode(["a"])
+
+
+def test_load_twice_raises():
+    enc = AsyncVisionEncoder(_FakeBackend())
+    enc.load("m", "cpu")
+    try:
+        with pytest.raises(RuntimeError, match="called twice"):
+            enc.load("m", "cpu")
+    finally:
+        enc.shutdown()
+
+
+def test_shutdown_runs_backend_close_on_actor_thread():
+    be = _FakeBackend()
+    enc = AsyncVisionEncoder(be)
+    enc.load("m", "cpu")
+    enc.shutdown()
+    assert be.closed is True
+    assert be.close_thread == be.build_thread  # close on the actor thread
+
+
+def test_load_fails_fast_on_build_error_and_reaps_threads():
+    class _BadBuild(_FakeBackend):
+        def build(self, model_id, device):
+            raise RuntimeError("build failed")
+
+    enc = AsyncVisionEncoder(_BadBuild())
+    with pytest.raises(RuntimeError, match="build failed"):
+        enc.load("m", "cpu")
+    # Both executors were reaped on the failure path (not leaked).
+    assert enc._actor is not None and enc._actor._shutdown is True
+    assert enc._pool is not None and enc._pool._shutdown is True
+
+
+def test_load_fails_fast_on_unresolvable_placeholder():
+    class _NoPlaceholder(_FakeBackend):
+        def get_image_placeholder_token_id(self):
+            raise ValueError("no placeholder")
+
+    enc = AsyncVisionEncoder(_NoPlaceholder())
+    with pytest.raises(ValueError, match="no placeholder"):
+        enc.load("m", "cpu")
+    assert enc._actor is not None and enc._actor._shutdown is True
+
+
+def test_shutdown_before_load_is_safe():
+    AsyncVisionEncoder(_FakeBackend()).shutdown()  # no-op, no raise
+
+
+def test_preprocess_concurrency_must_be_positive():
+    with pytest.raises(ValueError, match="preprocess_concurrency"):
+        AsyncVisionEncoder(_FakeBackend(), preprocess_concurrency=0)

components/src/dynamo/vllm/tests/multimodal_utils/test_vllm_async_vision_encoder.py

@@ -662,6 +662,9 @@ def _make_decode_handler(
     handler.input_param_manager = MagicMock()
     handler.input_param_manager.get_extra_params.return_value = {}
     handler._deferred_aborts = {}
+    # Real BaseWorkerHandler.__init__ (patched out above) sets this; the
+    # aggregated branch in _generate_token_mode reads it, so mirror the default.
+    handler._custom_encoder = None
     return handler
 
 

components/src/dynamo/vllm/tests/test_vllm_worker_handler.py

@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0