[Model] Make the dummy-run recipe feature-neutral

Per review, the warmup recipe is a property of the model's request
geometry, not of distributed layerwise offload: rename
dlo_dummy_run_recipe to dummy_run_recipe and drop the DLO gate in
_make_dummy_request. A model whose geometry needs recipe-specific sampling
arguments warms up at startup whatever features are enabled — under DLO
the generation also primes the component cache-retention budget; without
it, the same warmup covers any other first-request cost. Validated both
ways on L20X: DLO boot unchanged (releases stay in the startup window) and
a non-DLO MiniMax-H3 boot now runs the warmup and serves normally.

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

docs/design/feature/offloader/distributed_layerwise_offload.md

@@ -177,8 +177,24 @@ 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 is at most 25% of device capacity and at least 5% of device capacity
-remains physically free. Crossing either bound releases the allocator cache.
+while it stays within a budget learned from the running workload and at least
+5% of device capacity remains physically free. The budget is the largest
+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

docs/user_guide/diffusion/offloader/distributed_layerwise_offload.md

@@ -67,6 +67,16 @@ omni = Omni(
 | `--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

tests/diffusion/offloader/test_module_residency.py

@@ -202,21 +202,96 @@ def test_allocator_cache_policy_enforces_cache_and_free_memory_bounds(
     expected_release,
 ):
     empty_cache, _ = patched_runtime
-    monkeypatch.setattr(torch.accelerator, "memory_reserved", lambda _device: reserved)
-    monkeypatch.setattr(torch.accelerator, "memory_allocated", lambda _device: allocated)
+    telemetry = {"reserved": 20, "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",
-        lambda _device: (free, 100),
+        lambda _device: (telemetry["free"], 400),
     )
     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
 
@@ -228,3 +303,29 @@ 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,6 +70,161 @@ 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)

vllm_omni/diffusion/diffusion_engine.py

@@ -36,6 +36,7 @@ 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,
@@ -64,6 +65,9 @@ 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:
@@ -1017,8 +1021,36 @@ class DiffusionEngine:
     ) -> OmniDiffusionRequest | None:
         """Build a one-step model request for startup profiling or warmup."""
 
-        prompt: OmniTextPrompt = {"prompt": "dummy run"}
         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"}
         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)]
@@ -1028,9 +1060,6 @@ 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,6 +4,7 @@
 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
@@ -54,6 +55,22 @@ 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,6 +704,20 @@ 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]] = {}

vllm_omni/diffusion/offloader/module_residency.py

@@ -20,6 +20,17 @@ 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.
 
@@ -28,35 +39,100 @@ class BoundedAllocatorCache:
     cached blocks are immediately reusable. This policy keeps the cache while
     both of these bounds hold:
 
-    * cached-but-unallocated memory is at most 25% of device capacity; and
+    * cached-but-unallocated memory stays within the learned budget — the
+      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.
 
-    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.
+    The budget needs no user-chosen fraction, no multiplier constant, and no
+    warmup flag. It starts at zero, so the first boundaries of a cold start
+    release once per observation step while the peak climbs; afterwards the
+    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)
-        return cached > int(total * self.max_cached_fraction) or free < int(total * self.min_free_fraction)
+        capacity_cap = int(total * _MAX_BUDGET_CAPACITY_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."""
@@ -116,11 +192,12 @@ 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 = cache_retention
+        self.cache_retention = None
         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:
@@ -198,8 +275,19 @@ 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:
@@ -239,6 +327,10 @@ 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()