[Performance] Add per-component DLO offload policy

Signed-off-by: hsliuustc0106 <liuhongsheng4@huawei.com>

docs/design/feature/offloader/distributed_layerwise_offload.md

@@ -29,6 +29,7 @@ Legend: ✅ supported, ⚠️ compatibility path or limited validation, ❌ unsu
 | **Other online quantization methods** | ❌ Rejected until runtime packing and scale layouts are validated. | ⚠️ Allowed through the ordinary loader; validation is method-specific. |
 | **Model-level or standard layerwise CPU offload** | ❌ Disabled because DLO takes priority. | ❌ Disabled because DLO takes priority. |
 | **Resident leading layers** | ❌ Rejected. | ✅ Requires eligible resident paths in the model's `OffloadPlan`. |
+| **Auxiliary component policy** | ✅ Encoders/VAEs may remain resident. | ✅ Encoders/VAEs may remain resident. |
 
 See [Parallelism compatibility](#parallelism-compatibility) and
 [Request and loading constraints](#request-and-loading-constraints) for the
@@ -177,24 +178,8 @@ component can be reused by the next. The DiT prefetch path and its two shared
 device buffers are unchanged.
 
 After a component is offloaded, cached-but-unallocated memory is retained only
-while it stays within a budget learned from the running workload and at least
+while it is at most 25% of device capacity and at least 5% of device capacity
-5% of device capacity remains physically free. The budget is the largest
+remains physically free. Crossing either bound releases the allocator cache.
-cached value observed at any component boundary times a margin, so it tracks
-the weight-reuse payload plus the transient footprint whatever the
-resolution. It starts at zero: a cold start pays one release per observation
-step while the peak climbs, after which boundaries retain steadily, and a
-workload change whose cached footprint grows by more than the margin pays one
-further release while the new peak lands. Models whose request geometry opts
-out of the generic dummy run declare a feature-neutral `dummy_run_recipe`
-instead; MiniMax-H3 declares a maximum-duration t2va generation, so the
-engine's startup warmup runs it and, under DLO, the cold-start flushes land
-before real traffic. The budget is capped at a quarter
-of device capacity so a spurious observation cannot monopolize a small
-device, and detaching the last attached stager resets the learned peak so one
-workload never permanently influences another. No user-chosen fraction, no
-multiplier constant, and no warmup flag are needed. Each decision logs the
-staged byte total, observed peak, effective budget and cap,
-cached-but-unallocated bytes, and free memory.
 Missing allocator telemetry also releases it conservatively. Component or
 staging failure forces a release, and an out-of-memory allocation gets one
 retry after release. This is a component-local policy rather than a global
@@ -435,6 +420,22 @@ failure.
 - **HSDP + DP or TP:** rejected independently by the diffusion parallel
   configuration.
 
+### Auxiliary component policy
+
+`dlo_offload_components` maps discovered encoder/VAE paths to booleans. A
+component set to `False` is materialized on the target device once and is not
+entered through the model's manual offload lifecycle. A `True` value permits
+the existing `OffloadPlan.on_demand_component_paths` and encoder block-hook
+behavior; it does not add support where the model declares none. The optional
+`default` key handles unmatched auxiliary components, and an empty map
+preserves the current policy.
+
+The policy is resolved after component discovery and before DLO mutates DiT
+storage. Unknown names, including DiT paths, are rejected. DLO continues to
+own DiT streaming regardless of this map, which avoids an invalid state where
+the loader has skipped ordinary DiT materialization but the backend is asked
+to keep that DiT resident.
+
 ## Request and loading constraints
 
 AllGather DP multi-concurrency requires:
@@ -446,6 +447,15 @@ AllGather DP multi-concurrency requires:
 The no-AllGather path does not impose these DLO-specific synchronized-wave
 requirements.
 
+MiniMax-H3 currently builds one global encoder TP group and broadcasts the
+resulting conditioning from global rank 0. Therefore an encoder TP group that
+spans multiple DP replicas is not valid for independent-prompt AllGather
+waves. Exact-recipe DP2/TP2 and DP2/SP2 experiments with encoder TP4 completed
+the collective schedule, but two distinct prompts with the same seed produced
+byte-identical video and audio in every measured wave. Enabling or disabling
+encoder offload produced paired-identical results, so this is a pre-existing
+model topology limitation rather than component-policy behavior.
+
 Direct checkpoint mmap can back either transfer path. It is currently limited
 to proven TP1, non-HSDP, non-online-quantized layouts. Other layouts use the
 ordinary loader. Per-tensor online FP8 linears can use DLO AllGather after the
@@ -510,6 +520,9 @@ AllGather, because DP replicas did not share checkpoint-backed runtime
 weights. This is a functional and memory smoke test, not a production-quality
 performance or output-quality benchmark.
 
+These B300 DP waves intentionally reused the same prompt and seed, so they do
+not establish independent-prompt isolation across DP replicas.
+
 ### Host-memory measurement
 
 A two-worker MiniMax-H3 FL2VA measurement on one L20X node compared the

docs/user_guide/diffusion/offloader/distributed_layerwise_offload.md

@@ -36,7 +36,8 @@ vllm serve /path/to/model --omni \
 vllm serve /path/to/model --omni \
   --enable-distributed-layerwise-offload \
   --data-parallel-size 4 \
-  --dlo-no-use-allgather
+  --dlo-no-use-allgather \
+  --dlo-offload-components '{"text_encoder": false}'
 
 # Sequence parallel deployment
 vllm serve /path/to/model --omni \
@@ -51,6 +52,7 @@ omni = Omni(
     model="/path/to/model",
     enable_distributed_layerwise_offload=True,
     dlo_use_allgather=True,
+    dlo_offload_components={"text_encoder": False},
 )
 ```
 
@@ -63,20 +65,11 @@ omni = Omni(
 | `--dlo-use-allgather` | Shard host weights and reconstruct with AllGather | `true` |
 | `--dlo-no-use-allgather` | Stream complete rank-local blocks without a DLO weight collective | `false` |
 | `--dlo-resident-layers N` | Keep N leading main-DiT blocks on device; requires no-AllGather and model-declared resident paths | `0` |
+| `--dlo-offload-components JSON` | Permit or disable offload for discovered encoder/VAE components | `{}` |
 | `--host-weight-runtime-mode {disabled,preferred,required}` | HWR policy: no interaction, populate on a miss, or require an exact hit | `disabled` |
 | `--host-weight-runtime-root PATH` | Writable node-local HWR store shared by workers in one storage domain; required for `preferred` and `required` | unset |
 | `--dlo-host-registration-limit-gib N` | Optional per-worker ceiling for registering an HWR mmap; zero adds no ceiling | `0` |
 
-## Startup warmup
-
-Models whose request geometry needs model-specific sampling arguments for a
-warmup generation (currently MiniMax-H3: `t2va`, an explicit aspect ratio)
-run one at engine startup whatever features are enabled. It runs at the
-maximum output duration so it covers every production request; under DLO
-this means the component allocator-cache retention policy learns its
-footprint during startup and no real request pays the cold-start flushes.
-The tradeoff is one maximum-length generation added to boot time.
-
 ## Host-weight loading
 
 The diffusion loader chooses host storage before DLO is enabled. It first
@@ -254,6 +247,25 @@ class MyPipeline(nn.Module):
 When no plan exists, discovery falls back to
 `_layerwise_offload_blocks_attrs` and then heuristic attribute lookup.
 
+## Auxiliary component policy
+
+`dlo_offload_components` follows the same component-map style as diffusion
+quantization. Keys are discovered encoder or VAE paths, values are booleans,
+and `default` is the optional fallback:
+
+```python
+dlo_offload_components = {
+    "text_encoder": False,
+    "default": True,
+}
+```
+
+`True` permits the model's existing DLO behavior for that component. `False`
+keeps the whole component resident, including its child blocks and non-block
+modules. An empty map preserves current behavior. Unknown keys are rejected;
+the map does not control DiT blocks because enabling DLO gives the backend
+ownership of DiT streaming.
+
 ## Data-parallel concurrency
 
 With `data_parallel_size > 1` and AllGather enabled, the scheduler can process
@@ -278,6 +290,13 @@ must enter each collective.
   are validated.
 - Resident leading layers require `--dlo-no-use-allgather` and a model
   `OffloadPlan` that declares eligible `resident_dit_paths`.
+- Component policy applies only to discovered auxiliary encoders and VAEs;
+  it cannot disable DLO for a DiT.
+- MiniMax-H3's encoder TP group is global rather than DP-replica-local. An
+  encoder TP size that spans multiple DP replicas (for example TP4 with
+  DP2/TP2 or DP2/SP2 on four ranks) is not supported for independent-prompt
+  AllGather waves: the current pipeline broadcasts rank 0's conditioning to
+  both replicas. This limitation is independent of the component policy.
 - DP concurrency requires an explicit, identical inference-step count.
 
 Sharing quantized or otherwise unvalidated transformed layouts through a

examples/offline_inference/minimax_h3/dlo_lifecycle.py

@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """Validate MiniMax-H3 request-mode and DLO lifecycle behavior.
 
 The DLO modes accept any positive data-parallel size.  With AllGather enabled,
@@ -85,12 +85,20 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--height", type=int, default=768)
     parser.add_argument("--batch-wait-ms", type=float, default=500.0)
     parser.add_argument("--init-timeout", type=float, default=1800.0)
+    parser.add_argument(
+        "--dlo-offload-components",
+        type=json.loads,
+        default=None,
+        help="Component policy JSON, e.g. '{\"text_encoder\": false}'",
+    )
     parser.add_argument("--output", type=Path)
     return parser.parse_args()
 
 
 def engine_kwargs(args: argparse.Namespace) -> dict[str, Any]:
     is_dlo = args.mode in {"dlo", "dlo-no-allgather"}
+    if args.dlo_offload_components and not is_dlo:
+        raise ValueError("--dlo-offload-components requires a DLO mode")
     common: dict[str, Any] = {
         "model": args.model,
         "trust_remote_code": True,
@@ -114,6 +122,7 @@ def engine_kwargs(args: argparse.Namespace) -> dict[str, Any]:
             enable_distributed_layerwise_offload=True,
             dlo_use_allgather=args.mode == "dlo",
             dlo_resident_layers=0,
+            dlo_offload_components=({} if args.dlo_offload_components is None else args.dlo_offload_components),
         )
     else:
         common.update(

tests/diffusion/models/minimax_h3/test_minimax_h3_dlo_lifecycle.py

@@ -99,3 +99,22 @@ def test_manual_component_offload_failure_forces_retained_cache_release(mocker):
     assert component.load_to_device.call_count == 2
     assert component.offload_to_cpu.call_count == 2
     pipeline._dlo_component_cache.release_if_needed.assert_called_once_with(force=True)
+
+
+def test_resident_component_skips_manual_dlo_lifecycle(mocker):
+    from vllm_omni.diffusion.models.minimax_h3 import MiniMaxH3Pipeline
+
+    pipeline = object.__new__(MiniMaxH3Pipeline)
+    torch.nn.Module.__init__(pipeline)
+    pipeline.od_config = mocker.Mock()
+    pipeline.od_config.enable_layerwise_offload = False
+    pipeline.od_config.enable_distributed_layerwise_offload = True
+    pipeline._model_cpu_offload_modules = []
+    component = mocker.Mock()
+    component._omni_dlo_offload_enabled = False
+
+    with pipeline._component_on_device(component):
+        pass
+
+    component.load_to_device.assert_not_called()
+    component.offload_to_cpu.assert_not_called()

tests/diffusion/offloader/test_component_policy.py

@@ -0,0 +1,142 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+from types import SimpleNamespace
+
+import pytest
+import torch
+from torch import nn
+
+from vllm_omni.diffusion.offloader.base import OffloadConfig, OffloadStrategy
+from vllm_omni.diffusion.offloader.distributed_layerwise_backend import (
+    DistributedLayerwiseOffloadBackend,
+)
+from vllm_omni.diffusion.offloader.module_collector import PipelineModules
+from vllm_omni.diffusion.offloader.offload_plan import OffloadPlan
+
+pytestmark = [pytest.mark.diffusion, pytest.mark.cpu, pytest.mark.core_model]
+
+
+def _od_config(**overrides):
+    values = {
+        "enable_cpu_offload": False,
+        "enable_layerwise_offload": False,
+        "enable_distributed_layerwise_offload": True,
+        "dlo_use_allgather": False,
+        "dlo_resident_layers": 0,
+        "dlo_offload_components": {},
+        "pin_cpu_memory": False,
+        "parallel_config": SimpleNamespace(
+            use_hsdp=False,
+            data_parallel_size=1,
+            sequence_parallel_size=1,
+        ),
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+def _backend(policy: dict[str, bool]) -> DistributedLayerwiseOffloadBackend:
+    backend = object.__new__(DistributedLayerwiseOffloadBackend)
+    backend.config = OffloadConfig(
+        strategy=OffloadStrategy.DISTRIBUTED_LAYER_WISE,
+        pin_cpu_memory=False,
+        dlo_use_allgather=False,
+        dlo_offload_components=policy,
+    )
+    backend.device = torch.device("cpu")
+    return backend
+
+
+def _modules() -> PipelineModules:
+    return PipelineModules(
+        dits=[],
+        dit_names=[],
+        encoders=[nn.Module()],
+        encoder_names=["text_encoder"],
+        vaes=[nn.Module(), nn.Module()],
+        vae_names=["video_vae", "audio_vae"],
+    )
+
+
+def test_component_policy_resolves_exact_then_default():
+    backend = _backend({"text_encoder": False, "default": True})
+
+    assert not backend._component_offload_enabled("text_encoder")
+    assert backend._component_offload_enabled("video_vae")
+
+
+@pytest.mark.parametrize(
+    ("value", "error", "match"),
+    [
+        ([], TypeError, "must be a mapping"),
+        (["text_encoder"], TypeError, "must be a mapping"),
+        (False, TypeError, "must be a mapping"),
+        ({1: False}, TypeError, "keys must be non-empty strings"),
+        ({"text_encoder": 0}, TypeError, "must be a boolean"),
+    ],
+)
+def test_component_policy_rejects_invalid_shapes(value, error, match):
+    with pytest.raises(error, match=match):
+        OffloadConfig.from_od_config(_od_config(dlo_offload_components=value))
+
+
+def test_component_policy_accepts_none_as_default():
+    config = OffloadConfig.from_od_config(_od_config(dlo_offload_components=None))
+
+    assert config.dlo_offload_components == {}
+
+
+def test_component_policy_requires_dlo():
+    with pytest.raises(ValueError, match="requires distributed layerwise offload"):
+        OffloadConfig.from_od_config(
+            _od_config(
+                enable_distributed_layerwise_offload=False,
+                dlo_offload_components={"text_encoder": False},
+            )
+        )
+
+
+def test_component_policy_rejects_unknown_or_dit_names():
+    backend = _backend({"transformer": False})
+
+    with pytest.raises(ValueError, match="unknown auxiliary components.*transformer"):
+        backend._validate_component_offload_policy(_modules())
+
+
+@pytest.mark.parametrize(
+    ("policy", "encoder_stage", "video_stage", "audio_stage", "try_encoder"),
+    [
+        ({}, True, True, True, True),
+        ({"text_encoder": False}, False, True, True, False),
+        ({"default": False, "video_vae": True}, False, True, False, False),
+    ],
+)
+def test_prepare_auxiliary_components_applies_policy(
+    mocker,
+    policy,
+    encoder_stage,
+    video_stage,
+    audio_stage,
+    try_encoder,
+):
+    backend = _backend(policy)
+    backend._try_layerwise_offload_encoder = mocker.Mock()
+    backend._register_on_demand_hook = mocker.Mock()
+    modules = _modules()
+    plan = OffloadPlan(
+        encoder_block_attrs={"text_encoder": ("layers",)},
+        on_demand_component_paths=frozenset({"text_encoder", "video_vae", "audio_vae"}),
+    )
+
+    backend._prepare_auxiliary_components(modules, plan)
+
+    assert backend._try_layerwise_offload_encoder.called is try_encoder
+    assert modules.encoders[0]._omni_dlo_offload_enabled is encoder_stage
+    assert modules.vaes[0]._omni_dlo_offload_enabled is video_stage
+    assert modules.vaes[1]._omni_dlo_offload_enabled is audio_stage
+    assert backend._register_on_demand_hook.call_args_list == [
+        mocker.call(modules.encoders[0], "text_encoder", stage_on_demand=encoder_stage),
+        mocker.call(modules.vaes[0], "video_vae", stage_on_demand=video_stage),
+        mocker.call(modules.vaes[1], "audio_vae", stage_on_demand=audio_stage),
+    ]

tests/diffusion/offloader/test_module_residency.py

@@ -202,96 +202,21 @@ def test_allocator_cache_policy_enforces_cache_and_free_memory_bounds(
     expected_release,
 ):
     empty_cache, _ = patched_runtime
-    telemetry = {"reserved": 20, "allocated": 10, "free": 80}
+    monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: reserved)
-    monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: telemetry["reserved"])
+    monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: allocated)
-    monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: telemetry["allocated"])
     monkeypatch.setattr(
         residency_module.current_omni_platform,
         "get_device_memory",
-        lambda _device: (telemetry["free"], 400),
+        lambda _device: (free, 100),
     )
     retention = BoundedAllocatorCache(torch.device("cpu"))
 
-    # Prime the learned budget: the cold-start check releases (budget 0) and
-    # records cached = 10, so later checks retain at budget 12.5.
-    assert retention.release_if_needed()
-    assert retention.observed_cached_peak == 10
-    empty_cache.reset_mock()
-
-    telemetry.update(reserved=reserved, allocated=allocated, free=free)
-
     released = retention.release_if_needed()
 
     assert released is expected_release
     assert empty_cache.call_count == int(expected_release)
 
 
-def test_attached_stagers_form_a_detach_ledger_that_resets_the_observed_peak(patched_runtime):
-    left, right = _aliased_modules()
-    retention = BoundedAllocatorCache(torch.device("cpu"))
-    assert retention.budget_bytes == 0
-
-    stager = PinnedModuleStager(
-        [left, right],
-        torch.device("cpu"),
-        pin_memory=False,
-        cache_retention=retention,
-    )
-    expected_bytes = 12 * 4 + 8 * 8  # float32 parameter + int64 buffer storages
-    assert stager.staged_bytes == expected_bytes
-    assert retention._staged_bytes == expected_bytes
-
-    # Re-attaching the same cache does not double count.
-    stager.set_cache_retention(retention)
-    assert retention._staged_bytes == expected_bytes
-
-    # A second stager adds its own staged bytes to the ledger.
-    extra = PinnedModuleStager(torch.nn.Linear(2, 2), torch.device("cpu"), pin_memory=False)
-    extra.set_cache_retention(retention)
-    assert retention._staged_bytes == expected_bytes + extra.staged_bytes
-
-    # A learned peak survives a partial detach...
-    retention._observed_cached_peak = 30
-    extra.set_cache_retention(None)
-    assert retention.observed_cached_peak == 30
-    assert retention.budget_bytes == 37
-
-    # ...but the last detach resets it so one workload cannot influence the next.
-    stager.set_cache_retention(None)
-    assert retention.observed_cached_peak == 0
-    assert retention.budget_bytes == 0
-
-
-def test_allocator_cache_budget_is_capped_by_device_capacity(monkeypatch, patched_runtime):
-    empty_cache, _ = patched_runtime
-    telemetry = {"reserved": 40, "allocated": 10, "free": 80}
-    monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: telemetry["reserved"])
-    monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: telemetry["allocated"])
-    monkeypatch.setattr(
-        residency_module.current_omni_platform,
-        # Capacity 100 keeps the 25% cap at 25, below the learned budget of
-        # 30 * 1.25 = 37.5, so the cap must govern.
-        "get_device_memory",
-        lambda _device: (telemetry["free"], 100),
-    )
-    retention = BoundedAllocatorCache(torch.device("cpu"))
-
-    # Cold start: budget 0 releases and records the peak of 30.
-    assert retention.release_if_needed()
-    assert retention.observed_cached_peak == 30
-    empty_cache.reset_mock()
-
-    # The learned budget 37.5 is capped to 25, so cached 30 keeps releasing.
-    assert retention.release_if_needed()
-    assert retention.release_if_needed()
-    empty_cache.reset_mock()
-
-    # Within the cap (cached 20 <= 25) the policy retains.
-    telemetry.update(reserved=30, allocated=10)
-    assert not retention.release_if_needed()
-    assert empty_cache.call_count == 0
-
-
 def test_allocator_cache_releases_when_telemetry_is_unavailable(monkeypatch, patched_runtime):
     empty_cache, _ = patched_runtime
 
@@ -303,29 +228,3 @@ def test_allocator_cache_releases_when_telemetry_is_unavailable(monkeypatch, pat
 
     assert retention.release_if_needed()
     empty_cache.assert_called_once_with()
-
-
-def test_allocator_cache_budget_adapts_to_observed_peak(monkeypatch, patched_runtime):
-    empty_cache, _ = patched_runtime
-    telemetry = {"reserved": 40, "allocated": 10, "free": 80}
-    monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: telemetry["reserved"])
-    monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: telemetry["allocated"])
-    monkeypatch.setattr(
-        residency_module.current_omni_platform,
-        "get_device_memory",
-        # Capacity 400 keeps the 25% cap (100) above the learned budget so
-        # this test isolates the peak-adaptation behavior from the cap.
-        lambda _device: (telemetry["free"], 400),
-    )
-    retention = BoundedAllocatorCache(torch.device("cpu"))
-
-    # Cold-start boundary: budget 0, so the policy releases and records the
-    # observed peak.
-    assert retention.release_if_needed()
-    assert retention.observed_cached_peak == 30
-
-    # The observation raises the budget to 30 * 1.25 = 37, so the same cached
-    # level is retained on every later boundary without another release.
-    assert not retention.release_if_needed()
-    assert not retention.release_if_needed()
-    assert empty_cache.call_count == 1

tests/diffusion/test_diffusion_engine_dummy_run.py

@@ -70,161 +70,6 @@ def test_dummy_run_image_count_resolves_hunyuan_architecture_alias() -> None:
     assert io_support.get_dummy_run_num_image_inputs("unknown") == 1
 
 
-def test_dummy_run_recipe_returns_none_without_declaration(monkeypatch: pytest.MonkeyPatch) -> None:
-    class PlainModel:
-        pass
-
-    monkeypatch.setattr(
-        io_support.DiffusionModelRegistry,
-        "_try_load_model_cls",
-        lambda model_class_name: PlainModel,
-    )
-
-    assert io_support.get_dummy_run_recipe("plain") is None
-
-
-def test_dummy_run_recipe_returns_copy_of_declaration(monkeypatch: pytest.MonkeyPatch) -> None:
-    declared = {"task": "t2va", "duration": 4.0}
-
-    class RecipeModel:
-        dummy_run_recipe = declared
-
-    monkeypatch.setattr(
-        io_support.DiffusionModelRegistry,
-        "_try_load_model_cls",
-        lambda model_class_name: RecipeModel,
-    )
-
-    recipe = io_support.get_dummy_run_recipe("recipe")
-    assert recipe == declared
-    assert recipe is not declared
-
-
-def test_minimax_h3_declares_a_valid_dlo_warmup_recipe() -> None:
-    from vllm_omni.diffusion.models.minimax_h3.pipeline_minimax_h3 import (
-        MINIMAX_H3_MAX_OUTPUT_SECONDS,
-        MiniMaxH3Pipeline,
-    )
-
-    recipe = MiniMaxH3Pipeline.dummy_run_recipe
-    assert recipe is not None
-    assert recipe["task"] == "t2va"
-    assert recipe["aspect_ratio"] == "16:9"
-    # Maximum duration makes the learned peak dominate every production
-    # request at the fixed short edge, so no real request re-learns.
-    assert float(recipe["duration"]) == MINIMAX_H3_MAX_OUTPUT_SECONDS
-    assert MiniMaxH3Pipeline.dummy_run_num_frames == 0  # generic warmup stays opted out
-
-
-def _dlo_recipe_engine(enable_dlo: bool) -> DiffusionEngine:
-    engine = object.__new__(DiffusionEngine)
-    engine.od_config = SimpleNamespace(
-        model_class_name="RecipeModel",
-        enable_distributed_layerwise_offload=enable_dlo,
-    )
-    return engine
-
-
-def _opt_out_generic_warmup(
-    monkeypatch: pytest.MonkeyPatch,
-    recipe: dict | None,
-    *,
-    supports_image_input: bool = False,
-) -> None:
-    """Make the generic dummy run opt out and stub the declared recipe."""
-
-    monkeypatch.setattr(
-        "vllm_omni.diffusion.diffusion_engine.supports_multimodal_input",
-        lambda od_config: (supports_image_input, False),
-    )
-    monkeypatch.setattr(
-        "vllm_omni.diffusion.diffusion_engine.get_dummy_run_num_frames",
-        lambda model_class_name, supports_audio_input: 0,
-    )
-    monkeypatch.setattr(
-        "vllm_omni.diffusion.diffusion_engine.get_dummy_run_recipe",
-        lambda model_class_name: recipe,
-    )
-
-
-def test_make_dummy_request_applies_recipe_without_dlo(monkeypatch: pytest.MonkeyPatch) -> None:
-    # The recipe is feature-neutral: a model whose geometry needs it warms up
-    # whatever features are enabled, not only under DLO.
-    _opt_out_generic_warmup(monkeypatch, {"task": "t2va", "duration": 4.0, "aspect_ratio": "16:9"})
-
-    request = _dlo_recipe_engine(enable_dlo=False)._make_dummy_request(height=512, width=512, guidance_scale=0.0)
-
-    assert request is not None
-    assert request.sampling_params.extra_args["task"] == "t2va"
-
-
-def test_make_dummy_request_builds_recipe_request_when_generic_opted_out(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    _opt_out_generic_warmup(
-        monkeypatch,
-        {
-            "task": "t2va",
-            "duration": 4.0,
-            "aspect_ratio": "16:9",
-            "num_inference_steps": 2,
-            "guidance_scale": 1.0,
-        },
-    )
-
-    request = _dlo_recipe_engine(enable_dlo=True)._make_dummy_request(height=512, width=512, guidance_scale=0.0)
-
-    assert request is not None
-    assert request.is_dummy_run()
-    assert request.sampling_params.num_inference_steps == 2
-    assert request.sampling_params.guidance_scale == 1.0
-    extra = request.sampling_params.extra_args
-    assert extra["task"] == "t2va"
-    assert extra["duration"] == 4.0
-    assert extra["aspect_ratio"] == "16:9"
-    assert extra["cfg_text_scale"] == 1.0
-
-
-def test_make_dummy_request_recipe_stays_text_only_with_image_capable_model(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    # H3 advertises image input for fl2va, but the warmup recipe names t2va,
-    # which rejects image conditions; the recipe request must not inherit the
-    # generic dummy run's synthetic multimodal data.
-    _opt_out_generic_warmup(
-        monkeypatch,
-        {"task": "t2va", "duration": 4.0, "aspect_ratio": "16:9"},
-        supports_image_input=True,
-    )
-
-    request = _dlo_recipe_engine(enable_dlo=True)._make_dummy_request(height=512, width=512, guidance_scale=0.0)
-
-    assert request is not None
-    assert not request.prompt.get("multi_modal_data")
-
-
-def test_dummy_run_sends_recipe_request_when_generic_opted_out(monkeypatch: pytest.MonkeyPatch) -> None:
-    _opt_out_generic_warmup(monkeypatch, {"task": "t2va", "duration": 15.0, "aspect_ratio": "16:9"})
-    engine = _dlo_recipe_engine(enable_dlo=True)
-    engine.add_req_and_wait_for_response = Mock(return_value=SimpleNamespace(error=None, request_id="dlo-dummy"))
-
-    engine._dummy_run()
-
-    sent = engine.add_req_and_wait_for_response.call_args.args[0]
-    assert sent.is_dummy_run()
-    assert sent.sampling_params.extra_args["task"] == "t2va"
-
-
-def test_dummy_run_still_skips_without_recipe(monkeypatch: pytest.MonkeyPatch) -> None:
-    _opt_out_generic_warmup(monkeypatch, None)
-    engine = _dlo_recipe_engine(enable_dlo=True)
-    engine.add_req_and_wait_for_response = Mock()
-
-    engine._dummy_run()
-
-    engine.add_req_and_wait_for_response.assert_not_called()
-
-
 def test_dense_mode_does_not_build_kv_profile_request() -> None:
     engine = object.__new__(DiffusionEngine)
     engine.od_config = SimpleNamespace(diffusion_kv_mode=DiffusionKVCacheMode.DENSE_LEGACY)

tests/entrypoints/test_async_omni_diffusion_config.py

@@ -332,7 +332,7 @@ def test_serve_cli_forwards_distilled_lora_to_diffusion_stage():
     ]
 
 
-def test_serve_cli_forwards_distributed_offload_residency():
+def test_serve_cli_forwards_distributed_offload_options():
     """Ensure the two-GPU DLO placement controls reach the diffusion stage."""
     parser = TrackingArgumentParser()
     subparsers = parser.add_subparsers(dest="command")
@@ -347,6 +347,8 @@ def test_serve_cli_forwards_distributed_offload_residency():
             "--dlo-no-use-allgather",
             "--dlo-resident-layers",
             "20",
+            "--dlo-offload-components",
+            '{"text_encoder": false, "default": true}',
         ]
     )
 
@@ -357,9 +359,14 @@ def test_serve_cli_forwards_distributed_offload_residency():
     assert args.enable_distributed_layerwise_offload is True
     assert args.dlo_use_allgather is False
     assert args.dlo_resident_layers == 20
+    assert args.dlo_offload_components == {"text_encoder": False, "default": True}
     assert engine_args["enable_distributed_layerwise_offload"] is True
     assert engine_args["dlo_use_allgather"] is False
     assert engine_args["dlo_resident_layers"] == 20
+    assert engine_args["dlo_offload_components"] == {
+        "text_encoder": False,
+        "default": True,
+    }
 
 
 def test_serve_cli_forwards_hwr_policy_for_no_allgather_dlo():

vllm_omni/config/omni_config.py

@@ -697,6 +697,7 @@ class _DiffusionConfigProjection:
     enable_distributed_layerwise_offload: bool = False
     dlo_use_allgather: bool = True
     dlo_resident_layers: int = Field(default=0, ge=0)
+    dlo_offload_components: dict[str, bool] = Field(default_factory=dict)
     host_weight_runtime_mode: Literal["disabled", "preferred", "required"] = "disabled"
     host_weight_runtime_root: str | None = None
     dlo_host_registration_limit_gib: float = Field(default=0.0, ge=0)

vllm_omni/config/stage_config.py

@@ -389,6 +389,7 @@ class StageDeployConfig:
     enable_distributed_layerwise_offload: bool | None = None
     dlo_use_allgather: bool | None = None
     dlo_resident_layers: int | None = None
+    dlo_offload_components: dict[str, bool] | None = None
     host_weight_runtime_mode: str | None = None
     host_weight_runtime_root: str | None = None
     dlo_host_registration_limit_gib: float | None = None

vllm_omni/diffusion/data.py

@@ -795,6 +795,9 @@ class OmniDiffusionConfig:
     dlo_use_allgather: bool = True
     # Leading main-DiT blocks kept resident by distributed layerwise offload.
     dlo_resident_layers: int = 0
+    # Auxiliary encoder/VAE policy. False keeps the whole component resident;
+    # DiT offload remains owned by DLO.
+    dlo_offload_components: dict[str, bool] = field(default_factory=dict)
     # Final-layout Host Weight Runtime policy. The loader only activates this
     # for eligible no-AllGather DLO; all other configurations preserve their
     # existing loader/storage path.

vllm_omni/diffusion/diffusion_engine.py

@@ -36,7 +36,6 @@ from vllm_omni.diffusion.executor.abstract import DiffusionExecutor
 from vllm_omni.diffusion.io_support import (
     get_dummy_run_num_frames,
     get_dummy_run_num_image_inputs,
-    get_dummy_run_recipe,
     image_color_format,
     supports_audio_output,
     supports_multimodal_input,
@@ -65,9 +64,6 @@ logger = init_logger(__name__)
 
 _ASYNC_OUTPUT_TIMEOUT_ENV = "VLLM_OMNI_ASYNC_OUTPUT_TIMEOUT"
 _ASYNC_OUTPUT_TIMEOUT_DEFAULT = 600.0  # seconds
-# Warmup-recipe keys applied to the dummy request's sampling params; every
-# other recipe entry is forwarded through the request's extra_args.
-_DUMMY_RUN_RECIPE_SAMPLING_KEYS = frozenset({"height", "width", "num_inference_steps", "num_frames", "guidance_scale"})
 
 
 def _async_output_timeout() -> float:
@@ -1021,36 +1017,8 @@ class DiffusionEngine:
     ) -> OmniDiffusionRequest | None:
         """Build a one-step model request for startup profiling or warmup."""
 
-        supports_image_input, supports_audio_input = supports_multimodal_input(self.od_config)
-        num_frames = get_dummy_run_num_frames(self.od_config.model_class_name, supports_audio_input)
-        if num_frames <= 0:
-            # The model opted out of the generic dummy run; fall back to its
-            # declared warmup recipe, which supplies whatever feature-specific
-            # sampling arguments the model requires (under distributed
-            # layerwise offload the generation also primes the component
-            # allocator-cache retention policy ahead of real traffic). The
-            # recipe names its own task, so the request stays text-only even
-            # when the model's other tasks accept image or audio conditions.
-            recipe = get_dummy_run_recipe(self.od_config.model_class_name)
-            if recipe is None:
-                return None
-            extra_args: dict[str, Any] = {"cfg_text_scale": 1.0, "cfg_img_scale": 1.0}
-            extra_args.update({k: v for k, v in recipe.items() if k not in _DUMMY_RUN_RECIPE_SAMPLING_KEYS})
-            return OmniDiffusionRequest(
-                prompt={"prompt": "dummy run"},
-                request_id=DUMMY_DIFFUSION_REQUEST_ID,
-                sampling_params=OmniDiffusionSamplingParams(
-                    height=int(recipe.get("height", height)),
-                    width=int(recipe.get("width", width)),
-                    num_inference_steps=int(recipe.get("num_inference_steps", 1)),
-                    num_frames=int(recipe.get("num_frames", 0)),
-                    guidance_scale=float(recipe.get("guidance_scale", guidance_scale)),
-                    num_outputs_per_prompt=1,
-                    extra_args=extra_args,
-                ),
-            )
-
         prompt: OmniTextPrompt = {"prompt": "dummy run"}
+        supports_image_input, supports_audio_input = supports_multimodal_input(self.od_config)
         if supports_image_input:
             color_format = image_color_format(self.od_config.model_class_name)
             images = [PIL.Image.new(color_format, (width, height)) for _ in range(num_image_inputs)]
@@ -1060,6 +1028,9 @@ class DiffusionEngine:
             audio_sr = 16000
             prompt.setdefault("multi_modal_data", {})["audio"] = np.random.randn(audio_sr * 2).astype(np.float32)
 
+        num_frames = get_dummy_run_num_frames(self.od_config.model_class_name, supports_audio_input)
+        if num_frames <= 0:
+            return None
         return OmniDiffusionRequest(
             prompt=prompt,
             request_id=DUMMY_DIFFUSION_REQUEST_ID,

vllm_omni/diffusion/io_support.py

@@ -4,7 +4,6 @@
 from __future__ import annotations
 
 import inspect
-from typing import Any
 
 from vllm_omni.diffusion.data import OmniDiffusionConfig
 from vllm_omni.diffusion.model_metadata import get_diffusion_model_metadata
@@ -55,22 +54,6 @@ def get_dummy_run_num_frames(model_class_name: str, supports_audio_input: bool)
     return 2 if supports_audio_input or supports_audio_output(model_class_name) else 1
 
 
-def get_dummy_run_recipe(model_class_name: str) -> dict[str, Any] | None:
-    """Get the model-declared startup-warmup recipe, or None.
-
-    Models that opt out of the generic dummy run (``dummy_run_num_frames = 0``)
-    because their request geometry needs model-specific sampling arguments
-    (task, duration, aspect ratio, ...) declare a recipe instead. The recipe
-    is feature-neutral: the engine runs it at startup whatever features are
-    enabled — under distributed layerwise offload the generation also primes
-    the component allocator-cache retention policy's observed-peak budget.
-    """
-
-    model_cls = DiffusionModelRegistry._try_load_model_cls(model_class_name)
-    recipe = getattr(model_cls, "dummy_run_recipe", None) if model_cls is not None else None
-    return dict(recipe) if recipe else None
-
-
 def get_dummy_run_num_image_inputs(model_class_name: str) -> int:
     """Return the maximum advertised image-input count for profiling."""
 

vllm_omni/diffusion/models/minimax_h3/pipeline_minimax_h3.py

@@ -704,20 +704,6 @@ class MiniMaxH3Pipeline(
         "post_decode",
     ]
     dummy_run_num_frames: ClassVar[int] = 0
-    # Startup warmup recipe: the generic dummy run cannot satisfy t2va's
-    # required sampling arguments (explicit aspect ratio, bounded duration).
-    # Duration is the only workload variable at the fixed 768 short edge, so
-    # warming at the maximum covers every production request — under DLO this
-    # makes the learned cache-retention peak dominate, moving all cold-start
-    # flushes ahead of real traffic. The cost is one max-length generation in
-    # boot time whatever features are enabled.
-    dummy_run_recipe: ClassVar[Mapping[str, Any] | None] = {
-        "task": "t2va",
-        "duration": MINIMAX_H3_MAX_OUTPUT_SECONDS,
-        "aspect_ratio": "16:9",
-        "num_inference_steps": 2,
-        "guidance_scale": 1.0,
-    }
     # Only distilled releases pin a schedule, so the default keeps the legacy
     # uniform path available to partially constructed pipelines.
     _base_schedule_by_partition: ClassVar[Mapping[str, DMD2SigmaSchedule | None]] = {}
@@ -1450,7 +1436,7 @@ class MiniMaxH3Pipeline(
             with sequential_offload_component(component):
                 yield
             return
-        staged = self._uses_manual_component_offload()
+        staged = self._uses_manual_component_offload() and getattr(component, "_omni_dlo_offload_enabled", True)
         try:
             if staged:
                 component.load_to_device()

vllm_omni/diffusion/offloader/base.py

@@ -2,7 +2,8 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
+from collections.abc import Mapping
+from dataclasses import dataclass, field
 from enum import Enum
 from typing import Protocol, runtime_checkable
 
@@ -52,6 +53,9 @@ class OffloadConfig:
     # blocks from the loader-selected host backing with H2D only.
     dlo_use_allgather: bool = True
     dlo_resident_layers: int = 0  # leading DiT layers kept on device
+    # Per auxiliary component: True permits model-declared DLO offload;
+    # False keeps the whole component resident. "default" is the fallback.
+    dlo_offload_components: dict[str, bool] = field(default_factory=dict)
     # Optional per-worker ceiling for registering an HWR mmap. Zero means no
     # additional ceiling; pin_cpu_memory controls whether registration is tried.
     dlo_host_registration_limit_gib: float = 0.0
@@ -125,6 +129,25 @@ class OffloadConfig:
         # requirements (concurrent requests, dummy run skip).
         dlo_use_allgather = getattr(od_config, "dlo_use_allgather", True)
         dlo_resident_layers = int(getattr(od_config, "dlo_resident_layers", 0))
+        raw_component_policy = getattr(od_config, "dlo_offload_components", {})
+        if raw_component_policy is None:
+            raw_component_policy = {}
+        if not isinstance(raw_component_policy, Mapping):
+            raise TypeError(
+                "dlo_offload_components must be a mapping of component names to booleans, "
+                f"got {type(raw_component_policy).__name__}"
+            )
+        dlo_offload_components: dict[str, bool] = {}
+        for component, enabled in raw_component_policy.items():
+            if not isinstance(component, str) or not component:
+                raise TypeError("dlo_offload_components keys must be non-empty strings")
+            if not isinstance(enabled, bool):
+                raise TypeError(
+                    f"dlo_offload_components[{component!r}] must be a boolean, got {type(enabled).__name__}"
+                )
+            dlo_offload_components[component] = enabled
+        if dlo_offload_components and not enable_distributed_layerwise_offload:
+            raise ValueError("dlo_offload_components requires distributed layerwise offload")
         dlo_host_registration_limit_gib = validate_dlo_host_registration_options(
             limit_gib=getattr(od_config, "dlo_host_registration_limit_gib", 0.0),
             enable_dlo=enable_distributed_layerwise_offload,
@@ -166,6 +189,7 @@ class OffloadConfig:
             dp_size=dp_size,
             dlo_use_allgather=dlo_use_allgather,
             dlo_resident_layers=dlo_resident_layers,
+            dlo_offload_components=dlo_offload_components,
             dlo_host_registration_limit_gib=dlo_host_registration_limit_gib,
         )
 

vllm_omni/diffusion/offloader/distributed_layerwise_backend.py

@@ -43,7 +43,7 @@ from .host_registration import (
     HostRegistrationError,
     register_host_mappings,
 )
-from .module_collector import ModuleDiscovery
+from .module_collector import ModuleDiscovery, PipelineModules
 from .offload_plan import (
     OffloadPlan,
     get_offload_plan,
@@ -1400,6 +1400,39 @@ class DistributedLayerwiseOffloadBackend(OffloadBackend):
         module.to(self.device)
         logger.info("Moved %s (%s) to GPU (resident)", label, module.__class__.__name__)
 
+    def _component_offload_enabled(self, name: str) -> bool:
+        policy = self.config.dlo_offload_components
+        return policy.get(name, policy.get("default", True))
+
+    def _validate_component_offload_policy(self, modules: PipelineModules) -> None:
+        known_components = {"default", *modules.encoder_names, *modules.vae_names}
+        unknown = sorted(self.config.dlo_offload_components.keys() - known_components)
+        if unknown:
+            raise ValueError(
+                f"dlo_offload_components contains unknown auxiliary components: {unknown}. DLO always owns DiT offload."
+            )
+
+    def _prepare_auxiliary_components(self, modules: PipelineModules, plan: OffloadPlan | None) -> None:
+        for encoder, name in zip(modules.encoders, modules.encoder_names, strict=True):
+            offload = self._component_offload_enabled(name)
+            encoder._omni_dlo_offload_enabled = offload
+            if offload:
+                self._try_layerwise_offload_encoder(encoder, name, plan)
+            self._register_on_demand_hook(
+                encoder,
+                name,
+                stage_on_demand=(offload and plan is not None and name in plan.on_demand_component_paths),
+            )
+
+        for vae, name in zip(modules.vaes, modules.vae_names, strict=True):
+            offload = self._component_offload_enabled(name)
+            vae._omni_dlo_offload_enabled = offload
+            self._register_on_demand_hook(
+                vae,
+                name,
+                stage_on_demand=(offload and plan is not None and name in plan.on_demand_component_paths),
+            )
+
     def _try_layerwise_offload_encoder(self, module: nn.Module, name: str, plan: OffloadPlan | None) -> bool:
         """Stream plan-declared encoder blocks on each rank without AllGather."""
         if plan is None or name not in plan.encoder_block_attrs:
@@ -1668,6 +1701,7 @@ class DistributedLayerwiseOffloadBackend(OffloadBackend):
         # Retrieve optional declarative OffloadPlan from the pipeline.
         # When present, replaces heuristic block discovery.
         plan = get_offload_plan(pipeline)
+        self._validate_component_offload_policy(modules)
 
         if self.config.dlo_resident_layers and (plan is None or not plan.resident_dit_paths):
             logger.warning(
@@ -1728,21 +1762,10 @@ class DistributedLayerwiseOffloadBackend(OffloadBackend):
                 )
             logger.info("DLO is using host tensors materialized by the ordinary loader")
 
-        # Keep VAE/encoders on CPU; move to GPU on-demand via hooks.
+        # Apply the user policy to discovered auxiliary components. The default
-        # This saves ~4.3 GB HBM per card (VAE 1.3 + encoder 1.1 + sound 1.9)
+        # keeps the model-declared staged lifecycle; False leaves a component
-        # during the DiT forward pass.  They are only needed briefly for
+        # resident for repeated requests.
-        # text-encoding (before DiT) and VAE-decode (after DiT).
+        self._prepare_auxiliary_components(modules, plan)
-        for enc, enc_name in zip(modules.encoders, modules.encoder_names):
-            self._try_layerwise_offload_encoder(enc, enc_name, plan)
-            self._register_on_demand_hook(
-                enc, "encoder", stage_on_demand=plan is not None and enc_name in plan.on_demand_component_paths
-            )
-        for vae, vae_name in zip(modules.vaes, modules.vae_names):
-            self._register_on_demand_hook(
-                vae,
-                "vae",
-                stage_on_demand=(plan is not None and vae_name in plan.on_demand_component_paths),
-            )
 
         # Move resident modules to GPU (small modules needed every forward)
         for name, module in zip(modules.resident_names, modules.resident_modules):

vllm_omni/diffusion/offloader/module_residency.py

@@ -20,17 +20,6 @@ from .tensor_utils import is_dtensor, set_tensor_storage
 logger = init_logger(__name__)
 
 
-# Margin over the largest cached-but-unallocated value observed at a component
-# boundary. The observed peak already includes the weight-reuse payload and
-# the transient footprint of the running workload, so this only covers
-# check-to-check variation.
-_OBSERVED_PEAK_MARGIN = 1.25
-# Hard ceiling for the learned budget so a spurious observation can never
-# monopolize a small device; the #6526 L20X retention plateau (~23.8 GB on a
-# ~140 GB card) stayed well below this fraction.
-_MAX_BUDGET_CAPACITY_FRACTION = 0.25
-
-
 class BoundedAllocatorCache:
     """Retain reusable allocator blocks without monopolizing device memory.
 
@@ -39,100 +28,35 @@ class BoundedAllocatorCache:
     cached blocks are immediately reusable. This policy keeps the cache while
     both of these bounds hold:
 
-    * cached-but-unallocated memory stays within the learned budget — the
+    * cached-but-unallocated memory is at most 25% of device capacity; and
-      largest cached value observed at any component boundary times a margin,
-      capped at a quarter of device capacity; and
     * at least 5% of device capacity is physically free.
 
-    The budget needs no user-chosen fraction, no multiplier constant, and no
+    Missing memory telemetry is handled conservatively by releasing the cache.
-    warmup flag. It starts at zero, so the first boundaries of a cold start
+    Failure paths can force release; normal executor shutdown keeps its own
-    release once per observation step while the peak climbs; afterwards the
+    unconditional device-cache cleanup because this policy is not global.
-    budget tracks the observed peak and boundaries retain steadily. A
-    workload change whose cached footprint grows by more than the margin pays
-    one further release while the new peak lands. Detaching the last attached
-    ``PinnedModuleStager`` resets the observed peak, so one workload never
-    permanently influences another. Missing memory telemetry is handled
-    conservatively by releasing the cache. Failure paths can force release;
-    normal executor shutdown keeps its own unconditional device-cache cleanup
-    because this policy is not global.
     """
 
     def __init__(
         self,
         device: torch.device,
         *,
+        max_cached_fraction: float = 0.25,
         min_free_fraction: float = 0.05,
     ) -> None:
+        if not 0.0 <= max_cached_fraction <= 1.0:
+            raise ValueError(f"max_cached_fraction must be in [0, 1], got {max_cached_fraction}")
         if not 0.0 <= min_free_fraction <= 1.0:
             raise ValueError(f"min_free_fraction must be in [0, 1], got {min_free_fraction}")
         self.device = device
+        self.max_cached_fraction = max_cached_fraction
         self.min_free_fraction = min_free_fraction
-        self._staged_bytes = 0
-        self._observed_cached_peak = 0
-
-    def grow_retention_budget(self, staged_bytes: int) -> None:
-        """Record an attached stager's staged byte total."""
-        if staged_bytes < 0:
-            raise ValueError(f"staged_bytes must be >= 0, got {staged_bytes}")
-        self._staged_bytes += int(staged_bytes)
-
-    def shrink_retention_budget(self, staged_bytes: int) -> None:
-        """Record a stager detach; the last detach resets the observed peak."""
-        if staged_bytes < 0:
-            raise ValueError(f"staged_bytes must be >= 0, got {staged_bytes}")
-        self._staged_bytes = max(0, self._staged_bytes - int(staged_bytes))
-        if self._staged_bytes == 0:
-            self._observed_cached_peak = 0
-
-    @property
-    def observed_cached_peak(self) -> int:
-        """Largest cached-but-unallocated value seen at a boundary check."""
-        return self._observed_cached_peak
-
-    @property
-    def budget_bytes(self) -> int:
-        """Reusable-cache ceiling: the observed peak times its margin."""
-        return int(self._observed_cached_peak * _OBSERVED_PEAK_MARGIN)
 
     def _should_release(self) -> bool:
         reserved = int(torch.accelerator.memory_reserved(self.device))
         allocated = int(torch.accelerator.memory_allocated(self.device))
         free, total = current_omni_platform.get_device_memory(self.device)
         cached = max(0, reserved - allocated)
-        capacity_cap = int(total * _MAX_BUDGET_CAPACITY_FRACTION)
+        return cached > int(total * self.max_cached_fraction) or free < int(total * self.min_free_fraction)
-        budget = min(self.budget_bytes, capacity_cap)
-        # Decide against the pre-observation budget, then record the peak so a
-        # one-off over-budget release also teaches the budget for later checks.
-        release = cached > budget or free < int(total * self.min_free_fraction)
-        if cached > self._observed_cached_peak:
-            self._observed_cached_peak = cached
-        if release:
-            reason = "cached-over-budget" if cached > budget else "low-free-memory"
-            logger.debug(
-                "Releasing retained allocator cache (%s): cached=%d budget=%d staged=%d "
-                "observed_peak=%d capacity_cap=%d free=%d total=%d",
-                reason,
-                cached,
-                budget,
-                self._staged_bytes,
-                self._observed_cached_peak,
-                capacity_cap,
-                free,
-                total,
-            )
-            return True
-        logger.debug(
-            "Retaining allocator cache: cached=%d budget=%d staged=%d observed_peak=%d "
-            "capacity_cap=%d free=%d total=%d",
-            cached,
-            budget,
-            self._staged_bytes,
-            self._observed_cached_peak,
-            capacity_cap,
-            free,
-            total,
-        )
-        return False
 
     def release_if_needed(self, *, force: bool = False) -> bool:
         """Release cached blocks when a bound is crossed or release is forced."""
@@ -192,12 +116,11 @@ class PinnedModuleStager:
         self.device = device
         self.copy_stream = copy_stream if copy_stream is not None else current_omni_platform.Stream()
         self._ready_event = current_omni_platform.Event()
-        self.cache_retention = None
+        self.cache_retention = cache_retention
         self.loaded = False
         self._groups = self._snapshot_groups(modules, pin_memory=pin_memory)
         self._device_storages: list[torch.Tensor] = []
         self._restore_masters()
-        self.set_cache_retention(cache_retention)
 
     @staticmethod
     def _local_tensor(target: torch.Tensor) -> torch.Tensor:
@@ -275,19 +198,8 @@ class PinnedModuleStager:
     def _restore_masters(self) -> None:
         self._bind([group.master for group in self._groups])
 
-    @property
-    def staged_bytes(self) -> int:
-        """Total unique staged storage bytes (masters are uint8 byte views)."""
-        return sum(group.master.numel() for group in self._groups)
-
     def set_cache_retention(self, cache_retention: BoundedAllocatorCache | None) -> None:
-        if cache_retention is self.cache_retention:
-            return
-        if self.cache_retention is not None:
-            self.cache_retention.shrink_retention_budget(self.staged_bytes)
         self.cache_retention = cache_retention
-        if cache_retention is not None:
-            cache_retention.grow_retention_budget(self.staged_bytes)
 
     def _release_cache(self, *, force: bool = False) -> None:
         if self.cache_retention is None:
@@ -327,10 +239,6 @@ class PinnedModuleStager:
         except torch.OutOfMemoryError:
             # A retained cache is normally reusable by this process, but an
             # explicit flush gives external memory pressure one bounded retry.
-            logger.info(
-                "Staging out of memory (%d staged bytes); releasing retained cache and retrying once",
-                self.staged_bytes,
-            )
             self._cleanup_failed_load()
             try:
                 self._load_once()

vllm_omni/engine/arg_utils.py

@@ -569,6 +569,7 @@ class OrchestratorArgs:
     enable_distributed_layerwise_offload: bool = False
     dlo_use_allgather: bool = True
     dlo_resident_layers: int = 0
+    dlo_offload_components: dict[str, bool] = field(default_factory=dict)
     host_weight_runtime_mode: str = "disabled"
     host_weight_runtime_root: str | None = None
     dlo_host_registration_limit_gib: float = 0.0

vllm_omni/engine/async_omni_engine.py

@@ -1059,6 +1059,9 @@ class AsyncOmniEngine:
             "enable_distributed_layerwise_offload": kwargs.get("enable_distributed_layerwise_offload", False),
             "dlo_use_allgather": kwargs.get("dlo_use_allgather", True),
             "dlo_resident_layers": kwargs.get("dlo_resident_layers", 0),
+            "dlo_offload_components": (
+                {} if kwargs.get("dlo_offload_components") is None else kwargs["dlo_offload_components"]
+            ),
             "host_weight_runtime_mode": kwargs.get("host_weight_runtime_mode", "disabled"),
             "host_weight_runtime_root": kwargs.get("host_weight_runtime_root"),
             "dlo_host_registration_limit_gib": kwargs.get("dlo_host_registration_limit_gib", 0.0),

vllm_omni/entrypoints/cli/serve.py

@@ -770,6 +770,18 @@ class OmniServeCommand(CLISubcommand):
                 "under the existing pinned-memory policy and fall back to bounded staging when unavailable."
             ),
         )
+        omni_config_group.add_argument(
+            "--dlo-offload-components",
+            type=json.loads,
+            default=None,
+            metavar="JSON",
+            help=(
+                "Per auxiliary encoder/VAE DLO policy as a JSON object, for example "
+                "'{\"text_encoder\": false}'. False keeps the whole component "
+                "resident; true permits the model's existing offload behavior. "
+                "DiT offload is not controlled by this map."
+            ),
+        )
         # Video model parameters (e.g., Wan2.2) - engine-level
         omni_config_group.add_argument(
             "--boundary-ratio",