feat(decisions): initialize native read workers

components/src/dynamo/vllm/handlers.py

@@ -102,6 +102,14 @@
 from .cache_info import get_configured_kv_event_block_size
 from .capacity import publish_vllm_token_budget
 from .constants import MX_LOAD_FORMATS, DisaggregationMode, EmbeddingTransferMode
+from .decision_reads import (
+    DIFFUSION_PROFILE,
+    LIQUID_PROFILE,
+    PINNED_VLLM_COMMIT,
+    ReadExecutionAdapter,
+    ReadRuntimeError,
+    ReadValidationError,
+)
 from .dp_topology import get_dp_range_for_worker
 from .engine_generate import (
     adapt_engine_generate_request,
@@ -3779,11 +3787,67 @@ def __init__(
             encode_worker_client=encode_worker_client,
         )
         self._first_token_source = first_token_source
+        self.decision_read_adapter: ReadExecutionAdapter | None = None
+        self._decision_read_initialization_attempted = False
+        self._decision_read_initialization_lock = asyncio.Lock()
         self._external_encoder_handoff_consumer: Optional[
             ExternalEncoderHandoffConsumer
         ] = None
 
+    @property
+    def decision_read_profile(self) -> str | None:
+        adapter = self.decision_read_adapter
+        return adapter.profile if adapter is not None else None
+
+    async def initialize_decision_read(self) -> None:
+        async with self._decision_read_initialization_lock:
+            if self._decision_read_initialization_attempted:
+                return
+            self._decision_read_initialization_attempted = True
+            architecture = getattr(self.model_config, "architecture", None)
+            profile = {
+                "Lfm2VlForConditionalGeneration": LIQUID_PROFILE,
+                "DiffusionGemmaForBlockDiffusion": DIFFUSION_PROFILE,
+            }.get(architecture)
+            if profile is None:
+                return
+            try:
+                config = self.config
+                if (
+                    config.disaggregation_mode != DisaggregationMode.AGGREGATED
+                    or config.enable_rl
+                    or config.route_to_encoder
+                    or config.custom_encoder_class is not None
+                    or config.enable_multimodal
+                    or config.realtime
+                ):
+                    raise ReadRuntimeError(
+                        "worker configuration does not support native decisions"
+                    )
+                self.decision_read_adapter = await ReadExecutionAdapter.create(
+                    self.engine_client,
+                    profile,
+                    source_commit=PINNED_VLLM_COMMIT,
+                    abort_monitor=self._abort_monitor,
+                )
+            except ReadRuntimeError as cause:
+                logger.warning("vLLM native generative decisions disabled: %s", cause)
+
     async def generate(self, request, context):
+        if "decision_read" in request:
+            adapter = self.decision_read_adapter
+            if adapter is None:
+                raise ReadValidationError(
+                    "native decision read is not available on this worker"
+                )
+            yield await adapter.execute(
+                request,
+                context,
+                request_id=context.id(),
+                cancelled=lambda: context.is_stopped() or context.is_killed(),
+                trace_headers=context.trace_headers(),
+            )
+            return
         normalize_request_token_ids(request)
         settings = _backend_metadata_upload_settings(request)
         if settings is not None and "kind" in settings and self.use_vllm_tokenizer:

components/src/dynamo/vllm/health_check.py

@@ -61,8 +61,11 @@ def _get_bos_token_id_from_engine(engine_client) -> int:
 def _make_default_payload(
     engine_client: Optional["AsyncLLM"], use_text_input: bool
 ) -> dict:
+    # Older optional runtimes may not expose diffusion model metadata.
+    model_config = getattr(engine_client, "model_config", None)
+    is_diffusion = getattr(model_config, "is_diffusion", False) is True
     sampling_options = {
-        "temperature": 0.0,
+        "temperature": 1.0 if is_diffusion else 0.0,
     }
 
     stop_conditions = {

components/src/dynamo/vllm/main.py

@@ -882,6 +882,7 @@ async def register_vllm_model(
     worker_type: WorkerType,
     needs: list[list[WorkerType]] | None = None,
     decision_head_profile: str | None = None,
+    decision_read_profile: str | None = None,
 ) -> None:
     """
     Helper function to register a vLLM model with runtime configuration.
@@ -903,6 +904,7 @@ async def register_vllm_model(
         needs: Peer worker types required to serve traffic, in DNF form
             (list of alternative AND-sets).
         decision_head_profile: Native head profile initialized by the pooling handler.
+        decision_read_profile: Native generative profile initialized by the decode handler.
     """
     runtime_config = ModelRuntimeConfig()
     publish_vllm_structural_tag_reasoning_policy(runtime_config, vllm_config)

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

@@ -0,0 +1,144 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""CPU readiness and dispatch contracts for model-native generative reads."""
+
+import asyncio
+from argparse import Namespace
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from dynamo.common.constants import DisaggregationMode
+from dynamo.vllm import decision_reads as reads
+from dynamo.vllm.args import Config
+from dynamo.vllm.handlers import BaseWorkerHandler, DecodeWorkerHandler
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.pre_merge,
+    pytest.mark.vllm,
+    pytest.mark.core,
+    pytest.mark.gpu_0,
+]
+
+
+@pytest.fixture
+def handler(monkeypatch):
+    def base_init(self, runtime, config, engine, default_sampling_params, **kwargs):
+        self.config = config
+        self.engine_client = engine
+        self.model_config = kwargs["model_config"]
+
+    monkeypatch.setattr(BaseWorkerHandler, "__init__", base_init)
+    config = Config.from_cli_args(
+        Namespace(
+            disaggregation_mode=DisaggregationMode.AGGREGATED,
+            enable_rl=False,
+            route_to_encoder=False,
+            custom_encoder_class=None,
+            enable_multimodal=False,
+            realtime=False,
+        )
+    )
+    model = SimpleNamespace(architecture="Lfm2VlForConditionalGeneration")
+    return DecodeWorkerHandler(None, config, object(), None, model_config=model)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("architecture", "profile"),
+    [
+        ("Lfm2VlForConditionalGeneration", reads.LIQUID_PROFILE),
+        ("DiffusionGemmaForBlockDiffusion", reads.DIFFUSION_PROFILE),
+    ],
+)
+async def test_native_read_ready_only_after_successful_initialization(
+    monkeypatch, handler, architecture, profile
+):
+    handler.model_config.architecture = architecture
+    adapter = SimpleNamespace(profile=profile)
+    create = AsyncMock(return_value=adapter)
+    monkeypatch.setattr(reads.ReadExecutionAdapter, "create", create)
+    assert handler.decision_read_profile is None
+    await asyncio.gather(
+        handler.initialize_decision_read(), handler.initialize_decision_read()
+    )
+    assert handler.decision_read_profile == profile
+    create.assert_awaited_once_with(
+        handler.engine_client,
+        profile,
+        source_commit=reads.PINNED_VLLM_COMMIT,
+        abort_monitor=handler._abort_monitor,
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("disaggregation_mode", DisaggregationMode.PREFILL),
+        ("enable_rl", True),
+        ("route_to_encoder", True),
+        ("custom_encoder_class", "encoder.Custom"),
+        ("enable_multimodal", True),
+        ("realtime", True),
+    ],
+)
+@pytest.mark.parametrize(
+    "architecture",
+    ["Lfm2VlForConditionalGeneration", "DiffusionGemmaForBlockDiffusion"],
+)
+async def test_native_read_unsupported_worker_does_not_advertise(
+    monkeypatch, handler, field, value, architecture
+):
+    handler.model_config.architecture = architecture
+    setattr(handler.config, field, value)
+    create = AsyncMock()
+    monkeypatch.setattr(reads.ReadExecutionAdapter, "create", create)
+    await handler.initialize_decision_read()
+    assert handler.decision_read_profile is None
+    create.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_native_read_unknown_or_unavailable_profile_keeps_normal_generation(
+    monkeypatch, handler
+):
+    create = AsyncMock(side_effect=reads.ReadRuntimeError("unsupported native runtime"))
+    monkeypatch.setattr(reads.ReadExecutionAdapter, "create", create)
+    await handler.initialize_decision_read()
+    assert handler.decision_read_profile is None
+    create.assert_awaited_once()
+    handler._decision_read_initialization_attempted = False
+    handler.model_config.architecture = "Qwen3ForCausalLM"
+    await handler.initialize_decision_read()
+    assert create.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_native_read_private_dispatch_bypasses_token_pipeline(handler):
+    context = SimpleNamespace(
+        id=lambda: "native",
+        is_stopped=lambda: False,
+        is_killed=lambda: False,
+        trace_headers=lambda: {"traceparent": "test"},
+    )
+    wire = {"decision_read": {"protocol_version": 1}}
+    execute = AsyncMock(return_value={"profile": reads.LIQUID_PROFILE})
+    handler.decision_read_adapter = SimpleNamespace(
+        profile=reads.LIQUID_PROFILE, execute=execute
+    )
+    results = [result async for result in handler.generate(wire, context)]
+    assert results == [{"profile": reads.LIQUID_PROFILE}]
+    execute.assert_awaited_once()
+    assert execute.call_args.args == (wire, context)
+    assert execute.call_args.kwargs["request_id"] == "native"
+    assert execute.call_args.kwargs["cancelled"]() is False
+
+
+@pytest.mark.asyncio
+async def test_unready_native_read_is_error_not_generation(handler):
+    with pytest.raises(reads.ReadValidationError, match="not available"):
+        _ = [result async for result in handler.generate({"decision_read": {}}, Mock())]

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

@@ -0,0 +1,112 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Pinned native overloads retain HTTP 503 without exposing engine messages."""
+
+import asyncio
+from importlib import import_module
+from unittest.mock import Mock
+
+import pytest
+
+from dynamo.vllm import decision_heads as heads
+from dynamo.vllm.tests.test_vllm_decision_heads import Engine as HeadEngine
+from dynamo.vllm.tests.test_vllm_decision_heads import NativeApi as HeadNativeApi
+from dynamo.vllm.tests.test_vllm_decision_heads import adapter as head_adapter
+from dynamo.vllm.tests.test_vllm_decision_heads import question as head_question
+from dynamo.vllm.tests.test_vllm_decision_heads import wire as head_wire
+from dynamo.vllm.tests.test_vllm_decision_reads import adapter as read_adapter
+from dynamo.vllm.tests.test_vllm_decision_reads import question as read_question
+from dynamo.vllm.tests.test_vllm_decision_reads import wire as read_wire
+from dynamo.vllm.tests.test_vllm_decision_reads_native import Engine
+
+pytest_plugins = ["dynamo.vllm.tests.test_vllm_decision_reads_native"]
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.vllm,
+    pytest.mark.core,
+    pytest.mark.gpu_0,
+    pytest.mark.pre_merge,
+    pytest.mark.timeout(30),
+]
+
+
+@pytest.fixture
+def native_errors():
+    errors = pytest.importorskip("vllm.exceptions")
+    if not all(
+        hasattr(errors, name) for name in ("QueueOverflowError", "MaxQueuedTokensError")
+    ):
+        pytest.skip("Exact native overload error APIs are unavailable")
+    return (errors.QueueOverflowError, errors.MaxQueuedTokensError), import_module(
+        "dynamo.llm.exceptions"
+    ).HttpError
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["head", "read"])
+@pytest.mark.parametrize("phase", ["admission", "engine"])
+@pytest.mark.parametrize("error_index", [0, 1])
+async def test_actual_native_overload_is_a_bounded_503_and_drains_children(
+    native_reads, native_errors, monkeypatch, kind, phase, error_index
+):
+    error_types, http_error = native_errors
+    if kind == "head":
+        api, engine = HeadNativeApi(), HeadEngine()
+        monkeypatch.setattr(heads, "_load_native", lambda _: api)
+        instance = await head_adapter(engine)
+        request = head_wire(questions=[head_question(), head_question(1)])
+        method = "encode"
+    else:
+        api, engine = native_reads, Engine()
+        instance = await read_adapter(engine)
+        request = read_wire(questions=[read_question(), read_question(1)])
+        method = "generate"
+    api.overload_errors, api.http_error = error_types, http_error
+    error = error_types[error_index]()
+    error.message = "private engine diagnostic must not escape"
+    if phase == "admission":
+        engine.check_admission = Mock(side_effect=error)
+    else:
+        engine.block = asyncio.Event()
+        original = getattr(engine, method)
+
+        async def fail_one_child(*args, **kwargs):
+            request_id = kwargs["request_id"] if "request_id" in kwargs else args[2]
+            if request_id.endswith("-0"):
+                await engine.started.wait()
+                raise error
+            async for output in original(*args, **kwargs):
+                yield output
+
+        monkeypatch.setattr(engine, method, fail_one_child)
+    with pytest.raises(http_error) as caught:
+        await instance.execute(
+            request, [], request_id="overloaded", cancelled=lambda: False
+        )
+    assert caught.value.code == 503
+    assert caught.value.message == "Native decision engine is temporarily unavailable"
+    assert engine.active == 0
+    if phase == "admission":
+        assert engine.calls == []
+    else:
+        assert len(engine.aborted) == 2
+
+
+@pytest.mark.asyncio
+async def test_untrusted_status_like_errors_are_not_reclassified(
+    native_reads, native_errors
+):
+    class UnknownError(RuntimeError):
+        http_status = 503
+        message = "unknown"
+
+    error_types, http_error = native_errors
+    native_reads.overload_errors, native_reads.http_error = error_types, http_error
+    engine = Engine()
+    engine.check_admission = Mock(side_effect=UnknownError("original failure"))
+    instance = await read_adapter(engine)
+    with pytest.raises(UnknownError, match="original failure"):
+        await instance.execute(
+            read_wire(), [], request_id="unknown", cancelled=lambda: False
+        )