Expose remote preparation as active queue work

README.md

@@ -228,6 +228,8 @@ Subgraph containers inherit a derived summary decoration (marked with a `Σ` bad
 
 Active remote nodes show a small progress panel with per-bar `it/s` rates (LLM nodes show `tok/s`, token counts, and TTFT). When a prompt is scheduled across more than one remote environment, the panel also names where each component is running — the Modal container id or the SSH host name. Preview images and UI payloads from remote nodes stream back into the local session while the component is still running, and progress state is reconciled against queue history when the browser regains focus so cancelled prompts do not leave stale bars behind.
 
+Remote planning and capacity acquisition count as active queue work even though they happen before the rewritten prompt enters ComfyUI's native execution queue. During Vast startup, the canvas receives readable updates for rental, worker-image download, SSH readiness, and runtime initialization. A provider-side terminal image-pull failure is reported immediately rather than appearing as an empty queue until the startup timeout expires.
+
 While a Modal workflow is active, the global status pill lists this instance's active containers, an estimated accumulated GPU cost with the current burn rate (based on Modal's published GPU prices; excludes CPU, memory, storage, credits, and reservations), and — where the workspace plan allows billing-report access — Modal's actual metered app cost at hourly resolution.
 
 Cancelling a local prompt propagates a targeted interrupt to the active remote work. If the back-end is slow to observe it, the local prompt is released after a grace period while remote cleanup continues.

api_intercept.py

@@ -88,6 +88,7 @@ from .vast_api import VastApiClient
 from .vast_leases import VastLeaseRegistry
 
 logger = logging.getLogger(__name__)
+SetupStatusCallback = Callable[[str, int | None, int | None], None]
 
 _ROUTE_REGISTERED = False
 _MODAL_UI_EVENT_RETENTION_SECONDS = 2 * 60 * 60
@@ -95,6 +96,8 @@ _MODAL_UI_EVENT_LIMIT_PER_CLIENT = 512
 _MODAL_UI_EVENTS_LOCK = threading.Lock()
 _MODAL_UI_EVENTS_BY_CLIENT: dict[str, deque[dict[str, Any]]] = {}
 _MODAL_INTERRUPT_QUEUE_BRIDGE_ATTR = "__comfy_modal_interrupt_queue_bridge_installed"
+_REMOTE_PREPARATION_PROMPTS_ATTR = "__comfy_modal_remote_preparation_prompts"
+_REMOTE_PREPARATION_LOCK_ATTR = "__comfy_modal_remote_preparation_lock"
 _REMOTE_TOGGLE_WIDGET_NAME = "Run Remotely"
 _LEGACY_REMOTE_TOGGLE_WIDGET_NAME = "Run on Modal"
 _LOCAL_ONLY_REMOTE_CLASS_TYPES = frozenset(
@@ -1039,6 +1042,7 @@ def _plan_component_execution_assignments(
     prompt: Mapping[str, Any],
     workflow: Mapping[str, Any] | None,
     settings: ModalSyncSettings,
+    status_callback: SetupStatusCallback | None = None,
 ) -> dict[str, ExecutionAssignment]:
     """Assign components across Modal, SSH Docker, and workflow-declared Vast pools."""
     preferences = WorkflowExecutionPreferences.from_workflow(workflow)
@@ -1237,7 +1241,17 @@ def _plan_component_execution_assignments(
                     "Vast.ai was selected without a current marketplace quote."
                 )
             try:
-                lease = vast_service.acquire_sync(vast_quote)
+                if status_callback is None:
+                    lease = vast_service.acquire_sync(vast_quote)
+                else:
+                    lease = vast_service.acquire_sync(
+                        vast_quote,
+                        status_callback=lambda message: status_callback(
+                            message,
+                            None,
+                            None,
+                        ),
+                    )
             except (OSError, RuntimeError, ValueError) as exc:
                 raise ModalPromptValidationError(
                     f"Unable to acquire Vast.ai capacity: {exc}"
@@ -1270,6 +1284,7 @@ def _plan_component_execution(
     prompt: Mapping[str, Any],
     workflow: Mapping[str, Any] | None,
     settings: ModalSyncSettings,
+    status_callback: SetupStatusCallback | None = None,
 ) -> ComponentExecutionPlan:
     """Plan through connected configurations or synthesize the legacy behavior."""
     try:
@@ -1283,6 +1298,7 @@ def _plan_component_execution(
                 prompt=prompt,
                 workflow=workflow,
                 settings=settings,
+                status_callback=status_callback,
             )
         )
     return _plan_configured_component_execution(
@@ -1291,6 +1307,7 @@ def _plan_component_execution(
         workflow=workflow,
         settings=settings,
         configuration_set=configuration_set,
+        status_callback=status_callback,
     )
 
 
@@ -1301,6 +1318,7 @@ def _plan_configured_component_execution(
     workflow: Mapping[str, Any] | None,
     settings: ModalSyncSettings,
     configuration_set: RemoteConfigurationSet,
+    status_callback: SetupStatusCallback | None = None,
 ) -> ComponentExecutionPlan:
     """Resolve and prepare a capacity-aware plan from connected configurations."""
     configurations_by_id = {
@@ -1360,6 +1378,7 @@ def _plan_configured_component_execution(
         requirements_by_component=requirements_by_component,
         vast_quotes=vast_quotes,
         vast_service=vast_service,
+        status_callback=status_callback,
     )
     return ComponentExecutionPlan(
         assignments=assignments,
@@ -1650,6 +1669,7 @@ def _prepare_selected_vast_capacity(
     requirements_by_component: Mapping[str, ComponentResourceRequirements],
     vast_quotes: Mapping[tuple[str, str], VastProfileQuote],
     vast_service: VastService | None,
+    status_callback: SetupStatusCallback | None = None,
 ) -> dict[str, Any]:
     """Acquire only Vast slots selected by the completed provider-neutral plan."""
     selected_slots: dict[tuple[str, int], list[str]] = defaultdict(list)
@@ -1676,7 +1696,12 @@ def _prepare_selected_vast_capacity(
         if isinstance(configuration, VastRemoteConfiguration)
     }
     leases_by_environment: dict[str, Any] = {}
-    for (configuration_id, slot_index), component_ids in sorted(selected_slots.items()):
+    ordered_slots = sorted(selected_slots.items())
+    total_slots = len(ordered_slots)
+    for slot_number, ((configuration_id, slot_index), component_ids) in enumerate(
+        ordered_slots,
+        start=1,
+    ):
         configuration = configurations_by_id[configuration_id]
         quote = _quote_selected_vast_slot(
             configuration=configuration,
@@ -1686,12 +1711,37 @@ def _prepare_selected_vast_capacity(
             vast_service=vast_service,
         )
         try:
-            lease = vast_service.acquire_sync(quote, slot=slot_index)
+            if status_callback is not None:
+                status_callback(
+                    f"Acquiring Vast.ai capacity {slot_number} of {total_slots}",
+                    slot_number - 1,
+                    total_slots,
+                )
+            if status_callback is None:
+                lease = vast_service.acquire_sync(quote, slot=slot_index)
+            else:
+                lease = vast_service.acquire_sync(
+                    quote,
+                    slot=slot_index,
+                    status_callback=(
+                        lambda message, current=slot_number - 1: status_callback(
+                            message,
+                            current,
+                            total_slots,
+                        )
+                    ),
+                )
         except (OSError, RuntimeError, ValueError) as exc:
             raise ModalPromptValidationError(
                 f"Unable to acquire Vast.ai capacity for configuration "
                 f"{configuration.display_name!r} slot {slot_index}: {exc}"
             ) from exc
+        if status_callback is not None:
+            status_callback(
+                f"Vast.ai capacity {slot_number} of {total_slots} is ready",
+                slot_number,
+                total_slots,
+            )
         leases_by_environment[lease.environment_id] = lease
         cost_share = quote.predicted_incremental_cost_usd / len(component_ids)
         for component_id in component_ids:
@@ -6959,7 +7009,7 @@ def rewrite_prompt_for_modal(
     settings: ModalSyncSettings | None = None,
     nodes_module: Any | None = None,
     extra_data: dict[str, Any] | None = None,
-    status_callback: Any | None = None,
+    status_callback: SetupStatusCallback | None = None,
 ) -> tuple[dict[str, Any], RewriteSummary]:
     """Rewrite connected remote components into Modal proxy nodes."""
     resolved_settings = settings or get_settings()
@@ -7017,6 +7067,7 @@ def rewrite_prompt_for_modal(
         prompt=rewritten_prompt,
         workflow=workflow,
         settings=resolved_settings,
+        status_callback=status_callback,
     )
     assignments_by_component_id = execution_plan.assignments
     if execution_plan.configuration_set is not None:
@@ -7491,7 +7542,7 @@ async def rewrite_prompt_for_modal_async(
     settings: ModalSyncSettings | None = None,
     nodes_module: Any | None = None,
     extra_data: dict[str, Any] | None = None,
-    status_callback: Any | None = None,
+    status_callback: SetupStatusCallback | None = None,
 ) -> tuple[dict[str, Any], RewriteSummary]:
     """Prepare one Modal prompt without blocking ComfyUI's event loop."""
     return await asyncio.to_thread(
@@ -7856,7 +7907,7 @@ async def delete_modal_volume(settings: ModalSyncSettings) -> dict[str, Any]:
 
 
 def _install_modal_interrupt_queue_bridge(prompt_server: Any) -> None:
-    """Expose active Modal prompts to ComfyUI's targeted interrupt route."""
+    """Expose active remote work through every ComfyUI queue-state view."""
     prompt_queue = getattr(prompt_server, "prompt_queue", None)
     if prompt_queue is None or getattr(
         prompt_queue, _MODAL_INTERRUPT_QUEUE_BRIDGE_ATTR, False
@@ -7864,30 +7915,156 @@ def _install_modal_interrupt_queue_bridge(prompt_server: Any) -> None:
         return
 
     original_get_current_queue = getattr(prompt_queue, "get_current_queue", None)
-    if not callable(original_get_current_queue):
+    original_get_current_queue_volatile = getattr(
+        prompt_queue, "get_current_queue_volatile", None
+    )
+    original_get_tasks_remaining = getattr(prompt_queue, "get_tasks_remaining", None)
+    if not any(
+        callable(method)
+        for method in (
+            original_get_current_queue,
+            original_get_current_queue_volatile,
+            original_get_tasks_remaining,
+        )
+    ):
         logger.debug(
-            "Prompt queue does not expose get_current_queue(); skipping Modal interrupt bridge."
+            "Prompt queue does not expose queue-state methods; skipping remote queue bridge."
         )
         return
 
-    def modal_get_current_queue() -> tuple[list[Any], Any]:
-        """Return ComfyUI's current queue plus synthetic active Modal prompt entries."""
-        running, queued = original_get_current_queue()
-        running_items = list(running)
-        running_prompt_ids = {str(item[1]) for item in running_items if len(item) > 1}
-        try:
-            from .remote.modal_app import active_remote_modal_prompt_ids
-        except ImportError:
-            return running_items, queued
+    preparation_prompts: dict[str, tuple[Any, ...]] = {}
+    preparation_lock = threading.RLock()
+    setattr(prompt_queue, _REMOTE_PREPARATION_PROMPTS_ATTR, preparation_prompts)
+    setattr(prompt_queue, _REMOTE_PREPARATION_LOCK_ATTR, preparation_lock)
 
-        active_prompt_ids = active_remote_modal_prompt_ids()
-        for prompt_id in sorted(active_prompt_ids - running_prompt_ids):
-            running_items.append((0, prompt_id, {}, {}, [], {}))
+    def preparation_items() -> list[tuple[Any, ...]]:
+        """Return a stable snapshot of prompts still preparing remote capacity."""
+        with preparation_lock:
+            return list(preparation_prompts.values())
+
+    def append_missing_preparations(
+        running: Iterable[Any],
+        queued: Iterable[Any],
+    ) -> tuple[list[Any], Any]:
+        """Add preparation entries that are not already in ComfyUI's native queue."""
+        running_items = list(running)
+        queued_items = list(queued)
+        native_prompt_ids = _queue_item_prompt_ids((*running_items, *queued_items))
+        running_items.extend(
+            item
+            for item in preparation_items()
+            if str(item[1]) not in native_prompt_ids
+        )
         return running_items, queued
 
-    setattr(prompt_queue, "get_current_queue", modal_get_current_queue)
+    if callable(original_get_current_queue):
+
+        def remote_get_current_queue() -> tuple[list[Any], Any]:
+            """Return native work plus preparation and active remote prompt entries."""
+            running, queued = original_get_current_queue()
+            running_items, queued = append_missing_preparations(running, queued)
+            running_prompt_ids = _queue_item_prompt_ids(running_items)
+            try:
+                from .remote.modal_app import active_remote_modal_prompt_ids
+            except ImportError:
+                return running_items, queued
+
+            for prompt_id in sorted(
+                active_remote_modal_prompt_ids() - running_prompt_ids
+            ):
+                running_items.append((0, prompt_id, {}, {}, [], {}))
+            return running_items, queued
+
+        setattr(prompt_queue, "get_current_queue", remote_get_current_queue)
+
+    if callable(original_get_current_queue_volatile):
+
+        def remote_get_current_queue_volatile() -> tuple[list[Any], Any]:
+            """Include remote preparation in ComfyUI's public `/queue` response."""
+            running, queued = original_get_current_queue_volatile()
+            return append_missing_preparations(running, queued)
+
+        setattr(
+            prompt_queue,
+            "get_current_queue_volatile",
+            remote_get_current_queue_volatile,
+        )
+
+    if callable(original_get_tasks_remaining):
+
+        def remote_get_tasks_remaining() -> int:
+            """Count remote preparation as work in websocket queue status."""
+            remaining = int(original_get_tasks_remaining())
+            native_prompt_ids: set[str] = set()
+            native_queue_method = (
+                original_get_current_queue_volatile
+                if callable(original_get_current_queue_volatile)
+                else original_get_current_queue
+            )
+            if callable(native_queue_method):
+                running, queued = native_queue_method()
+                native_prompt_ids = _queue_item_prompt_ids((*running, *queued))
+            return remaining + sum(
+                str(item[1]) not in native_prompt_ids
+                for item in preparation_items()
+            )
+
+        setattr(prompt_queue, "get_tasks_remaining", remote_get_tasks_remaining)
+
     setattr(prompt_queue, _MODAL_INTERRUPT_QUEUE_BRIDGE_ATTR, True)
-    logger.info("Installed Modal targeted interrupt bridge on ComfyUI prompt queue.")
+    logger.info("Installed remote preparation bridge on ComfyUI prompt queue.")
+
+
+def _queue_item_prompt_ids(items: Iterable[Any]) -> set[str]:
+    """Return prompt IDs from well-formed native or synthetic queue items."""
+    return {
+        str(item[1])
+        for item in items
+        if isinstance(item, (list, tuple)) and len(item) > 1
+    }
+
+
+def _set_remote_preparation(
+    prompt_server: Any,
+    *,
+    prompt_id: str,
+    prompt: Mapping[str, Any],
+    extra_data: Mapping[str, Any],
+) -> bool:
+    """Register one pre-queue remote workflow and publish its active queue state."""
+    prompt_queue = getattr(prompt_server, "prompt_queue", None)
+    preparations = getattr(prompt_queue, _REMOTE_PREPARATION_PROMPTS_ATTR, None)
+    preparation_lock = getattr(prompt_queue, _REMOTE_PREPARATION_LOCK_ATTR, None)
+    if not isinstance(preparations, dict) or preparation_lock is None:
+        return False
+    with preparation_lock:
+        preparations[prompt_id] = (
+            0,
+            prompt_id,
+            copy.deepcopy(dict(prompt)),
+            copy.deepcopy(dict(extra_data)),
+            [],
+            {},
+        )
+    queue_updated = getattr(prompt_server, "queue_updated", None)
+    if callable(queue_updated):
+        queue_updated()
+    return True
+
+
+def _clear_remote_preparation(prompt_server: Any, prompt_id: str) -> None:
+    """Remove one pre-queue remote workflow and publish the resulting queue state."""
+    prompt_queue = getattr(prompt_server, "prompt_queue", None)
+    preparations = getattr(prompt_queue, _REMOTE_PREPARATION_PROMPTS_ATTR, None)
+    preparation_lock = getattr(prompt_queue, _REMOTE_PREPARATION_LOCK_ATTR, None)
+    if not isinstance(preparations, dict) or preparation_lock is None:
+        return
+    with preparation_lock:
+        removed = preparations.pop(prompt_id, None)
+    if removed is not None:
+        queue_updated = getattr(prompt_server, "queue_updated", None)
+        if callable(queue_updated):
+            queue_updated()
 
 
 def setup_modal_queue_route(
@@ -8376,6 +8553,7 @@ def setup_modal_queue_route(
         remote_node_ids: list[str] = []
         request_modal_gpu: str | None = None
         summary = RewriteSummary()
+        preparation_prompt_id: str | None = None
         try:
             request_started_at = time.perf_counter()
             json_data = await request.json()
@@ -8419,6 +8597,13 @@ def setup_modal_queue_route(
                     "Attached prompt-scoped Modal execution metadata prompt_id=%s.",
                     prompt_id,
                 )
+                if _set_remote_preparation(
+                    prompt_server,
+                    prompt_id=prompt_id,
+                    prompt=json_data.get("prompt", {}),
+                    extra_data=json_data.get("extra_data", {}),
+                ):
+                    preparation_prompt_id = prompt_id
 
             try:
                 request_settings = settings_for_modal_gpu(
@@ -8635,6 +8820,9 @@ def setup_modal_queue_route(
                     error_message=str(exc),
                 )
             return web.json_response({"error": str(exc), "node_errors": []}, status=500)
+        finally:
+            if preparation_prompt_id is not None:
+                _clear_remote_preparation(prompt_server, preparation_prompt_id)
 
     _ROUTE_REGISTERED = True
     logger.info(

docs/LEARNINGS.md

@@ -2,6 +2,11 @@
 
 - The pinned vLLM wheel is large enough that pip's default five resumptions and 15-second socket timeout can be exhausted by an unstable connection during an SSH/Vast worker build. Give the accelerator-wheel layer an extended timeout and bounded resume budget while leaving earlier package layers unchanged so Docker can reuse their successful cache entries.
 
+## 2026-08-23
+
+- Remote capacity acquisition happens before the rewritten prompt enters ComfyUI's native queue, so a one-shot frontend event is not enough to represent liveness. Register prompt preparation in a server-side bridge that augments `get_current_queue`, `get_current_queue_volatile`, and `get_tasks_remaining`, deduplicate it when the native prompt arrives, and publish queue updates on both registration and cleanup.
+- Vast can report a failed private-image pull as `actual_status=loading` even though `cur_state` and `intended_status` are both `stopped`. Preserve those provider fields, surface bounded status messages during every readiness poll, and treat the stopped/stopped combination as terminal instead of waiting for the full startup timeout.
+
 ## 2026-08-22
 
 - Current free VRAM is conservative only if the planner distinguishes unknown GPU consumers from its own idle warm workers. Compare normal placement with an optimistic total-capacity placement; recycle only the labeled idle worker selected by the cost model, then re-probe actual free memory before admission. A running worker is idle only when `docker top` shows neither an execution relay nor a model-staging process.

tests/test_api_intercept.py

@@ -6087,3 +6087,183 @@ def test_modal_interrupt_queue_bridge_exposes_active_remote_prompts(
 
     assert queued == ["queued"]
     assert [item[1] for item in running] == ["prompt-1"]
+
+
+def test_remote_preparation_bridge_exposes_work_to_all_queue_views(
+    api_intercept_module: Any,
+) -> None:
+    """Capacity acquisition should look active before the rewritten prompt is queued."""
+
+    class FakePromptQueue:
+        """Minimal native queue with every state method used by ComfyUI."""
+
+        def __init__(self) -> None:
+            """Initialize an empty native queue."""
+            self.running: list[tuple[Any, ...]] = []
+            self.queued: list[tuple[Any, ...]] = []
+
+        def get_current_queue(self) -> tuple[list[Any], list[Any]]:
+            """Return stable native queue state."""
+            return list(self.running), list(self.queued)
+
+        def get_current_queue_volatile(self) -> tuple[list[Any], list[Any]]:
+            """Return volatile native queue state."""
+            return list(self.running), list(self.queued)
+
+        def get_tasks_remaining(self) -> int:
+            """Count native running and pending prompts."""
+            return len(self.running) + len(self.queued)
+
+    prompt_queue = FakePromptQueue()
+    queue_update_counts: list[int] = []
+    prompt_server = SimpleNamespace(
+        prompt_queue=prompt_queue,
+        queue_updated=lambda: queue_update_counts.append(
+            prompt_queue.get_tasks_remaining()
+        ),
+    )
+    api_intercept_module._install_modal_interrupt_queue_bridge(prompt_server)
+
+    registered = api_intercept_module._set_remote_preparation(
+        prompt_server,
+        prompt_id="prompt-preparing",
+        prompt={"1": {"class_type": "RemoteImage", "inputs": {}}},
+        extra_data={"client_id": "client-1"},
+    )
+
+    assert registered is True
+    assert prompt_queue.get_tasks_remaining() == 1
+    assert [item[1] for item in prompt_queue.get_current_queue()[0]] == [
+        "prompt-preparing"
+    ]
+    assert [item[1] for item in prompt_queue.get_current_queue_volatile()[0]] == [
+        "prompt-preparing"
+    ]
+
+    prompt_queue.queued.append((1, "prompt-preparing", {}, {}, []))
+    assert prompt_queue.get_tasks_remaining() == 1
+    assert [item[1] for item in prompt_queue.get_current_queue_volatile()[0]] == []
+
+    api_intercept_module._clear_remote_preparation(
+        prompt_server,
+        "prompt-preparing",
+    )
+
+    assert prompt_queue.get_tasks_remaining() == 1
+    assert queue_update_counts == [1, 1]
+
+
+def test_remote_preparation_bridge_clears_failed_submission(
+    api_intercept_module: Any,
+) -> None:
+    """A rejected pre-queue prompt must not leave phantom active queue work."""
+
+    class FakePromptQueue:
+        """Minimal empty queue used to exercise preparation cleanup."""
+
+        def get_current_queue_volatile(self) -> tuple[list[Any], list[Any]]:
+            """Return empty native queue state."""
+            return [], []
+
+        def get_tasks_remaining(self) -> int:
+            """Return the empty native queue count."""
+            return 0
+
+    prompt_queue = FakePromptQueue()
+    prompt_server = SimpleNamespace(
+        prompt_queue=prompt_queue,
+        queue_updated=lambda: None,
+    )
+    api_intercept_module._install_modal_interrupt_queue_bridge(prompt_server)
+    assert api_intercept_module._set_remote_preparation(
+        prompt_server,
+        prompt_id="prompt-failed",
+        prompt={},
+        extra_data={},
+    )
+    assert prompt_queue.get_tasks_remaining() == 1
+
+    api_intercept_module._clear_remote_preparation(prompt_server, "prompt-failed")
+
+    assert prompt_queue.get_tasks_remaining() == 0
+    assert prompt_queue.get_current_queue_volatile() == ([], [])
+
+
+def test_selected_vast_capacity_streams_setup_status(
+    api_intercept_module: Any,
+    vast_models_module: Any,
+) -> None:
+    """Queue-time Vast acquisition should stream provider progress into the UI."""
+    profile = vast_models_module.VastResourceProfile(
+        profile_id="vast-config",
+        profile_name="Vast-pool",
+        maximum_instances=1,
+    )
+    configuration = api_intercept_module.VastRemoteConfiguration(
+        configuration_id="vast-config",
+        display_name="Vast-pool",
+        profile=profile,
+    )
+    configuration_set = api_intercept_module.RemoteConfigurationSet(
+        configurations=(configuration,)
+    )
+    assignments = {
+        "component-1": api_intercept_module.ExecutionAssignment(
+            environment_id=profile.environment_id,
+            provider=api_intercept_module.ExecutionProvider.VAST,
+            predicted_cost_usd=0.02,
+            predicted_completion_seconds=30.0,
+            configuration_id="vast-config",
+            capacity_slot_index=0,
+        )
+    }
+    requirements = {
+        "component-1": api_intercept_module.ComponentResourceRequirements(
+            estimated_execution_seconds=30.0
+        )
+    }
+    quote = SimpleNamespace(
+        profile=profile,
+        predicted_incremental_cost_usd=0.02,
+    )
+    status_events: list[tuple[str, int | None, int | None]] = []
+
+    class FakeVastService:
+        """Emit representative readiness phases without renting an instance."""
+
+        def acquire_sync(
+            self,
+            selected_quote: Any,
+            *,
+            slot: int,
+            status_callback: Any,
+        ) -> Any:
+            """Emit image and runtime phases before returning a fake lease."""
+            assert selected_quote is quote
+            assert slot == 0
+            status_callback("Vast.ai instance 42 is downloading the worker image")
+            status_callback("Initializing Vast.ai worker")
+            return SimpleNamespace(
+                environment_id="vast:vast-config:42",
+                idle_retention_seconds=3600.0,
+            )
+
+    leases = api_intercept_module._prepare_selected_vast_capacity(
+        assignments=assignments,
+        configuration_set=configuration_set,
+        requirements_by_component=requirements,
+        vast_quotes={("component-1", "vast-config"): quote},
+        vast_service=FakeVastService(),
+        status_callback=lambda message, current, total: status_events.append(
+            (message, current, total)
+        ),
+    )
+
+    assert list(leases) == ["vast:vast-config:42"]
+    assert status_events == [
+        ("Acquiring Vast.ai capacity 1 of 1", 0, 1),
+        ("Vast.ai instance 42 is downloading the worker image", 0, 1),
+        ("Initializing Vast.ai worker", 0, 1),
+        ("Vast.ai capacity 1 of 1 is ready", 1, 1),
+    ]
+    assert assignments["component-1"].environment_id == "vast:vast-config:42"

tests/test_vast_api_simulator.py

@@ -61,15 +61,23 @@ def test_client_search_create_poll_manage_and_destroy_lifecycle(
                 environment={},
             )
             created = await client.create_instance(offers[0].offer_id, launch)
+            startup_states: list[tuple[str, str | None]] = []
             instance = await client.wait_until_ready(
                 created.instance_id,
                 timeout_seconds=2.0,
                 poll_interval_seconds=0.01,
+                status_callback=lambda current: startup_states.append(
+                    (current.actual_status, current.status_message)
+                ),
             )
 
             assert instance.ready_for_ssh is True
             assert instance.ssh_host is not None
             assert instance.ssh_port is not None
+            assert startup_states == [
+                ("loading", "worker-layer: Download complete"),
+                ("running", "Worker ready"),
+            ]
             listed = await client.list_instances()
             assert [item.instance_id for item in listed] == [created.instance_id]
 
@@ -86,6 +94,55 @@ def test_client_search_create_poll_manage_and_destroy_lifecycle(
     asyncio.run(scenario())
 
 
+def test_client_fails_fast_when_vast_reports_stopped_image_pull(
+    vast_api_module: Any,
+    vast_models_module: Any,
+    vast_simulator_module: Any,
+) -> None:
+    """A rejected worker image must not look like loading until the full timeout."""
+
+    async def scenario() -> None:
+        """Create an instance, inject Vast's registry failure state, and poll once."""
+        state = vast_simulator_module.VastSimulatorState(polls_until_running=999)
+        async with _running_simulator(
+            vast_simulator_module.create_vast_simulator_app(state)
+        ) as base_url:
+            client = vast_api_module.VastApiClient(
+                state.api_key,
+                base_url=base_url,
+                retry_attempts=1,
+            )
+            profile = vast_models_module.VastResourceProfile(
+                profile_id="private-image",
+                profile_name="private-image",
+            )
+            offer = (await client.search_offers(profile))[0]
+            launch = vast_models_module.VastInstanceLaunchSpec(
+                image="ghcr.io/example/private-worker:latest",
+                disk_gb=200,
+                label="managed-private-image",
+                onstart="start",
+                environment={},
+            )
+            created = await client.create_instance(offer.offer_id, launch)
+            raw_instance = state.instances[created.instance_id]
+            raw_instance["intended_status"] = "stopped"
+            raw_instance["cur_state"] = "stopped"
+            raw_instance["status_msg"] = "unauthorized: authentication required"
+
+            with pytest.raises(
+                vast_api_module.VastApiError,
+                match="unauthorized: authentication required",
+            ):
+                await client.wait_until_ready(
+                    created.instance_id,
+                    timeout_seconds=60.0,
+                    poll_interval_seconds=10.0,
+                )
+
+    asyncio.run(scenario())
+
+
 def test_simulator_enforces_capacity_price_and_location_filters(
     vast_api_module: Any,
     vast_models_module: Any,

tests/test_vast_models.py

@@ -178,3 +178,27 @@ def test_launch_payload_uses_direct_ssh_and_fail_fast(
     assert payload["cancel_unavail"] is True
     assert payload["target_state"] == "running"
     assert "price" not in payload
+
+
+def test_instance_normalizes_provider_progress_and_terminal_state(
+    vast_models_module: Any,
+) -> None:
+    """Vast's stopped image-pull state should be bounded, readable, and terminal."""
+    instance = vast_models_module.VastInstance.from_api(
+        {
+            "id": 42,
+            "actual_status": "loading",
+            "intended_status": "stopped",
+            "cur_state": "stopped",
+            "status_msg": " unauthorized:\n authentication required " + "x" * 300,
+            "num_gpus": 1,
+            "gpu_ram": 48 * 1024,
+            "cpu_ram": 96 * 1024,
+        }
+    )
+
+    assert instance.current_state == "stopped"
+    assert instance.status_message is not None
+    assert "\n" not in instance.status_message
+    assert len(instance.status_message) == 240
+    assert instance.terminal_failure is True

tests/test_vast_service.py

@@ -110,6 +110,32 @@ def test_from_environment_requires_credential_before_other_setup(
         raise AssertionError("Missing Vast credential was accepted.")
 
 
+def test_startup_status_translates_vast_image_progress(
+    vast_models_module: Any,
+    vast_service_module: Any,
+) -> None:
+    """Provider layer messages should become useful workflow status without hashes."""
+    instance = vast_models_module.VastInstance.from_api(
+        {
+            "id": 48534708,
+            "actual_status": "loading",
+            "intended_status": "running",
+            "cur_state": "running",
+            "status_msg": "4c7d45f7c63c: Download complete",
+            "num_gpus": 1,
+            "gpu_ram": 48 * 1024,
+            "cpu_ram": 96 * 1024,
+        }
+    )
+
+    message = vast_service_module._vast_startup_status_message(instance)
+
+    assert message == (
+        "Vast.ai instance 48534708 is downloading the worker image (layer complete)"
+    )
+    assert "4c7d45f7c63c" not in message
+
+
 def test_prefetch_deduplicates_effective_profiles_and_searches_in_parallel(
     vast_models_module: Any,
     vast_runtime_module: Any,

vast_api.py

@@ -336,6 +336,7 @@ class VastApiClient:
         *,
         timeout_seconds: float,
         poll_interval_seconds: float = 10.0,
+        status_callback: Callable[[VastInstance], None] | None = None,
     ) -> VastInstance:
         """Poll until one instance exposes a running SSH endpoint."""
         if timeout_seconds <= 0 or poll_interval_seconds <= 0:
@@ -345,12 +346,20 @@ class VastApiClient:
         last_instance: VastInstance | None = None
         while loop.time() < deadline:
             last_instance = await self.show_instance(instance_id)
+            if status_callback is not None:
+                status_callback(last_instance)
             if last_instance.ready_for_ssh:
                 return last_instance
-            if last_instance.actual_status in {"error", "exited", "destroyed"}:
+            if last_instance.terminal_failure:
+                provider_detail = (
+                    f" Provider status: {last_instance.status_message}."
+                    if last_instance.status_message
+                    else ""
+                )
                 raise VastApiError(
                     f"Vast instance {instance_id} entered terminal state "
                     f"{last_instance.actual_status!r} before SSH was ready."
+                    f"{provider_detail}"
                 )
             await asyncio.sleep(min(poll_interval_seconds, max(0.0, deadline - loop.time())))
         last_status = last_instance.actual_status if last_instance is not None else "unknown"

vast_leases.py

@@ -49,6 +49,7 @@ VAST_LEASE_REGISTRY_VERSION = 1
 VAST_LEASE_REGISTRY_FILENAME = "vast-leases.json"
 VAST_MANAGED_LABEL_PREFIX = "comfy-modal-vast"
 _SAFE_LABEL_PART_PATTERN = re.compile(r"[^a-zA-Z0-9._-]+")
+VastInstanceStatusCallback = Callable[[VastInstance], None]
 
 
 @dataclass(frozen=True)
@@ -350,6 +351,7 @@ class VastLeaseManager:
         profile: VastResourceProfile,
         *,
         slot: int = 0,
+        status_callback: VastInstanceStatusCallback | None = None,
     ) -> VastLeaseRecord:
         """Reuse or rent one compatible SSH-ready lease for a profile slot."""
         if slot < 0 or slot >= profile.maximum_instances:
@@ -359,15 +361,25 @@ class VastLeaseManager:
             )
         slot_profile_id = _slot_profile_id(profile.profile_id, slot)
         async with self._profile_lock(slot_profile_id):
-            existing = await self._reusable_lease(profile, slot_profile_id)
+            existing = await self._reusable_lease(
+                profile,
+                slot_profile_id,
+                status_callback=status_callback,
+            )
             if existing is not None:
                 return existing
-            return await self._rent_lease(profile, slot_profile_id)
+            return await self._rent_lease(
+                profile,
+                slot_profile_id,
+                status_callback=status_callback,
+            )
 
     async def _reusable_lease(
         self,
         profile: VastResourceProfile,
         slot_profile_id: str,
+        *,
+        status_callback: VastInstanceStatusCallback | None,
     ) -> VastLeaseRecord | None:
         """Return one matching live lease after refreshing its API state."""
         fingerprint = vast_profile_fingerprint(profile)
@@ -391,6 +403,7 @@ class VastLeaseManager:
                 instance = await self.api_client.wait_until_ready(
                     lease.instance_id,
                     timeout_seconds=self.startup_timeout_seconds,
+                    status_callback=status_callback,
                 )
             refreshed = self._refresh_record(lease, instance)
             self.registry.upsert(refreshed)
@@ -409,6 +422,8 @@ class VastLeaseManager:
         self,
         profile: VastResourceProfile,
         slot_profile_id: str,
+        *,
+        status_callback: VastInstanceStatusCallback | None,
     ) -> VastLeaseRecord:
         """Search, rent, and persist the first still-available compatible offer."""
         offers = await self.api_client.search_offers(profile)
@@ -426,6 +441,7 @@ class VastLeaseManager:
                         profile,
                         slot_profile_id,
                         offer,
+                        status_callback=status_callback,
                     )
                 except VastOfferUnavailableError:
                     unavailable_offer_ids.append(offer.offer_id)
@@ -449,6 +465,8 @@ class VastLeaseManager:
         profile: VastResourceProfile,
         slot_profile_id: str,
         offer: VastOffer,
+        *,
+        status_callback: VastInstanceStatusCallback | None,
     ) -> VastLeaseRecord:
         """Create, wait for, and persist one selected offer."""
         label = vast_managed_label(
@@ -487,6 +505,7 @@ class VastLeaseManager:
             instance = await self.api_client.wait_until_ready(
                 created.instance_id,
                 timeout_seconds=self.startup_timeout_seconds,
+                status_callback=status_callback,
             )
         except (TimeoutError, VastApiError) as exc:
             failed = replace(provisional, last_error=str(exc))

vast_models.py

@@ -326,6 +326,8 @@ class VastInstance:
     instance_id: int
     actual_status: str
     intended_status: str | None
+    current_state: str | None
+    status_message: str | None
     label: str | None
     ssh_host: str | None
     ssh_port: int | None
@@ -342,6 +344,8 @@ class VastInstance:
             instance_id=_required_int(payload, "id"),
             actual_status=str(payload.get("actual_status") or "unknown").strip(),
             intended_status=_optional_string(payload.get("intended_status")),
+            current_state=_optional_string(payload.get("cur_state")),
+            status_message=_bounded_status_string(payload.get("status_msg")),
             label=_optional_string(payload.get("label")),
             ssh_host=_optional_string(payload.get("ssh_host")),
             ssh_port=_optional_int(payload.get("ssh_port"), "ssh_port"),
@@ -364,6 +368,18 @@ class VastInstance:
             and self.ssh_port > 0
         )
 
+    @property
+    def terminal_failure(self) -> bool:
+        """Return whether provider state proves startup cannot continue."""
+        terminal_states = {"destroyed", "error", "exited"}
+        if self.actual_status.casefold() in terminal_states:
+            return True
+        current_state = (self.current_state or "").casefold()
+        intended_status = (self.intended_status or "").casefold()
+        if current_state in terminal_states:
+            return True
+        return current_state == "stopped" and intended_status == "stopped"
+
 
 @dataclass(frozen=True)
 class VastInstanceLaunchSpec:
@@ -497,6 +513,14 @@ def _optional_string(value: Any) -> str | None:
     return normalized or None
 
 
+def _bounded_status_string(value: Any) -> str | None:
+    """Return one single-line provider status suitable for logs and errors."""
+    normalized = _optional_string(value)
+    if normalized is None:
+        return None
+    return " ".join(normalized.split())[:240]
+
+
 def _require_positive_int(value: int, field_name: str) -> None:
     """Require a positive non-boolean integer."""
     if isinstance(value, bool) or not isinstance(value, int) or value <= 0:

vast_service.py

@@ -9,7 +9,7 @@ import math
 import os
 from dataclasses import dataclass, replace
 from pathlib import Path
-from typing import Mapping, Sequence
+from typing import Callable, Mapping, Sequence
 
 if __package__:
     from .execution_environments import (
@@ -30,7 +30,7 @@ if __package__:
         VastLeaseRegistry,
         vast_profile_fingerprint,
     )
-    from .vast_models import VastOffer, VastResourceProfile
+    from .vast_models import VastInstance, VastOffer, VastResourceProfile
     from .vast_runtime import VastRuntimeConfiguration, VastRuntimeManager
     from .vast_ssh import VastSshRunner, VastSshVolumeBackend, vast_connection_from_lease
 else:  # pragma: no cover - direct debugging imports.
@@ -52,7 +52,7 @@ else:  # pragma: no cover - direct debugging imports.
         VastLeaseRegistry,
         vast_profile_fingerprint,
     )
-    from vast_models import VastOffer, VastResourceProfile
+    from vast_models import VastInstance, VastOffer, VastResourceProfile
     from vast_runtime import VastRuntimeConfiguration, VastRuntimeManager
     from vast_ssh import VastSshRunner, VastSshVolumeBackend, vast_connection_from_lease
 
@@ -63,6 +63,21 @@ VAST_SSH_IDENTITY_FILE_ENV = "COMFY_MODAL_VAST_SSH_IDENTITY_FILE"
 VAST_OFFER_PREFETCH_CONCURRENCY = 8
 
 
+def _vast_startup_status_message(instance: VastInstance) -> str:
+    """Translate one provider lifecycle record into stable user-facing progress."""
+    provider_message = (instance.status_message or "").casefold()
+    instance_label = f"Vast.ai instance {instance.instance_id}"
+    if "unauthorized" in provider_message or "denied" in provider_message:
+        return f"{instance_label} could not download the worker image"
+    if "download complete" in provider_message:
+        return f"{instance_label} is downloading the worker image (layer complete)"
+    if any(word in provider_message for word in ("download", "pull", "extract")):
+        return f"{instance_label} is downloading the worker image"
+    if instance.actual_status.casefold() == "running":
+        return f"{instance_label} is waiting for SSH access"
+    return f"{instance_label} is starting ({instance.actual_status})"
+
+
 @dataclass(frozen=True)
 class VastProfileQuote:
     """Describe the least effective-hourly-price profile for one component."""
@@ -355,10 +370,29 @@ class VastService:
         quote: VastProfileQuote,
         *,
         slot: int = 0,
+        status_callback: Callable[[str], None] | None = None,
     ) -> VastLeaseRecord:
         """Reuse or rent the selected quoted profile capacity slot."""
-        lease = await self.lease_manager.ensure_lease(quote.profile, slot=slot)
+        if status_callback is not None:
+            status_callback("Requesting Vast.ai capacity")
+
+        def emit_instance_status(instance: VastInstance) -> None:
+            """Forward normalized provider progress when a caller is listening."""
+            if status_callback is not None:
+                status_callback(_vast_startup_status_message(instance))
+
+        lease = await self.lease_manager.ensure_lease(
+            quote.profile,
+            slot=slot,
+            status_callback=(
+                emit_instance_status if status_callback is not None else None
+            ),
+        )
+        if status_callback is not None:
+            status_callback("Initializing Vast.ai worker")
         await asyncio.to_thread(self._initialize_runtime, lease)
+        if status_callback is not None:
+            status_callback("Vast.ai worker is ready")
         return lease
 
     def quote_best_profile_sync(self, *args: object, **kwargs: object) -> VastProfileQuote:
@@ -378,9 +412,16 @@ class VastService:
         quote: VastProfileQuote,
         *,
         slot: int = 0,
+        status_callback: Callable[[str], None] | None = None,
     ) -> VastLeaseRecord:
         """Synchronously acquire a quote slot from ComfyUI's queue worker thread."""
-        return asyncio.run(self.acquire(quote, slot=slot))
+        return asyncio.run(
+            self.acquire(
+                quote,
+                slot=slot,
+                status_callback=status_callback,
+            )
+        )
 
     def scheduling_state(self, quote: VastProfileQuote) -> EnvironmentSchedulingState:
         """Expose a quote to the provider-neutral cost scheduler."""

vast_simulator.py

@@ -217,6 +217,7 @@ class VastApiSimulator:
             "actual_status": "loading",
             "intended_status": "running",
             "cur_state": "loading",
+            "status_msg": "worker-layer: Download complete",
             "next_state": "running",
             "label": body.get("label"),
             "image_uuid": body.get("image"),
@@ -250,6 +251,7 @@ class VastApiSimulator:
         ):
             instance["actual_status"] = "running"
             instance["cur_state"] = "running"
+            instance["status_msg"] = "Worker ready"
         return web.json_response({"instances": _public_instance(instance)})
 
     async def list_instances(self, request: web.Request) -> web.Response: