ttulttul-ComfyUI-Modal/remote/modal_app.py

"""Remote Modal runtime and local execution fallback."""

import asyncio
import copy
from dataclasses import dataclass, field, replace
from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
import hashlib
import importlib
import importlib.util
import inspect
from io import BytesIO
import json
import logging
import os
import queue
import select
import shutil
import socket
import statistics
import subprocess
import sys
import tempfile
import threading
import time
from types import ModuleType
import uuid
import zipfile
from concurrent.futures import (
    Future,
    ThreadPoolExecutor,
    TimeoutError as FutureTimeoutError,
)
from contextlib import contextmanager, nullcontext
from functools import lru_cache
from pathlib import Path
from typing import Any, Callable, Iterable, Iterator, Mapping

from ..serialization import (
    coerce_serialized_node_outputs,
    deserialize_value,
    deserialize_node_inputs,
    deserialize_node_outputs,
    is_mapped_output_value,
    join_mapped_values_for_scheduler,
    serialize_mapping,
    serialize_node_inputs,
    serialize_node_outputs,
    serialize_value,
    split_mapped_value,
    unwrap_mapped_output_value,
)
from ..durable_state import (
    DurableObjectRef,
    DurableStateError,
    FileDurableObjectStore,
    read_modal_volume_file,
    stable_remote_invocation_id,
)
from ..output_artifacts import (
    RemoteExecutionResult,
    materialize_remote_output_artifacts,
    unpack_remote_execution_result,
)
from ..session_state import (
    InMemoryRemoteSessionBridgeStore,
    InMemoryRemoteSessionStore,
    RemoteSessionBridgeRecord,
    RemoteSessionBridgeRecoveryKind,
    RemoteSessionBridgeRef,
    RemoteSessionHandle,
    RemoteSessionStateError,
    RemoteSessionValueRef,
    is_remote_session_bridge_ref_payload,
    is_remote_session_handle_payload,
    is_remote_session_value_ref_payload,
    stable_session_bridge_key,
)
from ..runtime_environment import (
    REMOTE_APP_PROTOCOL_VERSION,
    RemoteRuntimeIdentity,
    build_remote_runtime_identity,
)
from ..llm_profiles import (
    get_llm_profile,
    llm_model_references_from_payload,
    rewrite_llm_model_references,
)
from ..llm_recovery import (
    LLM_FORCE_VLLM_THROUGHPUT_PAYLOAD_KEY,
    exhausted_recovery_used_vllm_throughput,
    is_llm_memory_recovery_exhausted,
)
from ..settings import (
    MODAL_GPU_TYPES,
    ModalSyncSettings,
    get_settings,
    modal_deployment_app_name,
    settings_for_modal_gpu,
)

logger = logging.getLogger(__name__)
_MODAL_CLOUD_MODULE_NAME = "comfyui_modal_sync_cloud"
_REMOTE_APP_PROTOCOL_VERSION = REMOTE_APP_PROTOCOL_VERSION
_MODAL_CLOUD_MODULE_LOCK = threading.Lock()
_MODAL_CLOUD_SETTINGS_STATE = threading.local()
_MODAL_AUTO_DEPLOY_LOCK = threading.Lock()
_MODAL_AUTO_DEPLOY_STATES: dict[tuple[str, str | None], "_ModalAutoDeployState"] = {}
_MODAL_REMOTE_APP_VERSION_OK: set[tuple[str, str | None, str]] = set()
_STAGED_LLM_PROFILES_LOCK = threading.Lock()
_STAGED_LLM_PROFILES: set[tuple[str, str, str]] = set()
_STAGED_LLM_PROFILE_RESULTS: dict[tuple[str, str], dict[str, Any]] = {}
_MODAL_APP_STOP_TIMEOUT_SECONDS = 120.0
_MODAL_INTERRUPT_DICTS_LOCK = threading.Lock()
_MODAL_INTERRUPT_DICTS: dict[tuple[str, str | None], Any] = {}
_MODAL_SESSION_BRIDGE_DICTS_LOCK = threading.Lock()
_MODAL_SESSION_BRIDGE_DICTS: dict[tuple[str, str | None], Any] = {}
_MODAL_DURABLE_VOLUMES_LOCK = threading.Lock()
_MODAL_DURABLE_VOLUMES: dict[tuple[str, str | None], Any] = {}
_MAPPED_PROGRESS_NODE_IDS_LOCK = threading.Lock()
_MAPPED_PROGRESS_NODE_IDS: dict[tuple[str, str, str], str] = {}
_PROMPT_WARMUP_STATES_LOCK = threading.Lock()
_PROMPT_WARMUP_STATES: dict[str, "_PromptWarmupState"] = {}
_PROMPT_WARMUP_STATE_ORDER: queue.SimpleQueue[str] | None = None
_PROMPT_WARMUP_STATE_CACHE_LIMIT = 256
_LOCAL_GAP_KEEPALIVES_LOCK = threading.Lock()
_LOCAL_GAP_KEEPALIVES: dict[tuple[str, int], "_LocalGapKeepaliveState"] = {}
_SNAPSHOT_PROFILE_RECORDS_LOCK = threading.Lock()
_SNAPSHOT_PROFILE_RECORDS: dict[str, dict[str, Any]] = {}
_PRIMITIVE_WIDGET_INPUT_TYPES = frozenset({"INT", "FLOAT", "BOOLEAN", "STRING"})
_ROOT_LOADER_PREWARM_CLASS_TYPES = frozenset(
    {
        "CheckpointLoaderSimple",
        "UNETLoader",
        "CLIPLoader",
        "DualCLIPLoader",
    }
)
_SPECULATIVE_PREWARM_TARGET_KEY = "speculative_remote_prewarm_target"
_BOUNDARY_INPUT_SIGNATURES_KEY = "__comfy_modal_boundary_input_signatures__"
_REMOTE_CONTAINER_LOG_STREAMS_LOCK = threading.Lock()
_REMOTE_CONTAINER_LOG_STREAMS: dict[str, "_RemoteContainerLogStreamState"] = {}
_REMOTE_CONTAINER_LOG_STDERR_LOCK = threading.Lock()
_REMOTE_CONTAINER_LOG_STREAM_IDLE_GRACE_SECONDS = 30.0
_MODAL_BILLING_COLLECTION_BUFFER = timedelta(minutes=10)
_MODAL_BILLING_INTERVAL = timedelta(hours=1)
_MODAL_HOURLY_BILLING_CACHE_LOCK = threading.Lock()
_MODAL_HOURLY_BILLING_CACHE: dict[
    tuple[str, str, datetime], "ModalHourlyBillingStatus"
] = {}
_MODAL_HOURLY_BILLING_ERROR_CACHE: dict[tuple[str, str, datetime], str] = {}
_MODAL_HOURLY_BILLING_CACHE_LIMIT = 64
_IMPLICIT_BATCH_PRESERVING_TARGETS = frozenset({("CreateVideo", "images")})
_ACTIVE_REMOTE_INVOCATIONS_LOCK = threading.Lock()
_ACTIVE_REMOTE_INVOCATIONS_BY_PROMPT: dict[
    str, dict[str, "_ActiveRemoteInvocation"]
] = {}
_REMOTE_SESSION_STORE = InMemoryRemoteSessionStore()
_REMOTE_SESSION_BRIDGE_STORE = InMemoryRemoteSessionBridgeStore()
_DURABLE_OBJECT_STORE_LOCK = threading.Lock()
_DURABLE_OBJECT_STORE: FileDurableObjectStore | None = None
_REMOTE_SESSION_BRIDGE_REPLAY_STATE = threading.local()
_REMOTE_SESSION_BRIDGE_VALUE_CACHE_LOCK = threading.Lock()
_REMOTE_SESSION_BRIDGE_VALUE_CACHE: dict[str, Any] = {}
_REMOTE_SESSION_BRIDGE_VALUE_CACHE_ORDER: list[str] = []
_REMOTE_SESSION_BRIDGE_VALUE_CACHE_LIMIT = 32
MODAL_GPU_PRICING_EFFECTIVE_DATE = "2026-08-13"
MODAL_GPU_ESTIMATED_USD_PER_SECOND: dict[str, float] = {
    "T4": 0.000164,
    "L4": 0.000222,
    "A10": 0.000306,
    "L40S": 0.000542,
    "A100": 0.000583,
    "A100-40GB": 0.000583,
    "A100-80GB": 0.000694,
    "RTX-PRO-6000": 0.000842,
    "H100": 0.001097,
    "H100!": 0.001097,
    "H200": 0.001261,
    "B200": 0.001736,
    "B200+": 0.001736,
    "B300": 0.001972,
}


def _settings_for_payload(payload: Mapping[str, Any]) -> ModalSyncSettings:
    """Resolve settings with the workflow-selected GPU carried by one payload."""
    settings = get_settings()
    modal_gpu = payload.get("modal_gpu")
    if modal_gpu is not None:
        settings = settings_for_modal_gpu(settings, modal_gpu)
    raw_max_containers = payload.get("modal_max_containers")
    if raw_max_containers is None:
        return settings
    max_containers = int(raw_max_containers)
    if max_containers <= 0:
        raise ValueError("modal_max_containers must be positive.")
    return replace(settings, max_containers=max_containers)


@contextmanager
def _modal_cloud_settings_override(settings: ModalSyncSettings) -> Iterator[None]:
    """Expose request-specific settings while constructing the deployable cloud module."""
    previous_settings = getattr(_MODAL_CLOUD_SETTINGS_STATE, "settings", None)
    _MODAL_CLOUD_SETTINGS_STATE.settings = settings
    try:
        yield
    finally:
        if previous_settings is None:
            try:
                delattr(_MODAL_CLOUD_SETTINGS_STATE, "settings")
            except AttributeError:
                pass
        else:
            _MODAL_CLOUD_SETTINGS_STATE.settings = previous_settings


_DURABLE_BRIDGE_SERIALIZATION_IO_TYPES = frozenset(
    {
        "AUDIO",
        "BOOLEAN",
        "CONDITIONING",
        "FLOAT",
        "IMAGE",
        "INT",
        "LATENT",
        "MASK",
        "SIGMAS",
        "STRING",
        "VIDEO",
    }
)
_DURABLE_BRIDGE_REHYDRATION_IO_TYPES = frozenset(
    {"CLIP", "MODEL", "NOISE", "SAMPLER", "VAE"}
)


@dataclass
class _PromptWarmupState:
    """Track proactive warmup state for one local prompt."""

    scheduled_slots: set[int] = field(default_factory=set)
    exact_component_parallelism: dict[str, int] = field(default_factory=dict)
    slot_futures: dict[int, Future[Any]] = field(default_factory=dict)
    scheduled_speculative_affinities: set[str] = field(default_factory=set)
    speculative_affinity_futures: dict[str, Future[Any]] = field(default_factory=dict)


@dataclass
class _LocalGapKeepaliveState:
    """Track one prompt-scoped worker-retention loop during local execution."""

    component_id: str
    stop_event: threading.Event
    future: Future[Any]


@dataclass
class _ModalAutoDeployState:
    """Track one thread-safe deployed-app readiness lifecycle."""

    condition: threading.Condition = field(default_factory=threading.Condition)
    deploy_in_progress: bool = False
    ready: bool = False
    last_error: BaseException | None = None


@dataclass
class _RemoteContainerLogLineBuffer:
    """Buffer partial remote log lines so stderr mirroring stays line-oriented."""

    task_id: str
    buffered_text: str = ""


@dataclass
class _RemoteContainerLogStreamState:
    """Track one active local watcher mirroring logs for a Modal container."""

    task_id: str
    stop_event: threading.Event
    thread: threading.Thread
    refcount: int = 0
    idle_stop_timer: threading.Timer | None = None


@dataclass
class _ActiveRemoteInvocation:
    """Track one local proxy call that is currently waiting on remote Modal work."""

    prompt_id: str
    component_id: str
    cancellation_event: threading.Event | None
    interrupt_remote_call: Callable[[], Any] | None


@dataclass
class _RemoteSessionBridgeResolutionStats:
    """Track how one local fallback payload resolved session-backed inputs."""

    input_ref_count: int = 0
    live_session_hits: int = 0
    bridge_cache_hits: int = 0
    durable_bridge_hits: int = 0
    bridge_record_lookups: int = 0
    bridge_record_lookup_seconds: float = 0.0
    replay_count: int = 0
    replay_seconds: float = 0.0
    direct_restore_seconds: float = 0.0
    session_restore_writes: int = 0


def _remote_modal_call_worker_count() -> int:
    """Return the number of local worker threads reserved for blocking Modal calls."""
    return max(1, int(get_settings().max_inflight_calls))


_REMOTE_MODAL_CALL_EXECUTOR = ThreadPoolExecutor(
    max_workers=_remote_modal_call_worker_count()
)
_REMOTE_MODAL_WARMUP_EXECUTOR = ThreadPoolExecutor(
    max_workers=_remote_modal_call_worker_count()
)
_REMOTE_MODAL_KEEPALIVE_EXECUTOR = ThreadPoolExecutor(
    max_workers=_remote_modal_call_worker_count()
)

try:
    import modal  # type: ignore
except ModuleNotFoundError:  # pragma: no cover - exercised by local fallback tests.
    modal = None


class RemoteSubgraphExecutionError(RuntimeError):
    """Raised when remote subgraph execution fails."""


class ModalRemoteInvocationError(RuntimeError):
    """Raised when the Modal client cannot invoke the remote runtime."""


class ModalRemoteAppOutOfDateError(ModalRemoteInvocationError):
    """Raised when a deployed Modal app is incompatible with the local client."""


class ModalContainerStatusError(RuntimeError):
    """Raised when active Modal container status cannot be queried."""


class ModalBillingStatusError(RuntimeError):
    """Raised when Modal hourly billing data cannot be queried."""


@dataclass(frozen=True)
class ModalContainerStatus:
    """Describe one active Modal container owned by this ComfyUI instance."""

    container_id: str
    app_id: str
    app_name: str
    modal_gpu: str
    estimated_gpu_cost_per_second: float
    state: str
    enqueued_at: float | None
    started_at: float | None

    def as_dict(self) -> dict[str, str | float | None]:
        """Return a JSON-serializable representation for the frontend."""
        return {
            "container_id": self.container_id,
            "app_id": self.app_id,
            "app_name": self.app_name,
            "modal_gpu": self.modal_gpu,
            "estimated_gpu_cost_per_second": self.estimated_gpu_cost_per_second,
            "state": self.state,
            "enqueued_at": self.enqueued_at,
            "started_at": self.started_at,
        }


@dataclass(frozen=True)
class ModalHourlyBillingStatus:
    """Describe one completed hourly billing interval for a GPU-specific app."""

    app_id: str | None
    app_name: str
    environment_name: str
    modal_gpu: str
    interval_start: datetime
    interval_end: datetime
    app_cost_usd_before_credits: Decimal
    has_usage: bool
    fetched_at: datetime
    next_refresh_at: datetime

    def as_dict(self) -> dict[str, str | float | bool | None]:
        """Return a JSON-serializable representation for the frontend."""
        return {
            "app_id": self.app_id,
            "app_name": self.app_name,
            "environment_name": self.environment_name,
            "modal_gpu": self.modal_gpu,
            "resolution": "hour",
            "interval_start": self.interval_start.isoformat(),
            "interval_end": self.interval_end.isoformat(),
            "app_cost_usd_before_credits": float(
                self.app_cost_usd_before_credits
            ),
            "has_usage": self.has_usage,
            "fetched_at": self.fetched_at.isoformat(),
            "next_refresh_at": self.next_refresh_at.isoformat(),
            "collection_buffer_seconds": int(
                _MODAL_BILLING_COLLECTION_BUFFER.total_seconds()
            ),
        }


def _is_remote_container_log_stream_enabled() -> bool:
    """Return whether remote Modal container logs should be mirrored locally."""
    return bool(get_settings().stream_remote_container_logs)


def _coerce_modal_task_id(value: Any) -> str | None:
    """Normalize one streamed Modal task id into a non-empty string."""
    if value is None:
        return None
    task_id = str(value).strip()
    return task_id or None


def _write_remote_container_log_line(
    task_id: str,
    line: str,
    *,
    stream: Any = None,
) -> None:
    """Write one complete remote container log line to the local stderr stream."""
    target_stream = stream if stream is not None else sys.stderr
    with _REMOTE_CONTAINER_LOG_STDERR_LOCK:
        target_stream.write(f"[modal:{task_id}] {line}")
        target_stream.flush()


def _write_remote_container_log_chunk(
    line_buffer: _RemoteContainerLogLineBuffer,
    chunk: str,
    *,
    stream: Any = None,
) -> None:
    """Buffer one remote log chunk and mirror complete lines to local stderr."""
    line_buffer.buffered_text += chunk
    while True:
        newline_index = line_buffer.buffered_text.find("\n")
        if newline_index < 0:
            return
        next_line = line_buffer.buffered_text[: newline_index + 1]
        line_buffer.buffered_text = line_buffer.buffered_text[newline_index + 1 :]
        _write_remote_container_log_line(line_buffer.task_id, next_line, stream=stream)


def _flush_remote_container_log_chunk(
    line_buffer: _RemoteContainerLogLineBuffer,
    *,
    stream: Any = None,
) -> None:
    """Flush any partial remote log text remaining in the local line buffer."""
    if not line_buffer.buffered_text:
        return
    _write_remote_container_log_line(
        line_buffer.task_id,
        f"{line_buffer.buffered_text}\n",
        stream=stream,
    )
    line_buffer.buffered_text = ""


def _managed_modal_app_gpus(settings: ModalSyncSettings) -> dict[str, str]:
    """Map every GPU-specific app name owned by this ComfyUI instance to its GPU."""
    return {
        modal_deployment_app_name(
            settings_for_modal_gpu(settings, modal_gpu)
        ): modal_gpu
        for modal_gpu in MODAL_GPU_TYPES
    }


def _optional_modal_timestamp(value: Any) -> float | None:
    """Normalize an unset Modal protobuf timestamp to ``None``."""
    timestamp = float(value or 0.0)
    return timestamp if timestamp > 0 else None


def _completed_modal_billing_interval(
    now: datetime,
) -> tuple[datetime, datetime, datetime]:
    """Return the buffered completed hour and its next eligible refresh time."""
    if now.tzinfo is None:
        normalized_now = now.replace(tzinfo=timezone.utc)
    else:
        normalized_now = now.astimezone(timezone.utc)
    buffered_now = normalized_now - _MODAL_BILLING_COLLECTION_BUFFER
    interval_end = buffered_now.replace(minute=0, second=0, microsecond=0)
    interval_start = interval_end - _MODAL_BILLING_INTERVAL
    next_refresh_at = (
        interval_end
        + _MODAL_BILLING_INTERVAL
        + _MODAL_BILLING_COLLECTION_BUFFER
    )
    return interval_start, interval_end, next_refresh_at


def _modal_billing_row_value(row: Any, key: str) -> Any:
    """Read one billing row field from a mapping or SDK dataclass."""
    if isinstance(row, Mapping):
        return row.get(key)
    return getattr(row, key, None)


def _modal_billing_cost(value: Any) -> Decimal:
    """Normalize one Modal billing cost without losing decimal precision."""
    try:
        return Decimal(str(value))
    except (InvalidOperation, TypeError, ValueError) as exc:
        raise ModalBillingStatusError(
            f"Modal returned an invalid billing cost {value!r}."
        ) from exc


def _resolved_modal_environment_name() -> str | None:
    """Return the explicitly selected Modal environment, if one is configured."""
    object_module = importlib.import_module("modal._object")
    environments_module = importlib.import_module("modal.environments")
    environment = environments_module.ensure_env(_modal_environment_name())
    environment_name = object_module._get_environment_name(environment)
    if environment_name is None:
        return None
    normalized_environment_name = str(environment_name).strip()
    return normalized_environment_name or None


@lru_cache(maxsize=None)
def _synchronized_modal_callable(
    async_callable: Callable[..., Any],
) -> Callable[..., Any]:
    """Wrap one async Modal SDK operation on Modal's managed event loop."""
    async_utils_module = importlib.import_module("modal._utils.async_utils")
    return async_utils_module.synchronize_api(async_callable)


async def _list_modal_tasks_on_sdk_loop(
    client_module: Any,
    api_pb2: Any,
    environment_name: str | None,
) -> Any:
    """List Modal tasks while running on Modal's managed event loop."""
    client = await client_module._Client.from_env()
    return await client.stub.TaskList(
        api_pb2.TaskListRequest(environment_name=environment_name)
    )


def _list_modal_tasks_synchronously(
    client_module: Any,
    api_pb2: Any,
    environment_name: str | None,
) -> Any:
    """Run one Modal task-list query through the synchronized SDK bridge."""
    blocking_callable = _synchronized_modal_callable(_list_modal_tasks_on_sdk_loop)
    return blocking_callable(client_module, api_pb2, environment_name)


def _prune_modal_hourly_billing_cache() -> None:
    """Bound retained hourly billing records across apps and GPU selections."""
    retained_keys = set(_MODAL_HOURLY_BILLING_CACHE) | set(
        _MODAL_HOURLY_BILLING_ERROR_CACHE
    )
    overflow = len(retained_keys) - _MODAL_HOURLY_BILLING_CACHE_LIMIT
    if overflow <= 0:
        return
    oldest_keys = sorted(
        retained_keys,
        key=lambda cache_key: cache_key[2],
    )[:overflow]
    for cache_key in oldest_keys:
        _MODAL_HOURLY_BILLING_CACHE.pop(cache_key, None)
        _MODAL_HOURLY_BILLING_ERROR_CACHE.pop(cache_key, None)


def _matching_modal_hourly_billing_rows(
    rows: Any,
    *,
    app_name: str,
    environment_name: str | None,
    interval_start: datetime,
) -> tuple[list[Any], str]:
    """Select every historical app identity from one billing environment."""
    candidate_rows = [
        row
        for row in rows
        if str(_modal_billing_row_value(row, "description") or "") == app_name
        and _modal_billing_row_value(row, "interval_start") == interval_start
    ]
    if environment_name is not None:
        matching_rows = [
            row
            for row in candidate_rows
            if str(_modal_billing_row_value(row, "environment_name") or "")
            == environment_name
        ]
        return matching_rows, environment_name

    candidate_environment_names = {
        str(_modal_billing_row_value(row, "environment_name") or "").strip()
        for row in candidate_rows
    }
    if len(candidate_environment_names) > 1:
        raise ModalBillingStatusError(
            f"Modal returned billing for app {app_name!r} in multiple "
            "environments; set MODAL_ENVIRONMENT to select one."
        )
    reported_environment_name = next(
        (
            str(_modal_billing_row_value(row, "environment_name") or "").strip()
            for row in candidate_rows
            if str(_modal_billing_row_value(row, "environment_name") or "").strip()
        ),
        "<workspace default>",
    )
    return candidate_rows, reported_environment_name


def _modal_hourly_billing_status_from_rows(
    rows: Any,
    *,
    app_name: str,
    environment_name: str | None,
    modal_gpu: str,
    interval_start: datetime,
    interval_end: datetime,
    fetched_at: datetime,
    next_refresh_at: datetime,
) -> ModalHourlyBillingStatus:
    """Build one app billing status from an hourly workspace report."""
    matching_rows, reported_environment_name = _matching_modal_hourly_billing_rows(
        rows,
        app_name=app_name,
        environment_name=environment_name,
        interval_start=interval_start,
    )
    app_cost = sum(
        (
            _modal_billing_cost(_modal_billing_row_value(row, "cost"))
            for row in matching_rows
        ),
        start=Decimal("0"),
    )
    app_ids = {
        str(app_id)
        for row in matching_rows
        if (app_id := _modal_billing_row_value(row, "object_id"))
    }
    return ModalHourlyBillingStatus(
        app_id=next(iter(app_ids)) if len(app_ids) == 1 else None,
        app_name=app_name,
        environment_name=reported_environment_name,
        modal_gpu=modal_gpu,
        interval_start=interval_start,
        interval_end=interval_end,
        app_cost_usd_before_credits=app_cost,
        has_usage=bool(matching_rows),
        fetched_at=fetched_at,
        next_refresh_at=next_refresh_at,
    )


def _fetch_modal_hourly_billing_synchronously(
    report: Callable[..., Any],
    error_types: tuple[type[BaseException], ...],
    *,
    app_name: str,
    environment_name: str | None,
    modal_gpu: str,
    interval_start: datetime,
    interval_end: datetime,
    fetched_at: datetime,
    next_refresh_at: datetime,
) -> ModalHourlyBillingStatus:
    """Fetch and cache one hourly report while serializing concurrent UI polls."""
    cache_environment_name = environment_name or "<workspace default>"
    cache_key = (cache_environment_name, app_name, interval_start)
    with _MODAL_HOURLY_BILLING_CACHE_LOCK:
        cached_status = _MODAL_HOURLY_BILLING_CACHE.get(cache_key)
        if cached_status is not None:
            return cached_status
        cached_error = _MODAL_HOURLY_BILLING_ERROR_CACHE.get(cache_key)
        if cached_error is not None:
            raise ModalBillingStatusError(cached_error)

        try:
            rows = report(
                start=interval_start,
                end=interval_end,
                resolution="h",
            )
            status = _modal_hourly_billing_status_from_rows(
                rows,
                app_name=app_name,
                environment_name=environment_name,
                modal_gpu=modal_gpu,
                interval_start=interval_start,
                interval_end=interval_end,
                fetched_at=fetched_at,
                next_refresh_at=next_refresh_at,
            )
        except error_types as exc:
            cached_error = str(exc)
            _MODAL_HOURLY_BILLING_ERROR_CACHE[cache_key] = cached_error
            _prune_modal_hourly_billing_cache()
            raise ModalBillingStatusError(cached_error) from exc
        _MODAL_HOURLY_BILLING_CACHE[cache_key] = status
        _prune_modal_hourly_billing_cache()
        return status


async def get_hourly_modal_app_billing(
    modal_gpu: str,
    settings: ModalSyncSettings | None = None,
    *,
    now: datetime | None = None,
) -> ModalHourlyBillingStatus:
    """Return cached billing for the latest buffered hour of one GPU app."""
    if modal is None:
        raise ModalBillingStatusError("The Modal SDK is unavailable.")

    resolved_settings = settings_for_modal_gpu(settings or get_settings(), modal_gpu)
    app_name = modal_deployment_app_name(resolved_settings)
    requested_now = now or datetime.now(timezone.utc)
    fetched_at = (
        requested_now.replace(tzinfo=timezone.utc)
        if requested_now.tzinfo is None
        else requested_now.astimezone(timezone.utc)
    )
    interval_start, interval_end, next_refresh_at = _completed_modal_billing_interval(
        fetched_at
    )
    try:
        environment_name = _resolved_modal_environment_name()
        billing_module = importlib.import_module("modal.billing")
        exception_module = importlib.import_module("modal.exception")
    except (ModuleNotFoundError, AttributeError, RuntimeError) as exc:
        raise ModalBillingStatusError(
            "The installed Modal SDK does not expose hourly billing reports."
        ) from exc

    report = getattr(billing_module, "workspace_billing_report", None)
    if not callable(report):
        raise ModalBillingStatusError(
            "The installed Modal SDK does not expose workspace_billing_report()."
        )
    modal_error_type = getattr(exception_module, "Error", RuntimeError)
    billing_error_types = (
        modal_error_type,
        ModalBillingStatusError,
        OSError,
        AttributeError,
        RuntimeError,
        TypeError,
        ValueError,
    )
    try:
        status = await asyncio.to_thread(
            _fetch_modal_hourly_billing_synchronously,
            report,
            billing_error_types,
            app_name=app_name,
            environment_name=environment_name,
            modal_gpu=resolved_settings.modal_gpu,
            interval_start=interval_start,
            interval_end=interval_end,
            fetched_at=fetched_at,
            next_refresh_at=next_refresh_at,
        )
    except billing_error_types as exc:
        raise ModalBillingStatusError(
            f"Unable to fetch Modal hourly billing: {exc}"
        ) from exc
    return status


async def list_active_modal_containers(
    settings: ModalSyncSettings | None = None,
) -> list[ModalContainerStatus]:
    """List active Modal containers belonging to this ComfyUI instance."""
    if modal is None:
        raise ModalContainerStatusError("The Modal SDK is unavailable.")

    resolved_settings = settings or get_settings()
    managed_app_gpus = _managed_modal_app_gpus(resolved_settings)
    try:
        client_module = importlib.import_module("modal.client")
        exception_module = importlib.import_module("modal.exception")
        api_pb2 = importlib.import_module("modal_proto.api_pb2")
    except ModuleNotFoundError as exc:
        raise ModalContainerStatusError(
            "The installed Modal SDK does not expose the container list API."
        ) from exc

    modal_error_type = getattr(exception_module, "Error", RuntimeError)
    try:
        environment_name = _resolved_modal_environment_name()
        response = await asyncio.to_thread(
            _list_modal_tasks_synchronously,
            client_module,
            api_pb2,
            environment_name,
        )
    except (modal_error_type, OSError, AttributeError, RuntimeError) as exc:
        raise ModalContainerStatusError(
            f"Unable to list Modal containers: {exc}"
        ) from exc

    containers: list[ModalContainerStatus] = []
    for task in response.tasks:
        app_name = str(task.app_description)
        modal_gpu = managed_app_gpus.get(app_name)
        if modal_gpu is None:
            continue
        started_at = _optional_modal_timestamp(task.started_at)
        containers.append(
            ModalContainerStatus(
                container_id=str(task.task_id),
                app_id=str(task.app_id),
                app_name=app_name,
                modal_gpu=modal_gpu,
                estimated_gpu_cost_per_second=MODAL_GPU_ESTIMATED_USD_PER_SECOND[
                    modal_gpu
                ],
                state="running" if started_at is not None else "starting",
                enqueued_at=_optional_modal_timestamp(task.enqueued_at),
                started_at=started_at,
            )
        )

    containers.sort(
        key=lambda container: (
            container.enqueued_at or container.started_at or 0.0,
            container.container_id,
        )
    )
    logger.debug(
        "Listed %d active Modal container(s) for the status UI.", len(containers)
    )
    return containers


async def _stream_remote_container_logs_via_modal_sdk_async(
    task_id: str,
    stop_event: threading.Event,
) -> None:
    """Follow one Modal container log stream through the Python SDK internals."""
    if modal is None:
        raise ModuleNotFoundError("Modal SDK is unavailable.")

    client_module = importlib.import_module("modal.client")
    exception_module = importlib.import_module("modal.exception")
    api_pb2 = importlib.import_module("modal_proto.api_pb2")
    grpclib_exceptions = importlib.import_module("grpclib.exceptions")
    client = await client_module._Client.from_env()
    line_buffer = _RemoteContainerLogLineBuffer(task_id=task_id)
    last_entry_id = ""

    try:
        while not stop_event.is_set():
            request = api_pb2.AppGetLogsRequest(
                task_id=task_id,
                timeout=5,
                last_entry_id=last_entry_id,
            )
            try:
                async for log_batch in client.stub.AppGetLogs.unary_stream(request):
                    if stop_event.is_set():
                        break
                    if log_batch.entry_id:
                        last_entry_id = str(log_batch.entry_id)
                    if bool(log_batch.app_done):
                        logger.info(
                            "Modal SDK log stream finished for task_id=%s.", task_id
                        )
                        return
                    for log_item in log_batch.items:
                        log_data = getattr(log_item, "data", "")
                        if not log_data:
                            continue
                        _write_remote_container_log_chunk(line_buffer, str(log_data))
            except (
                exception_module.ServiceError,
                exception_module.InternalError,
                grpclib_exceptions.StreamTerminatedError,
                socket.gaierror,
            ) as exc:
                if stop_event.is_set():
                    break
                logger.warning(
                    "Retrying Modal SDK log stream for task_id=%s after transient failure: %s",
                    task_id,
                    exc,
                )
                continue
            except AttributeError as exc:
                if stop_event.is_set():
                    break
                if "_write_appdata" in str(exc):
                    logger.warning(
                        "Retrying Modal SDK log stream for task_id=%s after connection loss: %s",
                        task_id,
                        exc,
                    )
                    continue
                raise
    finally:
        _flush_remote_container_log_chunk(line_buffer)


def _stream_remote_container_logs_via_modal_sdk(
    task_id: str,
    stop_event: threading.Event,
) -> bool:
    """Try to mirror one Modal container log stream through the installed SDK."""
    if modal is None:
        return False
    blocking_callable = _synchronized_modal_callable(
        _stream_remote_container_logs_via_modal_sdk_async
    )
    blocking_callable(task_id, stop_event)
    return True


def _stream_remote_container_logs_via_modal_cli(
    task_id: str,
    stop_event: threading.Event,
) -> bool:
    """Try to mirror one Modal container log stream through the Modal CLI."""
    modal_cli = shutil.which("modal")
    if modal_cli is None:
        return False

    command = [modal_cli, "container", "logs", task_id, "-f"]
    line_buffer = _RemoteContainerLogLineBuffer(task_id=task_id)
    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        bufsize=0,
    )
    line_buffer = _RemoteContainerLogLineBuffer(task_id=task_id)
    try:
        stdout_stream = process.stdout
        if stdout_stream is None:
            raise RuntimeError("Modal CLI log process did not expose a stdout stream.")
        while not stop_event.is_set():
            ready_streams, _, _ = select.select([stdout_stream], [], [], 0.25)
            if ready_streams:
                next_chunk = stdout_stream.read(4096)
                if next_chunk:
                    _write_remote_container_log_chunk(
                        line_buffer,
                        next_chunk.decode("utf-8", errors="replace"),
                    )
                    continue
            if process.poll() is not None:
                trailing_chunk = stdout_stream.read()
                if trailing_chunk:
                    _write_remote_container_log_chunk(
                        line_buffer,
                        trailing_chunk.decode("utf-8", errors="replace"),
                    )
                break
        if stop_event.is_set() and process.poll() is None:
            process.terminate()
    finally:
        if process.poll() is None:
            try:
                process.wait(timeout=2.0)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait(timeout=2.0)
        _flush_remote_container_log_chunk(line_buffer)

    if process.returncode not in {0, None} and not stop_event.is_set():
        raise RuntimeError(
            f"Modal CLI exited with status {process.returncode} while streaming logs for task_id={task_id}."
        )
    return True


def _run_remote_container_log_stream(task_id: str, stop_event: threading.Event) -> None:
    """Run the best available remote log streaming backend for one Modal container."""
    logger.info("Starting remote Modal container log stream for task_id=%s.", task_id)
    try:
        if _stream_remote_container_logs_via_modal_cli(task_id, stop_event):
            logger.info(
                "Stopped remote Modal container log stream for task_id=%s.", task_id
            )
            return
    except (OSError, RuntimeError, subprocess.SubprocessError, ValueError) as exc:
        logger.warning(
            "Falling back from Modal CLI log streaming for task_id=%s after failure: %s",
            task_id,
            exc,
        )

    try:
        if _stream_remote_container_logs_via_modal_sdk(task_id, stop_event):
            logger.info(
                "Stopped remote Modal container log stream for task_id=%s.", task_id
            )
            return
    except (
        AttributeError,
        ImportError,
        ModuleNotFoundError,
        OSError,
        RuntimeError,
        ValueError,
    ) as exc:
        logger.warning(
            "Modal SDK log streaming failed for task_id=%s: %s",
            task_id,
            exc,
        )

    logger.warning(
        "Unable to mirror remote Modal container logs for task_id=%s because neither the Modal SDK nor CLI is available locally.",
        task_id,
    )


def _new_remote_container_log_stream_state(
    task_id: str,
) -> _RemoteContainerLogStreamState:
    """Create an unstarted container-log watcher state."""
    stop_event = threading.Event()
    return _RemoteContainerLogStreamState(
        task_id=task_id,
        stop_event=stop_event,
        thread=threading.Thread(
            target=_run_remote_container_log_stream,
            args=(task_id, stop_event),
            name=f"modal-log-stream-{task_id}",
            daemon=True,
        ),
    )


def _cancel_remote_container_log_idle_stop(
    stream_state: _RemoteContainerLogStreamState,
) -> None:
    """Cancel one pending idle-stop timer before a container is reused."""
    idle_stop_timer = stream_state.idle_stop_timer
    if idle_stop_timer is None:
        return
    idle_stop_timer.cancel()
    stream_state.idle_stop_timer = None


def _stop_remote_container_log_stream_if_idle(
    task_id: str,
    expected_state: _RemoteContainerLogStreamState,
) -> None:
    """Stop one watcher after its grace period if no payload reused it."""
    with _REMOTE_CONTAINER_LOG_STREAMS_LOCK:
        stream_state = _REMOTE_CONTAINER_LOG_STREAMS.get(task_id)
        if stream_state is not expected_state or stream_state.refcount != 0:
            return
        stream_state.idle_stop_timer = None
        _REMOTE_CONTAINER_LOG_STREAMS.pop(task_id, None)

    logger.info(
        "Stopping idle remote Modal container log stream for task_id=%s.",
        task_id,
    )
    stream_state.stop_event.set()
    stream_state.thread.join(timeout=0.2)


def _schedule_remote_container_log_idle_stop(
    stream_state: _RemoteContainerLogStreamState,
) -> None:
    """Schedule one daemon timer to stop an unreferenced log watcher."""
    idle_stop_timer = threading.Timer(
        _REMOTE_CONTAINER_LOG_STREAM_IDLE_GRACE_SECONDS,
        _stop_remote_container_log_stream_if_idle,
        args=(stream_state.task_id, stream_state),
    )
    idle_stop_timer.daemon = True
    stream_state.idle_stop_timer = idle_stop_timer
    idle_stop_timer.start()


def _retain_remote_container_log_stream(task_id: str) -> str:
    """Increment one shared Modal container log watcher and start it if needed."""
    with _REMOTE_CONTAINER_LOG_STREAMS_LOCK:
        stream_state = _REMOTE_CONTAINER_LOG_STREAMS.get(task_id)
        if stream_state is not None and not stream_state.thread.is_alive():
            _cancel_remote_container_log_idle_stop(stream_state)
            _REMOTE_CONTAINER_LOG_STREAMS.pop(task_id, None)
            stream_state = None

        if stream_state is None:
            stream_state = _new_remote_container_log_stream_state(task_id)
            _REMOTE_CONTAINER_LOG_STREAMS[task_id] = stream_state
            logger.info(
                "Creating remote Modal container log stream for task_id=%s.", task_id
            )
            stream_state.thread.start()
        else:
            _cancel_remote_container_log_idle_stop(stream_state)
            logger.info(
                "Reusing remote Modal container log stream for task_id=%s.", task_id
            )
        stream_state.refcount += 1
        logger.info(
            "Remote Modal container log stream retain task_id=%s refcount=%d.",
            task_id,
            stream_state.refcount,
        )

    return task_id


def _release_remote_container_log_stream(task_id: str) -> None:
    """Release one watcher while allowing prompt-local container reuse."""
    with _REMOTE_CONTAINER_LOG_STREAMS_LOCK:
        stream_state = _REMOTE_CONTAINER_LOG_STREAMS.get(task_id)
        if stream_state is None:
            return
        stream_state.refcount = max(0, stream_state.refcount - 1)
        logger.info(
            "Remote Modal container log stream release task_id=%s refcount=%d.",
            task_id,
            stream_state.refcount,
        )
        if stream_state.refcount != 0:
            return
        if not stream_state.thread.is_alive():
            _REMOTE_CONTAINER_LOG_STREAMS.pop(task_id, None)
            return
        if stream_state.idle_stop_timer is None:
            logger.info(
                "Keeping remote Modal container log stream alive for %.1fs "
                "to allow task reuse without replaying history task_id=%s.",
                _REMOTE_CONTAINER_LOG_STREAM_IDLE_GRACE_SECONDS,
                task_id,
            )
            _schedule_remote_container_log_idle_stop(stream_state)


def _close_remote_payload_stream(stream_events: Iterator[dict[str, Any]]) -> None:
    """Best-effort close one streamed Modal iterator after the terminal result arrives."""
    close_callable = getattr(stream_events, "close", None)
    if not callable(close_callable):
        return
    try:
        close_result = close_callable()
    except (AttributeError, RuntimeError, TypeError, ValueError) as exc:
        logger.debug("Ignoring remote payload stream close failure: %s", exc)
        return
    if not asyncio.iscoroutine(close_result):
        return
    try:
        asyncio.run(close_result)
    except (RuntimeError, ValueError) as exc:
        logger.debug("Ignoring async remote payload stream close failure: %s", exc)


def _payload_remote_session_handle(
    payload: dict[str, Any]
) -> RemoteSessionHandle | None:
    """Return the decoded prompt-scoped remote session handle for one payload."""
    remote_session = payload.get("remote_session")
    if not is_remote_session_handle_payload(remote_session):
        return None
    return RemoteSessionHandle.from_payload(remote_session)


def _sanitize_payload_for_session_bridge_record(
    payload: dict[str, Any]
) -> dict[str, Any]:
    """Strip run-scoped fields from one producer payload before persisting replay metadata."""
    sanitized_payload = copy.deepcopy(payload)
    sanitized_payload.pop("prompt_id", None)
    sanitized_payload.pop("remote_session", None)
    sanitized_payload.pop("clear_remote_session", None)
    sanitized_payload["extra_data"] = {}
    return sanitized_payload


def _durable_object_store() -> FileDurableObjectStore:
    """Return the process-local handle for durable binary objects."""
    global _DURABLE_OBJECT_STORE

    with _DURABLE_OBJECT_STORE_LOCK:
        if _DURABLE_OBJECT_STORE is not None:
            return _DURABLE_OBJECT_STORE
        settings = get_settings()
        commit_callback: Callable[[], Any] | None = None
        committed_read_callback: Callable[[str], bytes] | None = None
        if os.getenv("MODAL_IS_REMOTE") == "1" or os.getenv("MODAL_TASK_ID"):
            object_root = Path(settings.remote_storage_root) / "durable_objects"
            volume = globals().get("vol")
            volume_commit = getattr(volume, "commit", None)
            if callable(volume_commit):
                commit_callback = volume_commit
            if callable(getattr(volume, "read_file", None)):

                def read_committed_object(object_path: str) -> bytes:
                    """Read one durable object without reloading the mounted volume."""
                    volume_path = (Path("durable_objects") / object_path).as_posix()
                    return read_modal_volume_file(volume, volume_path)

                committed_read_callback = read_committed_object
        else:
            object_root = settings.local_storage_root / "durable_objects"
        _DURABLE_OBJECT_STORE = FileDurableObjectStore(
            object_root,
            commit_callback=commit_callback,
            committed_read_callback=committed_read_callback,
        )
        return _DURABLE_OBJECT_STORE


def _json_payload_size_bytes(payload: Any) -> int:
    """Return the compact UTF-8 size of one JSON-safe payload."""
    return len(
        json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
    )


def _select_remote_session_bridge_recovery_kind(
    *,
    serialized_output: Any | None,
    rehydration_plan: Mapping[str, Any] | None,
) -> RemoteSessionBridgeRecoveryKind:
    """Choose the least expensive complete recovery mechanism for one bridge."""
    if serialized_output is not None:
        return RemoteSessionBridgeRecoveryKind.SERIALIZED_OUTPUT
    if isinstance(rehydration_plan, Mapping):
        plan_kind = str(rehydration_plan.get("kind") or "")
        if plan_kind == "single_node_output":
            return RemoteSessionBridgeRecoveryKind.SINGLE_NODE_PLAN
        if plan_kind == "subgraph_output":
            return RemoteSessionBridgeRecoveryKind.SUBGRAPH_PLAN
    return RemoteSessionBridgeRecoveryKind.PRODUCER_REPLAY


def _remote_session_bridge_recovery_input_names(
    *,
    recovery_kind: RemoteSessionBridgeRecoveryKind,
    rehydration_plan: Mapping[str, Any] | None,
    hydrated_inputs: Mapping[str, Any],
) -> tuple[str, ...]:
    """Return producer input names that the selected recovery mechanism needs."""
    if recovery_kind in {
        RemoteSessionBridgeRecoveryKind.SERIALIZED_OUTPUT,
        RemoteSessionBridgeRecoveryKind.SINGLE_NODE_PLAN,
    }:
        return ()
    if recovery_kind is RemoteSessionBridgeRecoveryKind.PRODUCER_REPLAY:
        return tuple(str(input_name) for input_name in hydrated_inputs)

    plan_payload = (
        rehydration_plan.get("payload")
        if isinstance(rehydration_plan, Mapping)
        else None
    )
    boundary_inputs = (
        plan_payload.get("boundary_inputs")
        if isinstance(plan_payload, Mapping)
        else None
    )
    if not isinstance(boundary_inputs, list):
        raise RemoteSessionStateError(
            "Durable bridge subgraph recovery plans must define boundary_inputs."
        )
    input_names = tuple(
        dict.fromkeys(
            str(boundary_input.get("proxy_input_name") or "").strip()
            for boundary_input in boundary_inputs
            if isinstance(boundary_input, Mapping)
            and str(boundary_input.get("proxy_input_name") or "").strip()
        )
    )
    missing_input_names = [
        input_name for input_name in input_names if input_name not in hydrated_inputs
    ]
    if missing_input_names:
        raise RemoteSessionStateError(
            "Durable bridge subgraph recovery inputs are missing hydrated values: "
            f"{missing_input_names}."
        )
    return input_names


def _offload_large_bridge_payloads(
    *,
    hydrated_inputs: Mapping[str, Any],
    producer_inputs: dict[str, Any],
    output_value: Any,
    serialized_output: Any | None,
) -> tuple[
    dict[str, Any],
    DurableObjectRef | None,
    Any | None,
    DurableObjectRef | None,
]:
    """Move oversized bridge inputs and outputs into durable binary objects."""
    max_inline_bytes = get_settings().bridge_inline_max_bytes
    producer_inputs_object: DurableObjectRef | None = None
    serialized_output_object: DurableObjectRef | None = None
    if producer_inputs and _json_payload_size_bytes(producer_inputs) > max_inline_bytes:
        producer_inputs_object = _durable_object_store().put(
            "bridge_objects",
            serialize_node_inputs(hydrated_inputs),
        )
        producer_inputs = {}
    if (
        serialized_output is not None
        and _json_payload_size_bytes(serialized_output) > max_inline_bytes
    ):
        serialized_output_object = _durable_object_store().put(
            "bridge_objects",
            serialize_node_outputs((output_value,)),
        )
        serialized_output = None
    return (
        producer_inputs,
        producer_inputs_object,
        serialized_output,
        serialized_output_object,
    )


def _deserialize_remote_session_bridge_producer_inputs(
    record: RemoteSessionBridgeRecord,
) -> dict[str, Any]:
    """Restore bridge producer inputs from inline metadata or durable storage."""
    if record.producer_inputs_object is not None:
        return deserialize_node_inputs(
            _durable_object_store().get(record.producer_inputs_object)
        )
    if record.producer_inputs_retained is False:
        recovery_kind = (
            record.recovery_kind.value
            if record.recovery_kind is not None
            else "unspecified"
        )
        raise RemoteSessionStateError(
            "Remote session bridge producer inputs were intentionally omitted for "
            f"bridge_key={record.bridge_key!r} recovery_kind={recovery_kind!r}."
        )
    return deserialize_node_inputs(record.producer_inputs)


def _build_remote_session_bridge_record(
    *,
    payload: dict[str, Any],
    hydrated_inputs: dict[str, Any],
    node_id: str,
    output_index: int,
    io_type: str,
    output_value: Any,
) -> RemoteSessionBridgeRecord:
    """Build one durable bridge record for a session-backed boundary output."""
    producer_payload = _sanitize_payload_for_session_bridge_record(payload)
    producer_input_identity = serialize_mapping(hydrated_inputs)
    serialized_output = _serialize_durable_bridge_output(output_value, io_type)
    rehydration_plan = _build_durable_bridge_rehydration_plan(
        payload=producer_payload,
        node_id=node_id,
        output_index=output_index,
        io_type=io_type,
    )
    recovery_kind = _select_remote_session_bridge_recovery_kind(
        serialized_output=serialized_output,
        rehydration_plan=rehydration_plan,
    )
    recovery_input_names = _remote_session_bridge_recovery_input_names(
        recovery_kind=recovery_kind,
        rehydration_plan=rehydration_plan,
        hydrated_inputs=hydrated_inputs,
    )
    producer_inputs_retained = recovery_kind in {
        RemoteSessionBridgeRecoveryKind.SUBGRAPH_PLAN,
        RemoteSessionBridgeRecoveryKind.PRODUCER_REPLAY,
    }
    retained_hydrated_inputs = {
        input_name: hydrated_inputs[input_name] for input_name in recovery_input_names
    }
    producer_inputs = {
        input_name: producer_input_identity[input_name]
        for input_name in recovery_input_names
    }
    bridge_key = stable_session_bridge_key(
        producer_payload=producer_payload,
        producer_inputs=producer_input_identity,
        node_id=node_id,
        output_index=output_index,
    )
    (
        producer_inputs,
        producer_inputs_object,
        serialized_output,
        serialized_output_object,
    ) = _offload_large_bridge_payloads(
        hydrated_inputs=retained_hydrated_inputs,
        producer_inputs=producer_inputs,
        output_value=output_value,
        serialized_output=serialized_output,
    )
    logger.info(
        "Prepared remote session bridge bridge_key=%s recovery_kind=%s "
        "producer_inputs_retained=%s retained_input_count=%d omitted_input_count=%d.",
        bridge_key,
        recovery_kind.value,
        producer_inputs_retained,
        len(recovery_input_names),
        len(producer_input_identity) - len(recovery_input_names),
    )
    return RemoteSessionBridgeRecord(
        bridge_key=bridge_key,
        node_id=node_id,
        output_index=output_index,
        producer_payload=producer_payload,
        producer_inputs=producer_inputs,
        recovery_kind=recovery_kind,
        producer_inputs_retained=producer_inputs_retained,
        producer_inputs_object=producer_inputs_object,
        serialized_output=serialized_output,
        serialized_output_object=serialized_output_object,
        serialized_output_io_type=(
            str(io_type)
            if serialized_output is not None or serialized_output_object is not None
            else None
        ),
        rehydration_plan=rehydration_plan,
        rehydration_plan_io_type=(
            str(io_type) if rehydration_plan is not None else None
        ),
    )


def _record_remote_session_resolution_event(
    resolution_stats: "_RemoteSessionBridgeResolutionStats | None",
    event_name: str,
    event_payload: Mapping[str, Any],
) -> None:
    """Accumulate one local fallback remote-session resolution event."""
    del event_payload
    if resolution_stats is None:
        return
    if event_name in {"session-value-hit", "bridge-target-hit", "bridge-source-hit"}:
        resolution_stats.live_session_hits += 1


def _clone_cached_bridge_value(value: Any) -> Any:
    """Clone one cached bridge value when the runtime object exposes a safe clone path."""
    clone_method = getattr(value, "clone", None)
    if callable(clone_method):
        return clone_method()
    return value


def _store_remote_session_bridge_value(
    bridge_key: str,
    value: Any,
) -> None:
    """Retain one live bridge value in-process so later mapped phases can skip replay."""
    with _REMOTE_SESSION_BRIDGE_VALUE_CACHE_LOCK:
        _REMOTE_SESSION_BRIDGE_VALUE_CACHE[bridge_key] = _clone_cached_bridge_value(
            value
        )
        if bridge_key in _REMOTE_SESSION_BRIDGE_VALUE_CACHE_ORDER:
            _REMOTE_SESSION_BRIDGE_VALUE_CACHE_ORDER.remove(bridge_key)
        _REMOTE_SESSION_BRIDGE_VALUE_CACHE_ORDER.append(bridge_key)
        while (
            len(_REMOTE_SESSION_BRIDGE_VALUE_CACHE_ORDER)
            > _REMOTE_SESSION_BRIDGE_VALUE_CACHE_LIMIT
        ):
            evicted_key = _REMOTE_SESSION_BRIDGE_VALUE_CACHE_ORDER.pop(0)
            _REMOTE_SESSION_BRIDGE_VALUE_CACHE.pop(evicted_key, None)


def _get_remote_session_bridge_value(bridge_key: str) -> Any | None:
    """Return one retained bridge value when the current process still has it."""
    with _REMOTE_SESSION_BRIDGE_VALUE_CACHE_LOCK:
        cached_value = _REMOTE_SESSION_BRIDGE_VALUE_CACHE.get(bridge_key)
        if cached_value is None:
            return None
        if bridge_key in _REMOTE_SESSION_BRIDGE_VALUE_CACHE_ORDER:
            _REMOTE_SESSION_BRIDGE_VALUE_CACHE_ORDER.remove(bridge_key)
        _REMOTE_SESSION_BRIDGE_VALUE_CACHE_ORDER.append(bridge_key)
        return _clone_cached_bridge_value(cached_value)


def _serialize_durable_bridge_output(output_value: Any, io_type: str) -> Any | None:
    """Serialize one bridge output when its io_type supports durable direct restore."""
    normalized_io_type = str(io_type or "")
    if normalized_io_type not in _DURABLE_BRIDGE_SERIALIZATION_IO_TYPES:
        return None
    try:
        return serialize_value(output_value)
    except TypeError:
        logger.warning(
            "Skipping durable bridge serialization for io_type=%s value_type=%s.",
            normalized_io_type,
            type(output_value).__name__,
        )
        return None


def _restore_serialized_remote_session_bridge_value(
    record: RemoteSessionBridgeRecord,
    *,
    target_session_handle: RemoteSessionHandle,
    resolution_stats: "_RemoteSessionBridgeResolutionStats | None" = None,
) -> Any | None:
    """Restore one bridge value directly from a durable serialized payload."""
    if record.serialized_output is None and record.serialized_output_object is None:
        return None

    restore_started_at = time.perf_counter()
    if record.serialized_output_object is not None:
        restored_outputs = deserialize_node_outputs(
            _durable_object_store().get(record.serialized_output_object)
        )
        if len(restored_outputs) != 1:
            raise DurableStateError(
                f"Bridge object {record.serialized_output_object.object_path!r} "
                "must contain exactly one output."
            )
        restored_value = restored_outputs[0]
        storage_kind = "object-backed"
    else:
        restored_value = deserialize_value(record.serialized_output)
        storage_kind = "inline"
    _REMOTE_SESSION_STORE.put_bridge_output(
        target_session_handle,
        bridge_key=record.bridge_key,
        node_id=record.node_id,
        output_index=record.output_index,
        value=restored_value,
    )
    _store_remote_session_bridge_value(record.bridge_key, restored_value)
    if resolution_stats is not None:
        resolution_stats.durable_bridge_hits += 1
        resolution_stats.session_restore_writes += 1
        resolution_stats.direct_restore_seconds += (
            time.perf_counter() - restore_started_at
        )
    logger.info(
        "Resolved remote session bridge bridge_key=%s from durable %s serialized %s payload into session_id=%s.",
        record.bridge_key,
        storage_kind,
        record.serialized_output_io_type or "bridge",
        target_session_handle.session_id,
    )
    return restored_value


def materialize_remote_session_bridge_ref_locally(
    ref_payload: Mapping[str, Any],
) -> Any:
    """Download and deserialize one durable bridge value for local ComfyUI work."""
    if not is_remote_session_bridge_ref_payload(ref_payload):
        raise TypeError(
            "Local Modal bridge materialization requires a remote session bridge ref."
        )
    ref = RemoteSessionBridgeRef.from_payload(ref_payload)
    materialization_started_at = time.perf_counter()
    logger.info(
        "Starting local remote-bridge materialization bridge_key=%s node=%s output=%d.",
        ref.bridge_key,
        ref.node_id,
        ref.output_index,
    )
    record_lookup_started_at = time.perf_counter()
    record = _load_local_materialization_bridge_record(ref.bridge_key)
    logger.info(
        "Loaded local materialization bridge record in %.3fs bridge_key=%s object_backed=%s.",
        time.perf_counter() - record_lookup_started_at,
        ref.bridge_key,
        record.serialized_output_object is not None,
    )
    if record.serialized_output_object is not None:
        object_download_started_at = time.perf_counter()
        restored_outputs = deserialize_node_outputs(
            _read_local_materialization_bridge_object(
                record.serialized_output_object
            )
        )
        logger.info(
            "Downloaded and deserialized local bridge object in %.3fs bridge_key=%s bytes=%d.",
            time.perf_counter() - object_download_started_at,
            ref.bridge_key,
            record.serialized_output_object.size_bytes,
        )
        if len(restored_outputs) != 1:
            raise DurableStateError(
                f"Bridge object {record.serialized_output_object.object_path!r} "
                "must contain exactly one output."
            )
        restored_value = restored_outputs[0]
        storage_kind = "object-backed"
    elif record.serialized_output is not None:
        restored_value = deserialize_value(record.serialized_output)
        storage_kind = "inline"
    else:
        raise RemoteSessionStateError(
            "Remote bridge cannot be materialized locally without a durable "
            f"serialized output bridge_key={ref.bridge_key!r}."
        )
    logger.info(
        "Materialized remote bridge locally in %.3fs bridge_key=%s node=%s output=%d storage=%s io_type=%s.",
        time.perf_counter() - materialization_started_at,
        ref.bridge_key,
        ref.node_id,
        ref.output_index,
        storage_kind,
        record.serialized_output_io_type or "bridge",
    )
    return restored_value


def _load_local_materialization_bridge_record(
    bridge_key: str,
) -> RemoteSessionBridgeRecord:
    """Load a bridge record from local fallback state or Modal's shared Dict."""
    settings = get_settings()
    if settings.execution_mode != "remote":
        return _REMOTE_SESSION_BRIDGE_STORE.get_record(bridge_key)
    bridge_store = _lookup_modal_session_bridge_store()
    payload = bridge_store.get(bridge_key)
    if not isinstance(payload, Mapping):
        raise RemoteSessionStateError(
            f"Remote session bridge record {bridge_key!r} was not found."
        )
    return RemoteSessionBridgeRecord.from_payload(payload)


def _read_local_materialization_bridge_object(
    object_ref: DurableObjectRef,
) -> bytes:
    """Read and validate one bridge object locally from fallback or Modal storage."""
    settings = get_settings()
    if settings.execution_mode != "remote":
        return _durable_object_store().get(object_ref)
    volume_path = (Path("durable_objects") / object_ref.object_path).as_posix()
    payload = read_modal_volume_file(
        _lookup_modal_durable_volume(),
        volume_path,
    )
    if len(payload) != object_ref.size_bytes:
        raise DurableStateError(
            f"Durable bridge object {object_ref.object_path!r} has an unexpected size."
        )
    if hashlib.sha256(payload).hexdigest() != object_ref.sha256:
        raise DurableStateError(
            f"Durable bridge object {object_ref.object_path!r} failed its content-address check."
        )
    return payload


def _build_durable_bridge_rehydration_plan(
    *,
    payload: dict[str, Any],
    node_id: str,
    output_index: int,
    io_type: str,
) -> dict[str, Any] | None:
    """Return a direct rehydration plan when one bridge output can be rebuilt without replay."""
    if str(io_type or "") not in _DURABLE_BRIDGE_REHYDRATION_IO_TYPES:
        return None
    prompt = payload.get("subgraph_prompt")
    if not isinstance(prompt, dict):
        return None
    prompt_node = prompt.get(str(node_id))
    if not isinstance(prompt_node, dict):
        return None
    class_type = prompt_node.get("class_type")
    inputs = prompt_node.get("inputs")
    if (
        not isinstance(class_type, str)
        or not class_type.strip()
        or not isinstance(inputs, dict)
    ):
        return None

    normalized_inputs: dict[str, Any] = {}
    has_linked_input = False
    for input_name, input_value in inputs.items():
        normalized_value = _normalize_prompt_input_value(copy.deepcopy(input_value))
        if _is_link(normalized_value):
            has_linked_input = True
            continue
        normalized_inputs[str(input_name)] = normalized_value

    if not has_linked_input:
        node_data: dict[str, Any] = {"class_type": class_type}
        custom_nodes_bundle = payload.get("custom_nodes_bundle")
        if isinstance(custom_nodes_bundle, str) and custom_nodes_bundle.strip():
            node_data["custom_nodes_bundle"] = custom_nodes_bundle
        return {
            "kind": "single_node_output",
            "node_data": node_data,
            "node_inputs": normalized_inputs,
        }

    required_node_ids = set(
        _resolve_required_subgraph_nodes(
            prompt=prompt,
            execute_node_ids=[str(node_id)],
        )
    )
    if str(node_id) not in required_node_ids:
        return None
    if _subgraph_contains_sampling_node(prompt, required_node_ids):
        logger.info(
            "Skipping durable %s bridge subgraph rehydration plan for node_id=%s class_type=%s because its dependency closure includes a sampler.",
            io_type,
            node_id,
            class_type,
        )
        return None

    rehydration_payload = copy.deepcopy(payload)
    rehydration_payload[
        "component_id"
    ] = f"{payload.get('component_id', 'component')}::rehydrate:{node_id}"
    rehydration_payload["subgraph_prompt"] = {
        str(current_node_id): copy.deepcopy(prompt[current_node_id])
        for current_node_id in prompt
        if str(current_node_id) in required_node_ids
    }
    rehydration_payload["component_node_ids"] = sorted(required_node_ids)
    rehydration_payload["execute_node_ids"] = [str(node_id)]
    rehydration_payload["mapped_execute_node_ids"] = []
    rehydration_payload["static_execute_node_ids"] = []
    rehydration_payload["boundary_inputs"] = [
        {
            **copy.deepcopy(boundary_input),
            "targets": [
                copy.deepcopy(target)
                for target in boundary_input.get("targets", [])
                if str(target.get("node_id")) in required_node_ids
            ],
        }
        for boundary_input in payload.get("boundary_inputs", [])
        if any(
            str(target.get("node_id")) in required_node_ids
            for target in boundary_input.get("targets", [])
        )
    ]
    rehydration_payload["boundary_outputs"] = [
        {
            "node_id": str(node_id),
            "output_index": int(output_index),
            "io_type": str(io_type),
            "is_list": False,
            "session_output": False,
            "proxy_output_name": f"{node_id}_rehydrated_{output_index}",
        }
    ]
    return {
        "kind": "subgraph_output",
        "payload": rehydration_payload,
    }


def _restore_planned_remote_session_bridge_value(
    record: RemoteSessionBridgeRecord,
    *,
    target_session_handle: RemoteSessionHandle,
    node_mapping: dict[str, type[Any]] | None = None,
    resolution_stats: "_RemoteSessionBridgeResolutionStats | None" = None,
) -> Any | None:
    """Restore one bridge value directly from a durable node rehydration plan."""
    if not isinstance(record.rehydration_plan, Mapping):
        return None

    restore_started_at = time.perf_counter()
    plan_kind = str(record.rehydration_plan.get("kind") or "")
    if plan_kind == "single_node_output":
        node_data = record.rehydration_plan.get("node_data")
        node_inputs = record.rehydration_plan.get("node_inputs")
        if not isinstance(node_data, Mapping) or not isinstance(node_inputs, Mapping):
            return None
        outputs = _execute_node_locally_raw(
            dict(node_data),
            dict(node_inputs),
            node_mapping=node_mapping,
        )
        output_index = record.output_index
    elif plan_kind == "subgraph_output":
        plan_payload = record.rehydration_plan.get("payload")
        if not isinstance(plan_payload, Mapping):
            return None
        executable_payload = dict(plan_payload)
        executable_payload["remote_session"] = target_session_handle.to_payload()
        executable_payload.pop("clear_remote_session", None)
        outputs = _execute_subgraph_prompt(
            executable_payload,
            _deserialize_remote_session_bridge_producer_inputs(record),
            node_mapping,
        )
        output_index = 0
    else:
        return None
    if output_index < 0 or output_index >= len(outputs):
        raise RemoteSessionStateError(
            f"Durable bridge rehydration plan for {record.bridge_key!r} did not produce output index {record.output_index}."
        )
    restored_value = outputs[output_index]
    _REMOTE_SESSION_STORE.put_bridge_output(
        target_session_handle,
        bridge_key=record.bridge_key,
        node_id=record.node_id,
        output_index=record.output_index,
        value=restored_value,
    )
    _store_remote_session_bridge_value(record.bridge_key, restored_value)
    if resolution_stats is not None:
        resolution_stats.durable_bridge_hits += 1
        resolution_stats.session_restore_writes += 1
        resolution_stats.direct_restore_seconds += (
            time.perf_counter() - restore_started_at
        )
    logger.info(
        "Resolved remote session bridge bridge_key=%s from durable %s rehydration plan into session_id=%s.",
        record.bridge_key,
        record.rehydration_plan_io_type or "bridge",
        target_session_handle.session_id,
    )
    return restored_value


def _remote_session_bridge_replay_stack() -> set[str]:
    """Return the thread-local guard set for bridge replay recursion detection."""
    replay_stack = getattr(_REMOTE_SESSION_BRIDGE_REPLAY_STATE, "bridge_keys", None)
    if replay_stack is None:
        replay_stack = set()
        _REMOTE_SESSION_BRIDGE_REPLAY_STATE.bridge_keys = replay_stack
    return replay_stack


def _bridge_record_replays_sampling_node(record: RemoteSessionBridgeRecord) -> bool:
    """Return whether replaying one bridge record would rerun sampler work."""
    execute_node_ids = [
        str(node_id)
        for node_id in record.producer_payload.get("execute_node_ids", [])
        if str(node_id)
    ]
    if not execute_node_ids:
        return False
    subgraph_prompt = record.producer_payload.get("subgraph_prompt")
    if not isinstance(subgraph_prompt, dict):
        return False
    required_node_ids = _resolve_required_subgraph_nodes(
        prompt=subgraph_prompt,
        execute_node_ids=execute_node_ids,
    )
    return _subgraph_contains_sampling_node(subgraph_prompt, required_node_ids)


def _subgraph_contains_sampling_node(
    subgraph_prompt: Mapping[str, Any],
    node_ids: Iterable[str],
) -> bool:
    """Return whether any selected node looks like sampler work."""
    for node_id in node_ids:
        prompt_node = subgraph_prompt.get(node_id)
        if not isinstance(prompt_node, Mapping):
            continue
        class_type = str(prompt_node.get("class_type") or "")
        if "sampler" in class_type.lower():
            return True
    return False


def _rehydrate_remote_session_bridge_value(
    ref: RemoteSessionBridgeRef,
    *,
    target_session_handle: RemoteSessionHandle | None,
    node_mapping: dict[str, type[Any]] | None = None,
    resolution_stats: "_RemoteSessionBridgeResolutionStats | None" = None,
) -> Any:
    """Replay one producer phase into the current session when the live value is gone."""
    if target_session_handle is None:
        raise RemoteSessionStateError(
            "Remote session bridge replay requires a target remote_session handle."
        )

    cached_value = _get_remote_session_bridge_value(ref.bridge_key)
    if cached_value is not None:
        restore_started_at = time.perf_counter()
        _REMOTE_SESSION_STORE.put_bridge_output(
            target_session_handle,
            bridge_key=ref.bridge_key,
            node_id=ref.node_id,
            output_index=ref.output_index,
            value=cached_value,
        )
        if resolution_stats is not None:
            resolution_stats.bridge_cache_hits += 1
            resolution_stats.session_restore_writes += 1
            resolution_stats.direct_restore_seconds += (
                time.perf_counter() - restore_started_at
            )
        logger.info(
            "Resolved remote session bridge bridge_key=%s directly from warm cache into session_id=%s.",
            ref.bridge_key,
            target_session_handle.session_id,
        )
        return cached_value

    replay_stack = _remote_session_bridge_replay_stack()
    if ref.bridge_key in replay_stack:
        raise RemoteSessionStateError(
            f"Detected recursive remote session bridge replay for {ref.bridge_key!r}."
        )

    record_lookup_started_at = time.perf_counter()
    record = _REMOTE_SESSION_BRIDGE_STORE.get_record(ref.bridge_key)
    if resolution_stats is not None:
        resolution_stats.bridge_record_lookups += 1
        resolution_stats.bridge_record_lookup_seconds += (
            time.perf_counter() - record_lookup_started_at
        )
    restored_value = _restore_serialized_remote_session_bridge_value(
        record,
        target_session_handle=target_session_handle,
        resolution_stats=resolution_stats,
    )
    if restored_value is not None:
        return restored_value
    restored_value = _restore_planned_remote_session_bridge_value(
        record,
        target_session_handle=target_session_handle,
        node_mapping=node_mapping,
        resolution_stats=resolution_stats,
    )
    if restored_value is not None:
        return restored_value
    if _bridge_record_replays_sampling_node(record):
        message = (
            "Remote session bridge replay would rerun a sampler component "
            f"for bridge_key={ref.bridge_key!r}; expected live or warm bridge reuse instead."
        )
        logger.error(message)
        raise RemoteSessionStateError(message)
    replay_payload = copy.deepcopy(record.producer_payload)
    replay_payload["remote_session"] = target_session_handle.to_payload()
    replay_payload.pop("clear_remote_session", None)
    if target_session_handle.prompt_id is not None:
        replay_payload["prompt_id"] = target_session_handle.prompt_id
    replay_inputs = _deserialize_remote_session_bridge_producer_inputs(record)

    logger.info(
        "Replaying remote session bridge bridge_key=%s into session_id=%s via component=%s.",
        ref.bridge_key,
        target_session_handle.session_id,
        replay_payload.get("component_id"),
    )
    replay_stack.add(ref.bridge_key)
    replay_started_at = time.perf_counter()
    try:
        _execute_subgraph_prompt(replay_payload, replay_inputs, node_mapping)
    finally:
        replay_stack.remove(ref.bridge_key)
    if resolution_stats is not None:
        resolution_stats.replay_count += 1
        resolution_stats.replay_seconds += time.perf_counter() - replay_started_at

    return _REMOTE_SESSION_STORE.get_output(
        RemoteSessionValueRef(
            session_id=target_session_handle.session_id,
            node_id=ref.node_id,
            output_index=ref.output_index,
        )
    )


def _resolve_remote_session_inputs(
    hydrated_inputs: dict[str, Any],
    *,
    component_id: str | None = None,
    target_session_handle: RemoteSessionHandle | None = None,
    node_mapping: dict[str, type[Any]] | None = None,
    resolution_stats: "_RemoteSessionBridgeResolutionStats | None" = None,
) -> dict[str, Any]:
    """Resolve any remote-session value refs embedded in boundary inputs."""

    def count_refs(value: Any) -> int:
        """Return the number of live or durable session refs nested in one value."""
        if is_remote_session_value_ref_payload(
            value
        ) or is_remote_session_bridge_ref_payload(value):
            return 1
        if isinstance(value, list | tuple):
            return sum(count_refs(item) for item in value)
        if isinstance(value, Mapping):
            return sum(count_refs(item) for item in value.values())
        return 0

    ref_counts_by_input = {
        input_name: count_refs(input_value)
        for input_name, input_value in hydrated_inputs.items()
    }
    ref_input_names = [
        input_name
        for input_name, ref_count in ref_counts_by_input.items()
        if ref_count > 0
    ]
    total_ref_count = sum(ref_counts_by_input.values())
    if ref_input_names:
        logger.info(
            "Resolving %d remote session input refs for component=%s inputs=%s.",
            total_ref_count,
            component_id or "<unknown>",
            sorted(ref_input_names),
        )
    if resolution_stats is not None:
        resolution_stats.input_ref_count += total_ref_count
    return {
        input_name: _REMOTE_SESSION_STORE.resolve_value_with_bridges(
            input_value,
            target_session_handle=target_session_handle,
            resolution_callback=(
                lambda event_name, event_payload: _record_remote_session_resolution_event(
                    resolution_stats,
                    event_name,
                    event_payload,
                )
            )
            if resolution_stats is not None
            else None,
            bridge_resolver=lambda ref: _rehydrate_remote_session_bridge_value(
                ref,
                target_session_handle=target_session_handle,
                node_mapping=node_mapping,
                resolution_stats=resolution_stats,
            ),
        )
        for input_name, input_value in hydrated_inputs.items()
    }


def _log_remote_session_resolution_summary(
    *,
    component_id: str,
    resolution_stats: "_RemoteSessionBridgeResolutionStats",
) -> None:
    """Emit one high-signal summary for local fallback bridge resolution work."""
    if resolution_stats.input_ref_count <= 0:
        return
    logger.info(
        "Remote session resolution summary component=%s refs=%d live_hits=%d warm_bridge_hits=%d durable_bridge_hits=%d bridge_record_lookups=%d bridge_record_lookup_seconds=%.3f replay_count=%d replay_seconds=%.3f direct_restore_seconds=%.3f session_restore_writes=%d.",
        component_id,
        resolution_stats.input_ref_count,
        resolution_stats.live_session_hits,
        resolution_stats.bridge_cache_hits,
        resolution_stats.durable_bridge_hits,
        resolution_stats.bridge_record_lookups,
        resolution_stats.bridge_record_lookup_seconds,
        resolution_stats.replay_count,
        resolution_stats.replay_seconds,
        resolution_stats.direct_restore_seconds,
        resolution_stats.session_restore_writes,
    )


def _remote_worker_pool_affinity_key(slot_index: int) -> str:
    """Return the stable worker-pool affinity key for one reusable remote slot."""
    return f"worker-pool:slot:{int(slot_index)}"


def _component_pool_slot_index(payload: dict[str, Any]) -> int:
    """Return the reusable worker-pool slot index for one ordinary remote payload."""
    warmup_slot_index = payload.get("warmup_slot_index")
    if warmup_slot_index is not None:
        return max(0, int(warmup_slot_index))
    if bool(payload.get("remote_local_gap_pool")):
        return 0

    prompt_parallelism_target = _prompt_parallelism_target(payload)
    slot_count = max(1, int(prompt_parallelism_target))
    component_id = str(payload.get("component_id") or "")
    if not component_id:
        return 0

    component_hash = hashlib.sha256(component_id.encode("utf-8")).digest()
    return int.from_bytes(component_hash[:8], "big") % slot_count


def _remote_worker_affinity_key(payload: dict[str, Any]) -> str:
    """Return the reusable worker-pool affinity key for one remote payload."""
    configured_affinity_group = payload.get("remote_worker_affinity_group")
    if configured_affinity_group is None:
        return _remote_worker_pool_affinity_key(_component_pool_slot_index(payload))
    affinity_group = str(configured_affinity_group).strip().lower()
    if affinity_group not in {"comfy", "llm"}:
        affinity_group = "comfy"
    slot_index = _component_pool_slot_index(payload)
    return f"worker-pool:{affinity_group}:slot:{slot_index}"


def _mapped_lane_affinity_key(payload: dict[str, Any], lane_index: int) -> str | None:
    """Return the stable per-lane worker-pool affinity key used for mapped execution."""
    del payload
    return _remote_worker_pool_affinity_key(lane_index)


class _NullPromptServer:
    """Minimal PromptExecutor server stub for headless subgraph execution."""

    def __init__(self) -> None:
        """Initialize the no-op prompt server state."""
        self.client_id: str | None = None
        self.last_node_id: str | None = None

    def send_sync(
        self, event: str, data: dict[str, Any], client_id: str | None
    ) -> None:
        """Discard PromptExecutor progress and status events."""
        logger.debug(
            "Suppressed remote prompt event %s for client %s.", event, client_id
        )


def _extract_custom_nodes_bundle(bundle_path: str | None) -> None:
    """Extract a mirrored custom_nodes bundle ZIP or manifest into a temporary import path."""
    if not bundle_path:
        return

    settings = get_settings()
    if settings.execution_mode == "local":
        logger.debug("Skipping custom_nodes bundle extraction in local execution mode.")
        return

    local_bundle = settings.local_storage_root / bundle_path.lstrip("/")
    if not local_bundle.exists():
        logger.warning(
            "Custom nodes bundle %s was not found in local storage.", local_bundle
        )
        return

    extraction_root = (
        Path(tempfile.gettempdir())
        / "comfy-modal-sync-custom-nodes"
        / local_bundle.stem
    )
    extraction_root.mkdir(parents=True, exist_ok=True)
    for archive_path in _resolve_local_custom_nodes_archives(
        local_bundle, settings.local_storage_root
    ):
        with zipfile.ZipFile(archive_path, "r") as archive:
            archive.extractall(extraction_root)
    _materialize_local_custom_node_assets(
        local_bundle,
        settings.local_storage_root,
        extraction_root,
    )

    if str(extraction_root) not in sys.path:
        sys.path.insert(0, str(extraction_root))
    logger.info("Extracted remote custom_nodes bundle to %s", extraction_root)


def _resolve_local_custom_nodes_archives(
    local_bundle: Path, storage_root: Path
) -> list[Path]:
    """Return the archive paths described by one local custom_nodes bundle ZIP or manifest."""
    if local_bundle.suffix.lower() == ".zip":
        return [local_bundle]
    if local_bundle.suffix.lower() != ".json":
        raise RuntimeError(
            f"Unsupported custom_nodes bundle format {local_bundle.suffix!r} for {local_bundle}."
        )

    manifest_payload = _load_local_custom_nodes_manifest(local_bundle)
    entry_payloads = manifest_payload.get("entries")
    if not isinstance(entry_payloads, list):
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} did not contain a valid entries list."
        )

    archive_paths: list[Path] = []
    for entry_payload in entry_payloads:
        if not isinstance(entry_payload, dict):
            raise RuntimeError(
                f"Custom nodes manifest {local_bundle} contained a non-object entry."
            )
        remote_path = entry_payload.get("remote_path")
        if not isinstance(remote_path, str) or not remote_path.strip():
            raise RuntimeError(
                f"Custom nodes manifest {local_bundle} contained an entry without remote_path."
            )
        archive_path = storage_root / remote_path.lstrip("/")
        if not archive_path.exists():
            raise RuntimeError(
                f"Custom nodes archive {remote_path} referenced by {local_bundle} was not found in local storage."
            )
        archive_paths.append(archive_path)
    return archive_paths


def _load_local_custom_nodes_manifest(local_bundle: Path) -> dict[str, Any]:
    """Load and validate one local custom-node bundle manifest."""
    try:
        manifest_payload = json.loads(local_bundle.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} is unreadable."
        ) from exc
    if not isinstance(manifest_payload, dict):
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} must be a JSON object."
        )
    if manifest_payload.get("version", 1) not in {1, 2}:
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} has an unsupported version."
        )
    return manifest_payload


def _materialize_local_custom_node_assets(
    local_bundle: Path,
    storage_root: Path,
    extraction_root: Path,
) -> None:
    """Link version-two package assets into the local fallback extraction tree."""
    if local_bundle.suffix.lower() != ".json":
        return
    manifest_payload = _load_local_custom_nodes_manifest(local_bundle)
    if manifest_payload.get("version", 1) < 2:
        return
    for asset_payload in _iter_local_custom_node_assets(local_bundle, manifest_payload):
        relative_path = _validated_local_custom_node_asset_path(
            local_bundle, asset_payload
        )
        source_path = storage_root / str(asset_payload["remote_path"]).lstrip("/")
        if not source_path.is_file():
            raise RuntimeError(
                f"Custom-node asset {source_path} was not found in local storage."
            )
        if source_path.stat().st_size != int(asset_payload["size_bytes"]):
            raise RuntimeError(
                f"Custom-node asset {source_path} size did not match its manifest."
            )
        destination = extraction_root / relative_path
        destination.parent.mkdir(parents=True, exist_ok=True)
        if destination.exists() or destination.is_symlink():
            if (
                destination.is_symlink()
                and destination.resolve() == source_path.resolve()
            ):
                continue
            raise RuntimeError(
                f"Custom-node asset destination {destination} already exists."
            )
        destination.symlink_to(source_path)


def _iter_local_custom_node_assets(
    local_bundle: Path,
    manifest_payload: Mapping[str, Any],
) -> Iterator[dict[str, Any]]:
    """Yield asset objects from one validated version-two local manifest."""
    entries = manifest_payload.get("entries")
    if not isinstance(entries, list):
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} has no valid entries list."
        )
    for entry in entries:
        if not isinstance(entry, dict) or not isinstance(entry.get("assets", []), list):
            raise RuntimeError(
                f"Custom nodes manifest {local_bundle} contains invalid assets."
            )
        for asset_payload in entry.get("assets", []):
            if not isinstance(asset_payload, dict):
                raise RuntimeError(
                    f"Custom nodes manifest {local_bundle} contains an invalid asset."
                )
            yield asset_payload


def _validated_local_custom_node_asset_path(
    local_bundle: Path,
    asset_payload: Mapping[str, Any],
) -> Path:
    """Return one safe local extraction-relative package asset path."""
    relative_path = asset_payload.get("relative_path")
    remote_path = asset_payload.get("remote_path")
    sha256 = asset_payload.get("sha256")
    size_bytes = asset_payload.get("size_bytes")
    if not isinstance(relative_path, str) or not relative_path:
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} contains an asset without a path."
        )
    if not isinstance(remote_path, str) or not remote_path:
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} contains an asset without storage."
        )
    if (
        not isinstance(sha256, str)
        or len(sha256) != 64
        or not Path(remote_path).name.startswith(f"{sha256}_")
    ):
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} contains an invalid asset digest."
        )
    if (
        isinstance(size_bytes, bool)
        or not isinstance(size_bytes, int)
        or size_bytes < 0
    ):
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} contains an invalid asset size."
        )
    normalized_path = Path(relative_path)
    if normalized_path.is_absolute() or ".." in normalized_path.parts:
        raise RuntimeError(
            f"Custom nodes manifest {local_bundle} contains an unsafe asset path."
        )
    return normalized_path


def _load_nodes_module() -> Any:
    """Import the ComfyUI nodes module lazily."""
    import nodes

    return nodes


def _load_execution_module() -> Any:
    """Import the ComfyUI execution module lazily."""
    import execution

    return execution


@contextmanager
def _temporary_node_mapping(
    node_mapping: dict[str, type[Any]] | None
) -> Iterator[None]:
    """Temporarily overlay node mappings for tests or custom runtimes."""
    if node_mapping is None:
        yield
        return

    nodes_module = _load_nodes_module()
    original_mappings = dict(nodes_module.NODE_CLASS_MAPPINGS)
    original_display_mappings = dict(
        getattr(nodes_module, "NODE_DISPLAY_NAME_MAPPINGS", {})
    )
    try:
        nodes_module.NODE_CLASS_MAPPINGS.update(node_mapping)
        for class_type in node_mapping:
            nodes_module.NODE_DISPLAY_NAME_MAPPINGS.setdefault(class_type, class_type)
        yield
    finally:
        nodes_module.NODE_CLASS_MAPPINGS.clear()
        nodes_module.NODE_CLASS_MAPPINGS.update(original_mappings)
        if hasattr(nodes_module, "NODE_DISPLAY_NAME_MAPPINGS"):
            nodes_module.NODE_DISPLAY_NAME_MAPPINGS.clear()
            nodes_module.NODE_DISPLAY_NAME_MAPPINGS.update(original_display_mappings)


def _invoke_original_node(
    node_class: type[Any],
    node_data: dict[str, Any],
    kwargs: dict[str, Any],
) -> tuple[Any, ...]:
    """Execute an original V1 or V3 node class and normalize its outputs."""
    class_type = node_data["class_type"]
    logger.info("Executing remote node %s", class_type)

    if hasattr(node_class, "GET_SCHEMA"):
        node_output = node_class.execute(**kwargs)
        if hasattr(node_output, "result"):
            result = node_output.result
            return tuple(result) if result is not None else tuple()
        return tuple(node_output)

    instance = node_class()
    function_name = getattr(node_class, "FUNCTION", "execute")
    function = getattr(instance, function_name)
    result = function(**kwargs)
    if result is None:
        return tuple()
    if isinstance(result, tuple):
        return result
    if isinstance(result, list):
        return tuple(result)
    return (result,)


def execute_node_locally(
    node_data: dict[str, Any],
    kwargs_payload: bytes | bytearray | str | dict[str, Any],
    node_mapping: dict[str, type[Any]] | None = None,
) -> bytes:
    """Execute a single target node in-process and return serialized outputs."""
    outputs = _execute_node_locally_raw(
        node_data,
        kwargs_payload,
        node_mapping=node_mapping,
    )
    return serialize_node_outputs(outputs)


def _execute_node_locally_raw(
    node_data: dict[str, Any],
    kwargs_payload: bytes | bytearray | str | dict[str, Any],
    node_mapping: dict[str, type[Any]] | None = None,
) -> tuple[Any, ...]:
    """Execute a single target node in-process and return raw node outputs."""
    _extract_custom_nodes_bundle(node_data.get("custom_nodes_bundle"))
    kwargs = deserialize_node_inputs(kwargs_payload)
    if node_mapping is not None:
        class_type = node_data["class_type"]
        if class_type not in node_mapping:
            raise KeyError(f"Remote node class {class_type!r} is not registered.")
        return _invoke_original_node(node_mapping[class_type], node_data, kwargs)

    with _temporary_node_mapping(node_mapping):
        resolved_node_mapping = _load_nodes_module().NODE_CLASS_MAPPINGS
        class_type = node_data["class_type"]
        if class_type not in resolved_node_mapping:
            raise KeyError(f"Remote node class {class_type!r} is not registered.")

        return _invoke_original_node(
            resolved_node_mapping[class_type], node_data, kwargs
        )


def _apply_boundary_inputs(
    prompt: dict[str, Any],
    boundary_input_specs: list[dict[str, Any]],
    hydrated_inputs: dict[str, Any],
) -> None:
    """Inject hydrated local boundary inputs into a remote subgraph prompt."""
    logger.info(
        "Applying %d hydrated boundary inputs to remote subgraph prompt.",
        len(boundary_input_specs),
    )
    for boundary_input in boundary_input_specs:
        proxy_input_name = str(boundary_input["proxy_input_name"])
        if proxy_input_name not in hydrated_inputs:
            raise KeyError(f"Missing hydrated boundary input {proxy_input_name!r}.")
        value = unwrap_mapped_output_value(hydrated_inputs[proxy_input_name])
        io_type = (
            str(boundary_input["io_type"])
            if boundary_input.get("io_type") is not None
            else None
        )
        logger.info(
            "Applying boundary input %s to %d targets.",
            proxy_input_name,
            len(boundary_input.get("targets", [])),
        )
        for target in boundary_input.get("targets", []):
            node_id = str(target["node_id"])
            input_name = str(target["input_name"])
            prompt_node = prompt[node_id]
            prompt_node["inputs"][input_name] = _normalize_prompt_input_value(
                value,
                io_type=io_type,
            )
            source_signature = boundary_input.get("source_signature")
            if isinstance(source_signature, str) and source_signature:
                boundary_signatures = prompt_node.setdefault(
                    _BOUNDARY_INPUT_SIGNATURES_KEY, {}
                )
                if isinstance(boundary_signatures, dict):
                    boundary_signatures[input_name] = source_signature


def _collapse_cache_slot(slot_values: Any, is_list: bool) -> Any:
    """Convert a PromptExecutor cache slot back into a node-style output value."""
    if is_list:
        return slot_values
    if not isinstance(slot_values, list):
        return slot_values
    if len(slot_values) == 1:
        return slot_values[0]
    return slot_values


def _extract_prompt_executor_error(executor: Any) -> str:
    """Extract a useful failure message from a PromptExecutor run."""
    for event, data in reversed(executor.status_messages):
        if event == "execution_error":
            return _format_prompt_executor_error_payload(data)
        if event == "execution_interrupted":
            return "Remote subgraph execution was interrupted."
    return "Remote subgraph execution failed."


def _format_prompt_executor_error_payload(data: Any) -> str:
    """Return a richer human-readable PromptExecutor failure message when available."""
    if not isinstance(data, dict):
        return "Remote subgraph execution failed."

    message = str(data.get("exception_message") or "Remote subgraph execution failed.")
    node_id = data.get("node_id")
    node_type = data.get("node_type")
    current_inputs = data.get("current_inputs")
    if node_id is None and node_type is None and not current_inputs:
        return message

    details: list[str] = [message]
    if node_id is not None or node_type is not None:
        details.append(f"node_id={node_id!r} node_type={node_type!r}")
    if current_inputs:
        details.append(f"current_inputs={current_inputs!r}")
    return " | ".join(details)


def _node_input_types(
    node_class: type[Any],
    live_inputs: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return one node class's input schema finalized for the live prompt inputs."""
    input_types_callable = getattr(node_class, "INPUT_TYPES", None)
    if not callable(input_types_callable):
        return {}

    raw_input_types = input_types_callable()
    if not isinstance(raw_input_types, dict):
        return {}

    has_v3_dynamic_inputs = any(
        isinstance(input_config, tuple)
        and bool(input_config)
        and isinstance(input_config[0], str)
        and input_config[0].startswith("COMFY_")
        and input_config[0].endswith("_V3")
        for section_name in ("required", "optional")
        for input_config in (
            raw_input_types.get(section_name, {}).values()
            if isinstance(raw_input_types.get(section_name), dict)
            else ()
        )
    )
    if not has_v3_dynamic_inputs:
        return raw_input_types

    from comfy_api.latest import _io

    finalized_input_types, _, _ = _io.get_finalized_class_inputs(
        raw_input_types,
        live_inputs or {},
    )
    logger.debug(
        "Finalized V3 input schema for %s against prompt inputs %s.",
        node_class.__name__,
        sorted(str(input_name) for input_name in (live_inputs or {})),
    )
    return finalized_input_types


def _node_input_type_map(
    node_class: type[Any],
    live_inputs: dict[str, Any] | None = None,
) -> dict[str, str]:
    """Return one node class's finalized input types keyed by prompt input name."""
    input_types = _node_input_types(node_class, live_inputs)

    input_type_map: dict[str, str] = {}
    for section_name in ("required", "optional", "hidden"):
        section = input_types.get(section_name)
        if not isinstance(section, dict):
            continue
        for input_name, input_config in section.items():
            if not isinstance(input_config, tuple) or not input_config:
                continue
            declared_type = input_config[0]
            if isinstance(declared_type, str):
                input_type_map[str(input_name)] = declared_type
    return input_type_map


def _node_required_input_names(
    node_class: type[Any],
    live_inputs: dict[str, Any] | None = None,
) -> set[str]:
    """Return required input names after expanding V3 dynamic prompt inputs."""
    required_inputs = _node_input_types(node_class, live_inputs).get("required")
    if not isinstance(required_inputs, dict):
        return set()
    return {str(input_name) for input_name in required_inputs}


def _coerce_primitive_prompt_input_value(
    *,
    node_id: str,
    class_type: str,
    input_name: str,
    declared_type: str,
    input_value: Any,
) -> Any:
    """Coerce one primitive prompt literal using ComfyUI's `validate_inputs` semantics."""
    literal_value = (
        input_value.get("__value__")
        if isinstance(input_value, dict) and "__value__" in input_value
        else input_value
    )
    if isinstance(literal_value, list):
        return input_value

    try:
        if declared_type == "INT":
            return int(literal_value)
        if declared_type == "FLOAT":
            return float(literal_value)
        if declared_type == "STRING":
            return str(literal_value)
        if declared_type == "BOOLEAN":
            return bool(literal_value)
    except (OverflowError, TypeError, ValueError) as exc:
        raise RemoteSubgraphExecutionError(
            "Remote subgraph input could not be coerced to the declared primitive socket type."
            f" node_id={node_id!r} node_type={class_type!r}"
            f" input_name={input_name!r} declared_type={declared_type!r}"
            f" received_value={literal_value!r}"
        ) from exc

    return input_value


def _coerce_prompt_primitive_input_values(
    prompt: dict[str, Any],
    node_mapping: dict[str, type[Any]],
) -> None:
    """Mutate prompt literals in-place to match ComfyUI's primitive widget coercion."""
    for node_id, prompt_node in sorted(prompt.items()):
        class_type = str(prompt_node.get("class_type"))
        node_class = node_mapping.get(class_type)
        if node_class is None:
            continue
        inputs = prompt_node.get("inputs") or {}
        input_type_map = _node_input_type_map(node_class, inputs)
        if not input_type_map:
            continue
        for input_name, input_value in list(inputs.items()):
            declared_type = input_type_map.get(str(input_name))
            if declared_type not in _PRIMITIVE_WIDGET_INPUT_TYPES:
                continue
            if (
                isinstance(input_value, list)
                and len(input_value) == 2
                and isinstance(input_value[0], str)
            ):
                continue
            coerced_value = _coerce_primitive_prompt_input_value(
                node_id=str(node_id),
                class_type=class_type,
                input_name=str(input_name),
                declared_type=declared_type,
                input_value=input_value,
            )
            if coerced_value is not input_value:
                logger.debug(
                    "Coerced remote primitive input %s.%s from %r to %r for type %s.",
                    node_id,
                    input_name,
                    input_value,
                    coerced_value,
                    declared_type,
                )
                inputs[input_name] = coerced_value


def _validate_prompt_input_shapes(
    prompt: dict[str, Any],
    node_mapping: dict[str, type[Any]],
    boundary_input_specs: list[dict[str, Any]] | None = None,
) -> None:
    """Reject prompt inputs that still look invalid for primitive widget sockets."""
    boundary_targets = {
        (str(target.get("node_id")), str(target.get("input_name")))
        for boundary_input in (boundary_input_specs or [])
        for target in boundary_input.get("targets", [])
        if target.get("node_id") is not None and target.get("input_name") is not None
    }
    for node_id, prompt_node in sorted(prompt.items()):
        class_type = str(prompt_node.get("class_type"))
        node_class = node_mapping.get(class_type)
        if node_class is None:
            continue
        inputs = prompt_node.get("inputs") or {}
        input_type_map = _node_input_type_map(node_class, inputs)
        if not input_type_map:
            continue
        for input_name, input_value in inputs.items():
            declared_type = input_type_map.get(str(input_name))
            if declared_type not in _PRIMITIVE_WIDGET_INPUT_TYPES:
                continue
            if (
                isinstance(input_value, list)
                and len(input_value) == 2
                and isinstance(input_value[0], str)
            ):
                continue
            if (str(node_id), str(input_name)) in boundary_targets:
                continue
            literal_value = (
                input_value.get("__value__")
                if isinstance(input_value, dict) and "__value__" in input_value
                else input_value
            )
            if isinstance(literal_value, list):
                raise RemoteSubgraphExecutionError(
                    "Remote subgraph input has an invalid list value for a primitive socket."
                    f" node_id={node_id!r} node_type={class_type!r}"
                    f" input_name={input_name!r} declared_type={declared_type!r}"
                    f" received_value={literal_value!r}"
                )


def _validate_required_prompt_inputs(
    prompt: dict[str, Any],
    node_mapping: dict[str, type[Any]],
) -> None:
    """Reject remote prompt nodes that are missing declared required inputs."""
    for node_id, prompt_node in sorted(prompt.items()):
        class_type = str(prompt_node.get("class_type"))
        node_class = node_mapping.get(class_type)
        if node_class is None:
            continue
        inputs = prompt_node.get("inputs") or {}
        missing_inputs = sorted(
            _node_required_input_names(node_class, inputs) - set(inputs.keys())
        )
        if not missing_inputs:
            continue
        raise RemoteSubgraphExecutionError(
            "Remote subgraph prompt is missing required node inputs before execution."
            f" node_id={node_id!r} node_type={class_type!r}"
            f" missing_inputs={missing_inputs!r} available_inputs={sorted(str(key) for key in inputs)!r}"
        )


def _resolve_required_subgraph_nodes(
    prompt: dict[str, Any],
    execute_node_ids: list[str],
) -> list[str]:
    """Return the dependency closure needed to execute the requested subgraph nodes."""
    required: set[str] = set()
    pending = list(execute_node_ids)
    logger.info(
        "Resolving dependency closure for remote execute targets: %s", execute_node_ids
    )
    while pending:
        node_id = str(pending.pop())
        if node_id not in prompt:
            logger.warning(
                "Skipping missing remote execute target %s while resolving dependency closure.",
                node_id,
            )
            continue
        if node_id in required:
            continue
        required.add(node_id)
        for input_value in (prompt[node_id].get("inputs") or {}).values():
            if _is_link(input_value):
                pending.append(str(input_value[0]))
    resolved = sorted(required)
    logger.info("Resolved remote dependency closure: %s", resolved)
    return resolved


def _trim_subgraph_payload_to_required_nodes(payload: dict[str, Any]) -> dict[str, Any]:
    """Trim a subgraph payload down to the dependency closure of its execute targets."""
    trimmed_payload = copy.deepcopy(payload)
    prompt = trimmed_payload.get("subgraph_prompt", {})
    if not isinstance(prompt, dict):
        return trimmed_payload

    prompt_node_ids = {str(node_id) for node_id in prompt}
    requested_execute_node_ids = [
        str(node_id) for node_id in trimmed_payload.get("execute_node_ids", [])
    ]
    available_execute_node_ids = [
        node_id for node_id in requested_execute_node_ids if node_id in prompt_node_ids
    ]
    dropped_execute_node_ids = [
        node_id
        for node_id in requested_execute_node_ids
        if node_id not in prompt_node_ids
    ]
    if dropped_execute_node_ids:
        logger.warning(
            "Dropping remote execute targets absent from subgraph prompt for component=%s: %s",
            payload.get("component_id"),
            dropped_execute_node_ids,
        )

    required_node_ids = set(
        _resolve_required_subgraph_nodes(
            prompt=prompt,
            execute_node_ids=available_execute_node_ids,
        )
    )
    if not required_node_ids:
        return trimmed_payload

    original_node_ids = list(prompt.keys())
    trimmed_payload["subgraph_prompt"] = {
        str(node_id): prompt[node_id]
        for node_id in original_node_ids
        if str(node_id) in required_node_ids
    }
    trimmed_payload["boundary_inputs"] = [
        {
            **copy.deepcopy(boundary_input),
            "targets": [
                copy.deepcopy(target)
                for target in boundary_input.get("targets", [])
                if str(target.get("node_id")) in required_node_ids
            ],
        }
        for boundary_input in trimmed_payload.get("boundary_inputs", [])
        if any(
            str(target.get("node_id")) in required_node_ids
            for target in boundary_input.get("targets", [])
        )
    ]
    trimmed_payload["boundary_outputs"] = [
        copy.deepcopy(boundary_output)
        for boundary_output in trimmed_payload.get("boundary_outputs", [])
        if str(boundary_output.get("node_id")) in required_node_ids
    ]
    trimmed_payload["component_node_ids"] = [
        str(node_id)
        for node_id in trimmed_payload.get("component_node_ids", [])
        if str(node_id) in required_node_ids
    ]
    trimmed_payload["execute_node_ids"] = [
        str(node_id)
        for node_id in trimmed_payload.get("execute_node_ids", [])
        if str(node_id) in required_node_ids
    ]
    trimmed_payload["mapped_execute_node_ids"] = [
        str(node_id)
        for node_id in trimmed_payload.get("mapped_execute_node_ids", [])
        if str(node_id) in required_node_ids
    ]
    trimmed_payload["static_execute_node_ids"] = [
        str(node_id)
        for node_id in trimmed_payload.get("static_execute_node_ids", [])
        if str(node_id) in required_node_ids
    ]
    logger.info(
        "Trimmed remote subgraph payload %s from %d prompt nodes to %d required nodes.",
        payload.get("component_id"),
        len(original_node_ids),
        len(trimmed_payload["subgraph_prompt"]),
    )
    return trimmed_payload


def _is_link(value: Any) -> bool:
    """Return whether a prompt input value is a ComfyUI link."""
    return (
        isinstance(value, list)
        and len(value) == 2
        and all(not isinstance(item, dict) for item in value)
    )


def _iter_prompt_links(value: Any) -> Iterator[list[Any]]:
    """Yield ComfyUI prompt links found inside a prompt input value."""
    if _is_link(value):
        yield value
        return
    if isinstance(value, Mapping):
        for nested_value in value.values():
            yield from _iter_prompt_links(nested_value)
        return
    if isinstance(value, list):
        for nested_value in value:
            yield from _iter_prompt_links(nested_value)


def _normalize_link_output_index(value: Any) -> Any:
    """Unwrap a singleton list around a prompt-link output index when present."""
    while isinstance(value, list) and len(value) == 1:
        value = value[0]
    return value


def _unwrap_wrapped_prompt_link(value: Any) -> Any:
    """Collapse nested singleton wrappers around one serialized prompt link when present."""
    candidate = value
    while isinstance(candidate, list) and len(candidate) == 1:
        candidate = candidate[0]
    if _is_link(candidate):
        return [candidate[0], _normalize_link_output_index(candidate[1])]
    return value


def _normalize_prompt_input_value(value: Any, io_type: str | None = None) -> Any:
    """Unwrap transport-added singleton wrappers only for scalar-like prompt input values."""
    wrapped_link = _unwrap_wrapped_prompt_link(value)
    if wrapped_link is not value:
        return wrapped_link
    while (
        isinstance(value, list)
        and len(value) == 1
        and (
            io_type in _PRIMITIVE_WIDGET_INPUT_TYPES
            or value[0] is None
            or isinstance(value[0], bool | int | float | str)
        )
    ):
        value = value[0]
    if isinstance(value, list) and len(value) == 2 and isinstance(value[0], str):
        return [value[0], _normalize_link_output_index(value[1])]
    if value is None or isinstance(value, bool | int | float | str):
        return value
    return value


def _normalize_subgraph_payload(payload: dict[str, Any]) -> dict[str, Any]:
    """Return a subgraph payload with canonical prompt-link and output-index shapes."""
    normalized_payload = copy.deepcopy(payload)

    for node_info in normalized_payload.get("subgraph_prompt", {}).values():
        inputs = node_info.get("inputs") or {}
        for input_name, input_value in list(inputs.items()):
            inputs[input_name] = _normalize_prompt_input_value(input_value)

    for boundary_output in normalized_payload.get("boundary_outputs", []):
        if "node_id" in boundary_output and isinstance(
            boundary_output["node_id"], list
        ):
            boundary_output["node_id"] = _normalize_prompt_input_value(
                boundary_output["node_id"]
            )
        if "output_index" in boundary_output:
            boundary_output["output_index"] = _normalize_link_output_index(
                boundary_output["output_index"]
            )

    normalized_payload["execute_node_ids"] = [
        _normalize_prompt_input_value(node_id)
        for node_id in normalized_payload.get("execute_node_ids", [])
    ]

    return normalized_payload


def _execute_subgraph_with_mapping(
    payload: dict[str, Any],
    hydrated_inputs: dict[str, Any],
    node_mapping: dict[str, type[Any]],
) -> tuple[Any, ...]:
    """Execute a rewritten remote component using an explicit node mapping."""
    normalized_payload = _trim_subgraph_payload_to_required_nodes(
        _normalize_subgraph_payload(payload)
    )
    session_handle = _payload_remote_session_handle(normalized_payload)
    resolution_stats = _RemoteSessionBridgeResolutionStats()
    resolved_inputs = _resolve_remote_session_inputs(
        dict(hydrated_inputs),
        component_id=str(payload.get("component_id") or ""),
        target_session_handle=session_handle,
        node_mapping=node_mapping,
        resolution_stats=resolution_stats,
    )
    _log_remote_session_resolution_summary(
        component_id=str(payload.get("component_id") or "modal-subgraph"),
        resolution_stats=resolution_stats,
    )
    short_circuit_outputs = _short_circuit_restored_session_output_subgraph(
        payload=normalized_payload,
        hydrated_inputs=hydrated_inputs,
        session_handle=session_handle,
        resolution_stats=resolution_stats,
    )
    if short_circuit_outputs is not None:
        logger.info(
            "Skipping mapped-node execution for component=%s because all %d session-backed outputs were restored into session_id=%s.",
            payload.get("component_id"),
            len(short_circuit_outputs),
            session_handle.session_id if session_handle is not None else None,
        )
        return short_circuit_outputs
    if session_handle is not None:
        logger.info(
            "Executing mapped remote subgraph %s with remote_session session_id=%s prompt_id=%s owner_component_id=%s.",
            payload.get("component_id"),
            session_handle.session_id,
            session_handle.prompt_id,
            session_handle.owner_component_id,
        )
    prompt = copy.deepcopy(normalized_payload["subgraph_prompt"])
    logger.info(
        "Executing remote subgraph %s via test mapping with %d prompt nodes.",
        payload.get("component_id"),
        len(prompt),
    )
    _apply_boundary_inputs(
        prompt=prompt,
        boundary_input_specs=list(normalized_payload.get("boundary_inputs", [])),
        hydrated_inputs=resolved_inputs,
    )
    _coerce_prompt_primitive_input_values(prompt, node_mapping)
    _validate_prompt_input_shapes(
        prompt,
        node_mapping,
        list(normalized_payload.get("boundary_inputs", [])),
    )
    _validate_required_prompt_inputs(prompt, node_mapping)
    required_node_ids = _resolve_required_subgraph_nodes(
        prompt=prompt,
        execute_node_ids=list(normalized_payload.get("execute_node_ids", [])),
    )
    executed_outputs: dict[str, tuple[Any, ...]] = {}
    pending = set(required_node_ids)

    while pending:
        progressed = False
        logger.info("Mapped remote subgraph pending nodes: %s", sorted(pending))
        for node_id in list(sorted(pending)):
            prompt_node = prompt[node_id]
            kwargs: dict[str, Any] = {}
            unresolved_dependency = False
            for input_name, input_value in (prompt_node.get("inputs") or {}).items():
                if _is_link(input_value):
                    upstream_node_id = str(input_value[0])
                    if upstream_node_id not in executed_outputs:
                        unresolved_dependency = True
                        break
                    kwargs[str(input_name)] = executed_outputs[upstream_node_id][
                        int(input_value[1])
                    ]
                else:
                    kwargs[str(input_name)] = input_value
            if unresolved_dependency:
                continue

            class_type = str(prompt_node["class_type"])
            if class_type not in node_mapping:
                raise KeyError(f"Remote node class {class_type!r} is not registered.")
            logger.info(
                "Executing mapped remote node %s (%s) with %d inputs.",
                node_id,
                class_type,
                len(kwargs),
            )
            executed_outputs[node_id] = _invoke_original_node(
                node_mapping[class_type],
                prompt_node,
                kwargs,
            )
            pending.remove(node_id)
            progressed = True
        if not progressed:
            raise RemoteSubgraphExecutionError(
                "Unable to resolve execution order for remote subgraph payload."
            )

    outputs: list[Any] = []
    for boundary_output in normalized_payload.get("boundary_outputs", []):
        node_id = str(boundary_output["node_id"])
        output_index = int(boundary_output["output_index"])
        node_outputs = executed_outputs.get(node_id)
        if node_outputs is None:
            raise RemoteSubgraphExecutionError(
                f"Remote subgraph did not execute boundary output node {node_id}."
            )
        output_value = node_outputs[output_index]
        if bool(boundary_output.get("session_output")):
            if session_handle is None:
                raise RemoteSessionStateError(
                    "Session-backed boundary outputs require payload.remote_session."
                )
            bridge_record = _build_remote_session_bridge_record(
                payload=normalized_payload,
                hydrated_inputs=hydrated_inputs,
                node_id=node_id,
                output_index=output_index,
                io_type=str(boundary_output.get("io_type") or "*"),
                output_value=output_value,
            )
            _REMOTE_SESSION_BRIDGE_STORE.put_record(bridge_record)
            _store_remote_session_bridge_value(bridge_record.bridge_key, output_value)
            live_ref = _REMOTE_SESSION_STORE.put_bridge_output(
                session_handle,
                bridge_key=bridge_record.bridge_key,
                node_id=node_id,
                output_index=output_index,
                value=output_value,
            )
            output_value = RemoteSessionBridgeRef(
                bridge_key=bridge_record.bridge_key,
                node_id=node_id,
                output_index=output_index,
                session_id=live_ref.session_id,
            ).to_payload()
        outputs.append(output_value)
    logger.info(
        "Mapped remote subgraph %s produced %d exported outputs.",
        payload.get("component_id"),
        len(outputs),
    )
    return tuple(outputs)


def _execute_subgraph_prompt(
    payload: dict[str, Any],
    hydrated_inputs: dict[str, Any],
    node_mapping: dict[str, type[Any]] | None = None,
) -> tuple[Any, ...]:
    """Execute a remote component prompt and return its exported outputs."""
    if node_mapping is not None:
        return _execute_subgraph_with_mapping(payload, hydrated_inputs, node_mapping)

    normalized_payload = _trim_subgraph_payload_to_required_nodes(
        _normalize_subgraph_payload(payload)
    )
    session_handle = _payload_remote_session_handle(normalized_payload)
    resolution_stats = _RemoteSessionBridgeResolutionStats()
    resolved_inputs = _resolve_remote_session_inputs(
        dict(hydrated_inputs),
        component_id=str(payload.get("component_id") or ""),
        target_session_handle=session_handle,
        node_mapping=node_mapping,
        resolution_stats=resolution_stats,
    )
    _log_remote_session_resolution_summary(
        component_id=str(payload.get("component_id") or "modal-subgraph"),
        resolution_stats=resolution_stats,
    )
    short_circuit_outputs = _short_circuit_restored_session_output_subgraph(
        payload=normalized_payload,
        hydrated_inputs=hydrated_inputs,
        session_handle=session_handle,
        resolution_stats=resolution_stats,
    )
    if short_circuit_outputs is not None:
        logger.info(
            "Skipping PromptExecutor execution for component=%s because all %d session-backed outputs were restored into session_id=%s.",
            payload.get("component_id"),
            len(short_circuit_outputs),
            session_handle.session_id if session_handle is not None else None,
        )
        return short_circuit_outputs
    if session_handle is not None:
        logger.info(
            "Executing PromptExecutor remote subgraph %s with remote_session session_id=%s prompt_id=%s owner_component_id=%s.",
            payload.get("component_id"),
            session_handle.session_id,
            session_handle.prompt_id,
            session_handle.owner_component_id,
        )
    prompt = copy.deepcopy(normalized_payload["subgraph_prompt"])
    logger.info(
        "Executing remote subgraph %s through PromptExecutor with %d prompt nodes, %d boundary inputs, and %d exported outputs.",
        payload.get("component_id"),
        len(prompt),
        len(normalized_payload.get("boundary_inputs", [])),
        len(normalized_payload.get("boundary_outputs", [])),
    )
    _apply_boundary_inputs(
        prompt=prompt,
        boundary_input_specs=list(normalized_payload.get("boundary_inputs", [])),
        hydrated_inputs=resolved_inputs,
    )
    execution = _load_execution_module()
    resolved_node_mapping = _load_nodes_module().NODE_CLASS_MAPPINGS
    _coerce_prompt_primitive_input_values(prompt, resolved_node_mapping)
    _validate_prompt_input_shapes(
        prompt,
        resolved_node_mapping,
        list(normalized_payload.get("boundary_inputs", [])),
    )
    _validate_required_prompt_inputs(prompt, resolved_node_mapping)

    with _temporary_node_mapping(node_mapping):
        executor = execution.PromptExecutor(_NullPromptServer())
        execution_started_at = time.perf_counter()
        logger.info(
            "Starting PromptExecutor for remote subgraph %s with execute targets %s.",
            payload.get("component_id"),
            normalized_payload.get("execute_node_ids", []),
        )
        executor.execute(
            prompt=prompt,
            prompt_id=str(
                payload.get("prompt_id")
                or payload.get("component_id", "modal-subgraph")
            ),
            extra_data=copy.deepcopy(normalized_payload.get("extra_data") or {}),
            execute_outputs=list(normalized_payload.get("execute_node_ids", [])),
        )
        logger.info(
            "PromptExecutor finished for remote subgraph %s in %.3fs with success=%s and %d status messages.",
            payload.get("component_id"),
            time.perf_counter() - execution_started_at,
            executor.success,
            len(executor.status_messages),
        )
        if executor.status_messages:
            logger.info(
                "Remote subgraph %s status events: %s",
                payload.get("component_id"),
                [event for event, _data in executor.status_messages],
            )
        if not executor.success:
            raise RemoteSubgraphExecutionError(_extract_prompt_executor_error(executor))

        outputs: list[Any] = []
        for boundary_output in normalized_payload.get("boundary_outputs", []):
            node_id = str(boundary_output["node_id"])
            output_index = int(boundary_output["output_index"])
            cache_entry = executor.caches.outputs.get(node_id)
            if cache_entry is None:
                raise RemoteSubgraphExecutionError(
                    f"Remote subgraph did not produce cache entry for node {node_id}."
                )
            if output_index >= len(cache_entry.outputs):
                raise RemoteSubgraphExecutionError(
                    f"Remote subgraph output index {output_index} is missing for node {node_id}."
                )
            output_value = _collapse_cache_slot(
                slot_values=cache_entry.outputs[output_index],
                is_list=bool(boundary_output.get("is_list", False)),
            )
            if bool(boundary_output.get("session_output")):
                if session_handle is None:
                    raise RemoteSessionStateError(
                        "Session-backed boundary outputs require payload.remote_session."
                    )
                bridge_record = _build_remote_session_bridge_record(
                    payload=normalized_payload,
                    hydrated_inputs=hydrated_inputs,
                    node_id=node_id,
                    output_index=output_index,
                    io_type=str(boundary_output.get("io_type") or "*"),
                    output_value=output_value,
                )
                _REMOTE_SESSION_BRIDGE_STORE.put_record(bridge_record)
                _store_remote_session_bridge_value(
                    bridge_record.bridge_key, output_value
                )
                live_ref = _REMOTE_SESSION_STORE.put_bridge_output(
                    session_handle,
                    bridge_key=bridge_record.bridge_key,
                    node_id=node_id,
                    output_index=output_index,
                    value=output_value,
                )
                output_value = RemoteSessionBridgeRef(
                    bridge_key=bridge_record.bridge_key,
                    node_id=node_id,
                    output_index=output_index,
                    session_id=live_ref.session_id,
                ).to_payload()
            outputs.append(output_value)
            logger.info(
                "Collected exported output %s from node %s output %d.",
                boundary_output.get("proxy_output_name"),
                node_id,
                output_index,
            )
        return tuple(outputs)


def _short_circuit_restored_session_output_subgraph(
    *,
    payload: dict[str, Any],
    hydrated_inputs: dict[str, Any],
    session_handle: RemoteSessionHandle | None,
    resolution_stats: _RemoteSessionBridgeResolutionStats,
) -> tuple[Any, ...] | None:
    """Return session-backed outputs directly when bridge restoration already satisfied them all."""
    boundary_outputs = list(payload.get("boundary_outputs", []))
    if session_handle is None or not boundary_outputs:
        return None
    if resolution_stats.input_ref_count <= 0 or resolution_stats.replay_count > 0:
        return None
    if any(
        not bool(boundary_output.get("session_output"))
        for boundary_output in boundary_outputs
    ):
        return None

    restored_outputs: list[Any] = []
    for boundary_output in boundary_outputs:
        node_id = str(boundary_output["node_id"])
        output_index = int(boundary_output["output_index"])
        try:
            output_value = _REMOTE_SESSION_STORE.get_output(
                RemoteSessionValueRef(
                    session_id=session_handle.session_id,
                    node_id=node_id,
                    output_index=output_index,
                )
            )
        except RemoteSessionStateError:
            return None

        bridge_record = _build_remote_session_bridge_record(
            payload=payload,
            hydrated_inputs=hydrated_inputs,
            node_id=node_id,
            output_index=output_index,
            io_type=str(boundary_output.get("io_type") or "*"),
            output_value=output_value,
        )
        _REMOTE_SESSION_BRIDGE_STORE.put_record(bridge_record)
        _store_remote_session_bridge_value(bridge_record.bridge_key, output_value)
        live_ref = _REMOTE_SESSION_STORE.put_bridge_output(
            session_handle,
            bridge_key=bridge_record.bridge_key,
            node_id=node_id,
            output_index=output_index,
            value=output_value,
        )
        restored_outputs.append(
            RemoteSessionBridgeRef(
                bridge_key=bridge_record.bridge_key,
                node_id=node_id,
                output_index=output_index,
                session_id=live_ref.session_id,
            ).to_payload()
        )
    return tuple(restored_outputs)


def execute_subgraph_locally(
    payload: dict[str, Any],
    kwargs_payload: bytes | bytearray | str | dict[str, Any],
    node_mapping: dict[str, type[Any]] | None = None,
) -> bytes:
    """Execute a rewritten remote component in-process and return serialized outputs."""
    _extract_custom_nodes_bundle(payload.get("custom_nodes_bundle"))
    hydrated_inputs = deserialize_node_inputs(kwargs_payload)
    session_handle = _payload_remote_session_handle(payload)
    logger.info(
        "Executing local fallback subgraph %s with %d hydrated inputs session_id=%s clear_remote_session=%s.",
        payload.get("component_id"),
        len(hydrated_inputs),
        session_handle.session_id if session_handle is not None else None,
        bool(payload.get("clear_remote_session")),
    )
    try:
        with ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(
                _execute_subgraph_prompt, payload, hydrated_inputs, node_mapping
            )
            try:
                outputs = future.result()
            except Exception:
                logger.exception(
                    "Local fallback subgraph %s raised while running in worker thread.",
                    payload.get("component_id"),
                )
                raise
    finally:
        if bool(payload.get("clear_remote_session")) and session_handle is not None:
            logger.info(
                "Clearing remote session after component=%s session_id=%s.",
                payload.get("component_id"),
                session_handle.session_id,
            )
            _REMOTE_SESSION_STORE.clear_session(session_handle)
    logger.info(
        "Local fallback subgraph %s completed with %d outputs.",
        payload.get("component_id"),
        len(outputs),
    )
    return serialize_node_outputs(outputs)


def _modal_lookup_error_types() -> tuple[type[BaseException], ...]:
    """Return Modal exception types that indicate lookup or hydration failure."""
    if modal is None:
        return tuple()
    exception_module = getattr(modal, "exception", None)
    if exception_module is None:
        return tuple()

    error_types: list[type[BaseException]] = []
    for error_name in ("NotFoundError", "ExecutionError", "InvalidError"):
        error_type = getattr(exception_module, error_name, None)
        if isinstance(error_type, type) and issubclass(error_type, BaseException):
            error_types.append(error_type)
    return tuple(error_types)


def _is_missing_modal_deployment_error(exc: BaseException) -> bool:
    """Return whether one Modal lookup failure indicates missing deployed app state."""
    message = str(exc).strip().lower()
    if any(
        marker in message
        for marker in (
            "could not deserialize remote exception",
            "remote traceback",
            "remotesubgraphexecutionerror",
            "object of type",
            "is not json serializable",
        )
    ):
        return False
    if "not deployed" in message:
        return True
    if "not found" not in message:
        return False
    return any(
        marker in message
        for marker in (
            "lookup failed for cls",
            "app '",
            'app "',
            "class '",
            'class "',
            "not found in environment",
        )
    )


def _load_modal_cloud_module() -> Any:
    """Load the stable Modal cloud entry module under a valid Python name."""
    settings = getattr(_MODAL_CLOUD_SETTINGS_STATE, "settings", None) or get_settings()
    deployment_app_name = modal_deployment_app_name(settings)
    with _MODAL_CLOUD_MODULE_LOCK:
        existing_module = sys.modules.get(_MODAL_CLOUD_MODULE_NAME)
        existing_gpu = getattr(
            existing_module, "__comfy_modal_gpu__", settings.modal_gpu
        )
        existing_app_name = getattr(existing_module, "__comfy_modal_app_name__", None)
        existing_secret_name = getattr(
            existing_module, "__comfy_modal_secret_name__", None
        )
        if (
            existing_module is not None
            and getattr(existing_module, "app", None) is not None
            and existing_gpu == settings.modal_gpu
            and existing_app_name == deployment_app_name
            and existing_secret_name == settings.modal_secret_name
        ):
            return existing_module
        if existing_module is not None:
            logger.warning(
                "Discarding Modal cloud module %s before reload for app=%s gpu=%s secret=%s (previous_app=%s previous_gpu=%s previous_secret=%s).",
                _MODAL_CLOUD_MODULE_NAME,
                deployment_app_name,
                settings.modal_gpu,
                settings.modal_secret_name,
                existing_app_name,
                existing_gpu,
                existing_secret_name,
            )
            sys.modules.pop(_MODAL_CLOUD_MODULE_NAME, None)

        cloud_module_path = (
            Path(__file__).resolve().parents[1] / f"{_MODAL_CLOUD_MODULE_NAME}.py"
        )
        module_spec = importlib.util.spec_from_file_location(
            _MODAL_CLOUD_MODULE_NAME,
            cloud_module_path,
        )
        if module_spec is None or module_spec.loader is None:
            raise ModalRemoteInvocationError(
                f"Unable to create module spec for Modal cloud entrypoint at {cloud_module_path}."
            )

        cloud_module = importlib.util.module_from_spec(module_spec)
        setattr(cloud_module, "__comfy_modal_settings_override__", settings)
        setattr(cloud_module, "__comfy_modal_gpu__", settings.modal_gpu)
        setattr(cloud_module, "__comfy_modal_app_name__", deployment_app_name)
        setattr(cloud_module, "__comfy_modal_secret_name__", settings.modal_secret_name)
        sys.modules[_MODAL_CLOUD_MODULE_NAME] = cloud_module
        try:
            module_spec.loader.exec_module(cloud_module)
        except BaseException:
            sys.modules.pop(_MODAL_CLOUD_MODULE_NAME, None)
            raise
        return cloud_module


def _install_modal_cloud_exception_compatibility_module() -> None:
    """Expose cloud exception definitions without loading the deployable cloud app."""
    if _MODAL_CLOUD_MODULE_NAME in sys.modules:
        return

    compatibility_module = ModuleType(_MODAL_CLOUD_MODULE_NAME)
    setattr(
        compatibility_module,
        "RemoteSubgraphExecutionError",
        RemoteSubgraphExecutionError,
    )
    exception_bases: dict[str, type[BaseException]] = {
        "RemoteInvocationInProgressError": RuntimeError,
        "RemoteInvocationAbandonedError": RuntimeError,
        "RemoteCanaryInterruptedError": RuntimeError,
        "RemoteCanaryBarrierTimeoutError": TimeoutError,
        "ExistingModalAppError": RuntimeError,
    }
    for exception_name, exception_base in exception_bases.items():
        setattr(
            compatibility_module,
            exception_name,
            type(
                exception_name,
                (exception_base,),
                {"__module__": _MODAL_CLOUD_MODULE_NAME},
            ),
        )
    sys.modules[_MODAL_CLOUD_MODULE_NAME] = compatibility_module


def _lookup_local_prompt_server() -> Any | None:
    """Return the live local ComfyUI PromptServer instance when available."""
    try:
        import server
    except ModuleNotFoundError:
        return None

    return getattr(server.PromptServer, "instance", None)


def _record_local_modal_ui_event(
    event: str, payload: Mapping[str, Any], client_id: str | None
) -> None:
    """Record one local Modal UI event for browser refocus replay when available."""
    if client_id is None:
        return
    try:
        from ..api_intercept import record_modal_ui_event
    except (ImportError, AttributeError):
        logger.debug(
            "Modal UI event replay recorder is unavailable for event %s.", event
        )
        return

    record_modal_ui_event(event, payload, client_id)


def _emit_local_modal_status(
    *,
    prompt_id: str | None,
    client_id: str | None,
    phase: str,
    node_ids: list[str],
    modal_gpu: str | None = None,
    active_node_id: str | None = None,
    active_node_class_type: str | None = None,
    active_node_role: str | None = None,
    error_message: str | None = None,
    status_message: str | None = None,
    status_current: int | None = None,
    status_total: int | None = None,
    completed_ancestor_node_ids: list[str] | None = None,
    execution_provider: str | None = None,
    execution_environment_id: str | None = None,
    execution_location: str | None = None,
) -> None:
    """Forward remote execution progress into the local ComfyUI websocket stream."""
    if client_id is None:
        return

    prompt_server = _lookup_local_prompt_server()
    if prompt_server is None:
        return

    payload: dict[str, Any] = {
        "phase": phase,
        "prompt_id": prompt_id,
        "node_ids": list(node_ids),
    }
    if modal_gpu is not None:
        payload["modal_gpu"] = modal_gpu
    if active_node_id is not None:
        payload["active_node_id"] = active_node_id
    if active_node_class_type is not None:
        payload["active_node_class_type"] = active_node_class_type
    if active_node_role is not None:
        payload["active_node_role"] = active_node_role
    if error_message is not None:
        payload["error_message"] = error_message
    if status_message is not None:
        payload["status_message"] = status_message
    if status_current is not None:
        payload["status_current"] = int(status_current)
    if status_total is not None:
        payload["status_total"] = int(status_total)
    if completed_ancestor_node_ids:
        payload["completed_ancestor_node_ids"] = list(completed_ancestor_node_ids)
    if execution_provider:
        payload["execution_provider"] = execution_provider
    if execution_environment_id:
        payload["execution_environment_id"] = execution_environment_id
    if execution_location:
        payload["execution_location"] = execution_location
    _record_local_modal_ui_event("modal_status", payload, client_id)
    prompt_server.send_sync("modal_status", payload, client_id)


def _remote_execution_destination(payload: Mapping[str, Any]) -> str:
    """Return the user-facing destination for a provider-stamped payload."""
    environment_id = str(payload.get("execution_environment_id") or "").strip()
    provider = str(payload.get("execution_provider") or "modal").strip()
    return (
        environment_id
        if provider != "modal" and environment_id
        else "Modal"
    )


def _remote_execution_identity(
    payload: Mapping[str, Any],
    modal_task_id: str | None = None,
) -> dict[str, str]:
    """Return non-empty provider, environment, and runtime location fields."""
    provider = str(
        payload.get("execution_provider") or ("modal" if modal_task_id else "")
    ).strip()
    environment_id = str(payload.get("execution_environment_id") or "").strip()
    location = (
        modal_task_id
        if provider == "modal" and modal_task_id
        else str(payload.get("execution_location") or "").strip()
    )
    return {
        key: value
        for key, value in {
            "execution_provider": provider,
            "execution_environment_id": environment_id,
            "execution_location": location,
        }.items()
        if value
    }


def _emit_local_remote_dispatch_status(payload: dict[str, Any]) -> None:
    """Tell the local UI a remote component was dispatched before progress streams."""
    _emit_local_remote_startup_status(
        payload,
        phase="starting",
        status_message=(
            f"Starting remote component on {_remote_execution_destination(payload)}"
        ),
    )


def _emit_local_remote_startup_status(
    payload: Mapping[str, Any],
    *,
    phase: str,
    status_message: str,
) -> None:
    """Show one prompt-scoped Modal startup phase in the local global status pill."""
    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    extra_data = payload.get("extra_data") or {}
    client_id = (
        str(extra_data.get("client_id"))
        if extra_data.get("client_id") is not None
        else None
    )
    node_ids = [
        str(node_id)
        for node_id in payload.get("component_node_ids", [])
        if str(node_id)
    ]
    if not node_ids and payload.get("component_id") is not None:
        node_ids = [str(payload["component_id"])]
    if not prompt_id or not client_id or not node_ids:
        return
    _emit_local_modal_status(
        prompt_id=prompt_id,
        client_id=client_id,
        phase=phase,
        node_ids=node_ids,
        modal_gpu=(
            str(payload["modal_gpu"]) if payload.get("modal_gpu") is not None else None
        ),
        status_message=status_message,
        **_remote_execution_identity(payload),
    )


def _emit_local_modal_progress(
    *,
    prompt_id: str | None,
    client_id: str | None,
    node_id: str,
    value: float,
    max_value: float,
    display_node_id: str | None = None,
    real_node_id: str | None = None,
    lane_id: str | None = None,
    clear: bool = False,
    item_index: int | None = None,
    aggregate_only: bool = False,
    setup_only: bool = False,
    cached_hit: bool = False,
    completed_ancestor_node_ids: list[str] | None = None,
    stage: str | None = None,
    message: str | None = None,
    unit: str | None = None,
    indeterminate: bool = False,
    elapsed_seconds: float | None = None,
    time_to_first_token_seconds: float | None = None,
    tokens_per_second: float | None = None,
    pre_gpu: bool = False,
    execution_provider: str | None = None,
    execution_environment_id: str | None = None,
    execution_location: str | None = None,
) -> None:
    """Forward remote numeric and stage progress into the local websocket stream."""
    if client_id is None:
        return

    prompt_server = _lookup_local_prompt_server()
    if prompt_server is None:
        return

    payload: dict[str, Any] = {
        "prompt_id": prompt_id,
        "node_id": node_id,
        "value": float(value),
        "max": float(max_value),
    }
    if display_node_id is not None:
        payload["display_node_id"] = display_node_id
    if real_node_id is not None:
        payload["real_node_id"] = real_node_id
    if lane_id is not None:
        payload["lane_id"] = lane_id
    if clear:
        payload["clear"] = True
    if item_index is not None:
        payload["item_index"] = int(item_index)
    if aggregate_only:
        payload["aggregate_only"] = True
    if setup_only:
        payload["setup_only"] = True
    if cached_hit:
        payload["cached_hit"] = True
    if completed_ancestor_node_ids:
        payload["completed_ancestor_node_ids"] = list(completed_ancestor_node_ids)
    if stage:
        payload["stage"] = stage
    if message:
        payload["message"] = message
    if unit:
        payload["unit"] = unit
    if indeterminate:
        payload["indeterminate"] = True
    if elapsed_seconds is not None:
        payload["elapsed_seconds"] = float(elapsed_seconds)
    if time_to_first_token_seconds is not None:
        payload["time_to_first_token_seconds"] = float(
            time_to_first_token_seconds
        )
    if tokens_per_second is not None:
        payload["tokens_per_second"] = float(tokens_per_second)
    if pre_gpu:
        payload["pre_gpu"] = True
    if execution_provider:
        payload["execution_provider"] = execution_provider
    if execution_environment_id:
        payload["execution_environment_id"] = execution_environment_id
    if execution_location:
        payload["execution_location"] = execution_location
    _record_local_modal_ui_event("modal_progress", payload, client_id)
    prompt_server.send_sync("modal_progress", payload, client_id)


def _emit_local_executed_output(
    *,
    prompt_id: str | None,
    client_id: str | None,
    node_id: str,
    display_node_id: str | None,
    output_payload: Any,
) -> None:
    """Forward one remote node's executed UI payload into the local websocket stream."""
    if client_id is None:
        return

    prompt_server = _lookup_local_prompt_server()
    if prompt_server is None:
        return

    payload = {
        "prompt_id": prompt_id,
        "node": node_id,
        "display_node": display_node_id or node_id,
        "output": output_payload,
    }
    prompt_server.send_sync("executed", payload, client_id)


def _emit_local_preview_image(
    *,
    prompt_id: str | None,
    client_id: str | None,
    node_id: str,
    display_node_id: str | None,
    parent_node_id: str | None,
    real_node_id: str | None,
    image_type: str,
    image_bytes: bytes,
    max_size: int | None,
) -> None:
    """Forward one remote preview image into the local ComfyUI preview websocket path."""
    if client_id is None:
        return

    prompt_server = _lookup_local_prompt_server()
    if prompt_server is None:
        return

    try:
        from PIL import Image
        from protocol import BinaryEventTypes
    except ModuleNotFoundError:
        logger.warning(
            "Preview forwarding is unavailable because Pillow or ComfyUI protocol imports failed."
        )
        return

    with BytesIO(image_bytes) as image_buffer:
        image = Image.open(image_buffer)
        image.load()

    metadata: dict[str, Any] = {
        "node_id": node_id,
        "prompt_id": prompt_id,
        "display_node_id": display_node_id or node_id,
        "real_node_id": real_node_id or node_id,
    }
    if parent_node_id is not None:
        metadata["parent_node_id"] = parent_node_id

    prompt_server.send_sync(
        BinaryEventTypes.PREVIEW_IMAGE_WITH_METADATA,
        ((image_type, image, max_size), metadata),
        client_id,
    )


def _emit_local_preview_boundary_output(
    *,
    prompt_id: str | None,
    client_id: str | None,
    preview_target_node_ids: list[str],
    image_value: Any,
) -> None:
    """Render one streamed remote boundary IMAGE value into local PreviewImage UI events."""
    if client_id is None or not preview_target_node_ids:
        return

    prompt_server = _lookup_local_prompt_server()
    if prompt_server is None:
        return

    try:
        import nodes
    except ModuleNotFoundError:
        logger.warning(
            "Preview boundary streaming is unavailable because ComfyUI nodes could not be imported."
        )
        return

    preview_factory = getattr(nodes, "PreviewImage", None)
    if preview_factory is None:
        logger.warning(
            "Preview boundary streaming is unavailable because PreviewImage is not registered."
        )
        return

    preview_result = preview_factory().save_images(images=image_value)
    if not isinstance(preview_result, dict):
        return
    output_payload = preview_result.get("ui")
    if not isinstance(output_payload, dict):
        return

    for preview_target_node_id in preview_target_node_ids:
        prompt_server.send_sync(
            "executed",
            {
                "prompt_id": prompt_id,
                "node": preview_target_node_id,
                "display_node": preview_target_node_id,
                "output": output_payload,
            },
            client_id,
        )


def _allowed_suppressed_stream_node_ids(payload: dict[str, Any]) -> set[str]:
    """Return the node ids that may surface UI events for a suppressed mapped/static stream."""
    allowed_node_ids = {
        str(node_id) for node_id in payload.get("execute_node_ids", []) if str(node_id)
    }
    allowed_node_ids.update(
        str(boundary_output["node_id"])
        for boundary_output in payload.get("boundary_outputs", [])
        if boundary_output.get("node_id") is not None
        and str(boundary_output["node_id"])
    )
    return allowed_node_ids


def _should_forward_suppressed_stream_event(
    payload: dict[str, Any],
    reported_node_id: Any,
) -> bool:
    """Return whether a suppressed mapped/static stream event belongs to this payload."""
    if not bool(payload.get("suppress_status_stream")):
        return True
    if reported_node_id is None:
        return False
    allowed_node_ids = _allowed_suppressed_stream_node_ids(payload)
    if not allowed_node_ids:
        return True
    return str(reported_node_id) in allowed_node_ids


def _progress_stream_event_node_id(stream_event: dict[str, Any]) -> str | None:
    """Return the best node id to use for progress filtering and forwarding."""
    for candidate in (
        stream_event.get("real_node_id"),
        stream_event.get("node_id"),
        stream_event.get("display_node_id"),
    ):
        if candidate is None:
            continue
        candidate_text = str(candidate)
        if candidate_text:
            return candidate_text
    return None


def _progress_stream_event_metadata(
    stream_event: dict[str, Any]
) -> dict[str, str | None] | None:
    """Return normalized metadata for one streamed progress event."""
    reported_node_id = (
        str(stream_event["node_id"])
        if stream_event.get("node_id") is not None
        else None
    )
    display_node_id = (
        str(stream_event["display_node_id"])
        if stream_event.get("display_node_id") is not None
        else reported_node_id
    )
    real_node_id = (
        str(stream_event["real_node_id"])
        if stream_event.get("real_node_id") is not None
        else None
    )
    filter_node_id = real_node_id or reported_node_id or display_node_id
    if filter_node_id is None:
        return None
    return {
        "node_id": reported_node_id or filter_node_id,
        "display_node_id": display_node_id,
        "real_node_id": real_node_id,
        "filter_node_id": filter_node_id,
    }


def _remote_prompt_ancestor_node_ids(
    payload: dict[str, Any], node_id: str | None
) -> list[str]:
    """Return subgraph dependency ancestors for one currently executing remote node."""
    if node_id is None:
        return []
    prompt = payload.get("subgraph_prompt", {})
    if not isinstance(prompt, dict) or node_id not in prompt:
        return []

    ancestors: set[str] = set()
    pending = [node_id]
    while pending:
        current_node_id = str(pending.pop())
        prompt_node = prompt.get(current_node_id)
        if not isinstance(prompt_node, dict):
            continue
        inputs = prompt_node.get("inputs") or {}
        if not isinstance(inputs, dict):
            continue
        for input_value in inputs.values():
            for input_link in _iter_prompt_links(input_value):
                upstream_node_id = str(input_link[0])
                if upstream_node_id == node_id or upstream_node_id in ancestors:
                    continue
                if upstream_node_id not in prompt:
                    continue
                ancestors.add(upstream_node_id)
                pending.append(upstream_node_id)
    return sorted(ancestors)


def _should_stream_remote_progress(payload: dict[str, Any]) -> bool:
    """Return whether the local client has enough context to mirror remote node progress."""
    extra_data = payload.get("extra_data") or {}
    return (
        payload.get("payload_kind") in {"subgraph", "mapped_subgraph"}
        and isinstance(payload.get("prompt_id"), str)
        and bool(payload.get("prompt_id"))
        and isinstance(extra_data.get("client_id"), str)
        and bool(extra_data.get("client_id"))
        and isinstance(payload.get("component_node_ids"), list)
        and len(payload.get("component_node_ids")) > 0
    )


def _local_processing_interrupted() -> bool:
    """Return whether the current local ComfyUI execution was interrupted."""
    try:
        import comfy.model_management
    except ModuleNotFoundError:
        return False

    return bool(comfy.model_management.processing_interrupted())


def _raise_local_interrupt() -> None:
    """Raise ComfyUI's native interruption exception for the current execution."""
    import comfy.model_management

    raise comfy.model_management.InterruptProcessingException()


def _exception_indicates_interruption(exc: BaseException) -> bool:
    """Return whether an exception represents cancellation or interrupted execution."""
    if isinstance(exc, asyncio.CancelledError):
        return True
    message = str(exc).lower()
    return "interrupt" in message or "cancel" in message


def _remote_interrupt_key(payload: dict[str, Any]) -> tuple[str, str]:
    """Return the prompt/component pair used to interrupt one remote execution."""
    prompt_id = str(
        payload.get("prompt_id") or payload.get("component_id") or "modal-subgraph"
    )
    component_id = str(payload.get("component_id") or "single-node")
    return prompt_id, component_id


def _remote_interrupt_flag_key(prompt_id: str, component_id: str) -> str:
    """Return the shared Modal interrupt-store key for one payload execution."""
    return f"{prompt_id}:{component_id}"


def _lookup_modal_interrupt_store() -> Any | None:
    """Return the shared Modal Dict used to signal remote cancellation requests."""
    if modal is None or not hasattr(modal, "Dict"):
        return None

    settings = get_settings()
    cache_key = (settings.interrupt_dict_name, _modal_environment_name())
    with _MODAL_INTERRUPT_DICTS_LOCK:
        cached_store = _MODAL_INTERRUPT_DICTS.get(cache_key)
        if cached_store is not None:
            return cached_store

    interrupt_store = modal.Dict.from_name(
        settings.interrupt_dict_name,
        environment_name=cache_key[1],
        create_if_missing=True,
    )
    with _MODAL_INTERRUPT_DICTS_LOCK:
        _MODAL_INTERRUPT_DICTS[cache_key] = interrupt_store
    return interrupt_store


def _remote_interrupt_flag_value() -> dict[str, float]:
    """Return the shared Modal interrupt-store value for one cancellation request."""
    return {"requested_at": time.time()}


def _write_remote_interrupt_flag(
    interrupt_store: Any, prompt_id: str, component_id: str
) -> None:
    """Write one remote cancellation request with the blocking Modal Dict API."""
    interrupt_store.put(
        _remote_interrupt_flag_key(prompt_id, component_id),
        _remote_interrupt_flag_value(),
    )


async def _write_remote_interrupt_flag_async(
    interrupt_store: Any,
    prompt_id: str,
    component_id: str,
) -> None:
    """Write one remote cancellation request without blocking the async caller."""
    put_method = getattr(interrupt_store, "put", None)
    put_async = getattr(put_method, "aio", None)
    if callable(put_async):
        result = put_async(
            _remote_interrupt_flag_key(prompt_id, component_id),
            _remote_interrupt_flag_value(),
        )
        if inspect.isawaitable(result):
            await result
        return

    await asyncio.to_thread(
        _write_remote_interrupt_flag, interrupt_store, prompt_id, component_id
    )


def _request_remote_interrupt(payload: dict[str, Any]) -> bool:
    """Write one remote cancellation request into the shared Modal interrupt store."""
    _abandon_local_modal_workflow_gate(payload, "local interrupt requested")
    interrupt_store = _lookup_modal_interrupt_store()
    if interrupt_store is None:
        return False

    prompt_id, component_id = _remote_interrupt_key(payload)
    _write_remote_interrupt_flag(interrupt_store, prompt_id, component_id)
    logger.info(
        "Propagated local interrupt to Modal prompt=%s component=%s through shared control state.",
        prompt_id,
        component_id,
    )
    return True


async def _request_remote_interrupt_async(payload: dict[str, Any]) -> bool:
    """Write one remote cancellation request into the shared Modal interrupt store asynchronously."""
    _abandon_local_modal_workflow_gate(payload, "local interrupt requested")
    interrupt_store = await asyncio.to_thread(_lookup_modal_interrupt_store)
    if interrupt_store is None:
        return False

    prompt_id, component_id = _remote_interrupt_key(payload)
    await _write_remote_interrupt_flag_async(interrupt_store, prompt_id, component_id)
    logger.info(
        "Propagated local interrupt to Modal prompt=%s component=%s through shared control state.",
        prompt_id,
        component_id,
    )
    return True


def active_remote_modal_prompt_ids() -> set[str]:
    """Return prompt ids that currently have local proxies waiting on Modal work."""
    with _ACTIVE_REMOTE_INVOCATIONS_LOCK:
        return set(_ACTIVE_REMOTE_INVOCATIONS_BY_PROMPT)


@contextmanager
def _registered_active_remote_invocation(
    payload: dict[str, Any],
    cancellation_event: threading.Event | None,
    interrupt_remote_call: Callable[[], Any] | None,
) -> Iterator[None]:
    """Register one active Modal call so targeted ComfyUI interrupts can find it."""
    prompt_id, component_id = _remote_interrupt_key(payload)
    invocation = _ActiveRemoteInvocation(
        prompt_id=prompt_id,
        component_id=component_id,
        cancellation_event=cancellation_event,
        interrupt_remote_call=interrupt_remote_call,
    )
    with _ACTIVE_REMOTE_INVOCATIONS_LOCK:
        prompt_invocations = _ACTIVE_REMOTE_INVOCATIONS_BY_PROMPT.setdefault(
            prompt_id, {}
        )
        prompt_invocations[component_id] = invocation
    logger.info(
        "Registered active Modal invocation prompt=%s component=%s for targeted cancellation.",
        prompt_id,
        component_id,
    )
    try:
        yield
    finally:
        with _ACTIVE_REMOTE_INVOCATIONS_LOCK:
            prompt_invocations = _ACTIVE_REMOTE_INVOCATIONS_BY_PROMPT.get(prompt_id)
            if prompt_invocations is not None:
                prompt_invocations.pop(component_id, None)
                if not prompt_invocations:
                    _ACTIVE_REMOTE_INVOCATIONS_BY_PROMPT.pop(prompt_id, None)
        logger.info(
            "Unregistered active Modal invocation prompt=%s component=%s.",
            prompt_id,
            component_id,
        )


def request_remote_modal_prompt_interrupt(prompt_id: str) -> bool:
    """Request cancellation for every active Modal invocation belonging to one prompt."""
    normalized_prompt_id = str(prompt_id)
    _abandon_local_modal_workflow_gate(
        {"prompt_id": normalized_prompt_id},
        "prompt-level interrupt requested",
    )
    with _ACTIVE_REMOTE_INVOCATIONS_LOCK:
        invocations = list(
            _ACTIVE_REMOTE_INVOCATIONS_BY_PROMPT.get(normalized_prompt_id, {}).values()
        )
    if not invocations:
        return False

    logger.info(
        "Requesting remote Modal cancellation for prompt=%s across %d active component(s).",
        normalized_prompt_id,
        len(invocations),
    )
    for invocation in invocations:
        if invocation.cancellation_event is not None:
            invocation.cancellation_event.set()
        _propagate_remote_interrupt_request(
            {
                "prompt_id": invocation.prompt_id,
                "component_id": invocation.component_id,
            },
            invocation.interrupt_remote_call,
        )
    return True


async def request_remote_modal_prompt_interrupt_async(prompt_id: str) -> bool:
    """Request cancellation for every active Modal invocation belonging to one prompt asynchronously."""
    normalized_prompt_id = str(prompt_id)
    _abandon_local_modal_workflow_gate(
        {"prompt_id": normalized_prompt_id},
        "prompt-level interrupt requested",
    )
    with _ACTIVE_REMOTE_INVOCATIONS_LOCK:
        invocations = list(
            _ACTIVE_REMOTE_INVOCATIONS_BY_PROMPT.get(normalized_prompt_id, {}).values()
        )
    if not invocations:
        return False

    logger.info(
        "Requesting async remote Modal cancellation for prompt=%s across %d active component(s).",
        normalized_prompt_id,
        len(invocations),
    )
    for invocation in invocations:
        if invocation.cancellation_event is not None:
            invocation.cancellation_event.set()
        await _request_remote_interrupt_async(
            {"prompt_id": invocation.prompt_id, "component_id": invocation.component_id}
        )
    return True


def _sync_local_interrupt_to_cancellation_event(
    payload: dict[str, Any],
    cancellation_event: threading.Event | None,
) -> bool:
    """Mirror ComfyUI's interrupt flag into the current Modal cancellation event."""
    if cancellation_event is not None and cancellation_event.is_set():
        return True
    if not _local_processing_interrupted():
        return False
    if cancellation_event is not None and not cancellation_event.is_set():
        logger.info(
            "Observed local interrupt while Modal component=%s was running; requesting remote cancellation.",
            payload.get("component_id"),
        )
        cancellation_event.set()
    _abandon_local_modal_workflow_gate(payload, "observed local interrupt")
    return True


def _abandon_local_modal_workflow_gate(payload: dict[str, Any], reason: str) -> None:
    """Release the local prompt gate for a Modal prompt that ComfyUI has cancelled."""
    prompt_id = payload.get("prompt_id")
    if prompt_id is None:
        return

    try:
        from ..modal_executor_node import abandon_modal_workflow_execution_prompt
    except ImportError:
        logger.debug(
            "Unable to import Modal workflow gate helper while abandoning prompt."
        )
        return

    abandon_modal_workflow_execution_prompt(str(prompt_id), reason)


def _propagate_remote_interrupt_request(
    payload: dict[str, Any],
    interrupt_remote_call: Callable[[], Any] | None,
) -> None:
    """Send one best-effort remote cancellation request for an active Modal payload."""
    prompt_id, component_id = _remote_interrupt_key(payload)
    if interrupt_remote_call is None:
        logger.warning(
            "Local interrupt requested for component=%s, but no remote interrupt method is available.",
            component_id,
        )
        return
    try:
        interrupt_remote_call()
        logger.info(
            "Propagated local interrupt to Modal prompt=%s component=%s.",
            prompt_id,
            component_id,
        )
    except Exception:
        logger.exception(
            "Failed to propagate local interrupt to Modal prompt=%s component=%s.",
            prompt_id,
            component_id,
        )


def _handle_modal_wait_cancellation(
    payload: dict[str, Any],
    cancellation_event: threading.Event,
    *,
    interrupt_sent: bool,
    cancellation_started_at: float | None,
) -> tuple[bool, float | None]:
    """Propagate and bound local waiting after cancellation during any Modal wait phase."""
    if not _sync_local_interrupt_to_cancellation_event(payload, cancellation_event):
        return interrupt_sent, cancellation_started_at

    if not interrupt_sent:
        _request_remote_interrupt(payload)
        return True, time.monotonic()

    if cancellation_started_at is None:
        return interrupt_sent, time.monotonic()

    grace_seconds = max(0.0, get_settings().remote_cancel_grace_seconds)
    if time.monotonic() - cancellation_started_at >= grace_seconds:
        logger.info(
            "Modal component=%s did not reach a cancellable remote call within %.3fs of local interrupt; releasing the local prompt while remote cancellation continues.",
            payload.get("component_id"),
            grace_seconds,
        )
        raise ModalRemoteInvocationError(
            "Remote Modal call did not reach a cancellable remote phase after local interrupt propagation."
        )

    return interrupt_sent, cancellation_started_at


async def _handle_modal_wait_cancellation_async(
    payload: dict[str, Any],
    cancellation_event: threading.Event,
    *,
    interrupt_sent: bool,
    cancellation_started_at: float | None,
) -> tuple[bool, float | None]:
    """Propagate and bound local waiting after cancellation during an async Modal wait phase."""
    if not _sync_local_interrupt_to_cancellation_event(payload, cancellation_event):
        return interrupt_sent, cancellation_started_at

    if not interrupt_sent:
        await _request_remote_interrupt_async(payload)
        return True, time.monotonic()

    if cancellation_started_at is None:
        return interrupt_sent, time.monotonic()

    grace_seconds = max(0.0, get_settings().remote_cancel_grace_seconds)
    if time.monotonic() - cancellation_started_at >= grace_seconds:
        logger.info(
            "Modal component=%s did not reach a cancellable remote call within %.3fs of local interrupt; releasing the local prompt while remote cancellation continues.",
            payload.get("component_id"),
            grace_seconds,
        )
        raise ModalRemoteInvocationError(
            "Remote Modal call did not reach a cancellable remote phase after local interrupt propagation."
        )

    return interrupt_sent, cancellation_started_at


def _invoke_remote_call_with_interrupts(
    *,
    payload: dict[str, Any],
    invoke_remote_call: Callable[[], bytes],
    interrupt_remote_call: Callable[[], Any] | None,
    cancellation_event: threading.Event | None,
) -> bytes:
    """Run one blocking remote call while optionally propagating cancellation to Modal."""
    result_queue: queue.Queue[tuple[str, Any]] = queue.Queue()
    cancellation_started_at: float | None = None

    def execute_remote_call() -> None:
        """Run the blocking Modal request in a worker thread."""
        try:
            result_queue.put(("result", invoke_remote_call()))
        except BaseException as exc:
            result_queue.put(("error", exc))

    request_thread = threading.Thread(
        target=execute_remote_call,
        name=f"modal-request-{payload.get('component_id', 'payload')}",
        daemon=True,
    )
    request_thread.start()
    interrupt_sent = False
    try:
        with _registered_active_remote_invocation(
            payload, cancellation_event, interrupt_remote_call
        ):
            while True:
                try:
                    result_kind, result_payload = result_queue.get(timeout=0.1)
                except queue.Empty:
                    if _sync_local_interrupt_to_cancellation_event(
                        payload, cancellation_event
                    ):
                        if not interrupt_sent:
                            _propagate_remote_interrupt_request(
                                payload, interrupt_remote_call
                            )
                            interrupt_sent = True
                            cancellation_started_at = time.monotonic()
                        elif cancellation_started_at is not None:
                            grace_seconds = max(
                                0.0, get_settings().remote_cancel_grace_seconds
                            )
                            if (
                                time.monotonic() - cancellation_started_at
                                >= grace_seconds
                            ):
                                logger.info(
                                    "Modal component=%s did not return within %.3fs of local interrupt propagation; releasing the local prompt while remote cancellation continues.",
                                    payload.get("component_id"),
                                    grace_seconds,
                                )
                                raise ModalRemoteInvocationError(
                                    "Remote Modal call did not finish after local interrupt propagation."
                                )
                    continue

                if result_kind == "result":
                    return bytes(result_payload)
                raise result_payload
    finally:
        request_thread.join(
            timeout=0.1
            if cancellation_event is not None and cancellation_event.is_set()
            else 1.0
        )


def _consume_remote_payload_stream(
    payload: dict[str, Any],
    stream_events: Iterator[dict[str, Any]],
) -> bytes:
    """Forward remote progress events into the local UI and return the final payload bytes."""
    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    extra_data = payload.get("extra_data") or {}
    client_id = (
        str(extra_data.get("client_id"))
        if extra_data.get("client_id") is not None
        else None
    )
    node_ids = [str(node_id) for node_id in payload.get("component_node_ids", [])]
    modal_gpu = (
        str(payload["modal_gpu"]) if payload.get("modal_gpu") is not None else None
    )
    suppress_status_stream = bool(payload.get("suppress_status_stream"))
    result_payload: bytes | bytearray | None = None
    suppressed_progress_node_metadata: dict[str, dict[str, str | None]] = {}
    active_remote_task_id: str | None = None
    active_remote_log_task_id: str | None = None
    should_close_stream = False
    component_id = str(payload.get("component_id") or "payload")
    invocation_id = str(payload.get("invocation_id") or "none")
    stream_started_at = time.monotonic()
    previous_event_at = stream_started_at
    event_count = 0
    progress_event_count = 0
    speculative_prewarm_checked = False
    logger.info(
        "Starting local Modal stream consumption component=%s prompt_id=%s invocation_id=%s.",
        component_id,
        prompt_id or "none",
        invocation_id,
    )

    try:
        for stream_event in stream_events:
            event_received_at = time.monotonic()
            seconds_since_previous_event = event_received_at - previous_event_at
            previous_event_at = event_received_at
            event_count += 1
            if not speculative_prewarm_checked:
                speculative_prewarm_checked = True
                _schedule_speculative_affinity_prewarm(
                    payload,
                    reason="current_remote_stream_started",
                )
            event_kind = str(stream_event.get("kind", ""))
            if event_kind == "remote_logs":
                task_id = _coerce_modal_task_id(stream_event.get("task_id"))
                if task_id is not None and active_remote_task_id is None:
                    active_remote_task_id = task_id
                if (
                    task_id is not None
                    and active_remote_log_task_id is None
                    and _is_remote_container_log_stream_enabled()
                ):
                    active_remote_log_task_id = _retain_remote_container_log_stream(
                        task_id
                    )
                continue
            if event_kind == "progress":
                progress_event_count += 1
                event_type = str(stream_event.get("event_type", ""))
                if event_type == "node_progress":
                    progress_metadata = _progress_stream_event_metadata(stream_event)
                    filter_node_id = (
                        progress_metadata["filter_node_id"]
                        if progress_metadata is not None
                        else None
                    )
                    lane_id = (
                        str(stream_event["lane_id"])
                        if stream_event.get("lane_id") is not None
                        else (
                            str(payload["mapped_progress_lane_id"])
                            if payload.get("mapped_progress_lane_id") is not None
                            else None
                        )
                    )
                    aggregate_only = bool(stream_event.get("aggregate_only", False))
                    if (
                        suppress_status_stream
                        and lane_id is None
                        and not aggregate_only
                        and not _should_forward_suppressed_stream_event(
                            payload, filter_node_id
                        )
                    ):
                        logger.debug(
                            "Suppressing streamed Modal node progress for component=%s node_id=%s real_node_id=%s because it does not belong to this mapped/static payload.",
                            payload.get("component_id"),
                            stream_event.get("node_id"),
                            stream_event.get("real_node_id"),
                        )
                        continue
                    reported_node_id = (
                        progress_metadata["node_id"]
                        if progress_metadata is not None
                        else None
                    )
                    if reported_node_id is not None:
                        display_node_id = (
                            progress_metadata["display_node_id"]
                            if progress_metadata is not None
                            else str(reported_node_id)
                        )
                        real_node_id = (
                            progress_metadata["real_node_id"]
                            if progress_metadata is not None
                            else None
                        )
                        progress_node_id = real_node_id or display_node_id
                        completed_ancestor_node_ids = _remote_prompt_ancestor_node_ids(
                            payload,
                            progress_node_id,
                        )
                        if lane_id is not None:
                            _remember_mapped_lane_node_id(
                                payload, lane_id, progress_node_id
                            )
                        elif (
                            suppress_status_stream
                            and not aggregate_only
                            and progress_metadata is not None
                        ):
                            suppressed_progress_node_metadata[
                                str(progress_metadata["filter_node_id"])
                            ] = {
                                "node_id": str(reported_node_id),
                                "display_node_id": display_node_id,
                                "real_node_id": real_node_id,
                            }
                        logger.debug(
                            "Forwarding streamed Modal node progress for component=%s node_id=%s real_node_id=%s value=%s max=%s lane_id=%s.",
                            payload.get("component_id"),
                            reported_node_id,
                            real_node_id,
                            stream_event.get("value"),
                            stream_event.get("max"),
                            lane_id,
                        )
                        progress_kwargs = {
                            "prompt_id": prompt_id,
                            "client_id": client_id,
                            "node_id": str(reported_node_id),
                            "value": float(stream_event.get("value", 0.0)),
                            "max_value": float(stream_event.get("max", 1.0)),
                            "display_node_id": display_node_id,
                            "real_node_id": real_node_id,
                            "lane_id": lane_id,
                            "clear": bool(stream_event.get("clear", False)),
                            "item_index": (
                                int(stream_event["item_index"])
                                if stream_event.get("item_index") is not None
                                else (
                                    int(payload["map_item_index"])
                                    if payload.get("map_item_index") is not None
                                    else None
                                )
                            ),
                            "aggregate_only": aggregate_only,
                        }
                        progress_kwargs.update(
                            _remote_execution_identity(payload, active_remote_task_id)
                        )
                        for string_field in ("stage", "message", "unit"):
                            if stream_event.get(string_field) is not None:
                                progress_kwargs[string_field] = str(
                                    stream_event[string_field]
                                )
                        if stream_event.get("indeterminate") is not None:
                            progress_kwargs["indeterminate"] = bool(
                                stream_event["indeterminate"]
                            )
                        for numeric_field in (
                            "elapsed_seconds",
                            "time_to_first_token_seconds",
                            "tokens_per_second",
                        ):
                            if stream_event.get(numeric_field) is not None:
                                progress_kwargs[numeric_field] = float(
                                    stream_event[numeric_field]
                                )
                        if completed_ancestor_node_ids:
                            progress_kwargs[
                                "completed_ancestor_node_ids"
                            ] = completed_ancestor_node_ids
                        _emit_local_modal_progress(**progress_kwargs)
                    continue
                if event_type == "executed":
                    reported_node_id = stream_event.get("node_id")
                    if reported_node_id is not None:
                        if not _should_forward_suppressed_stream_event(
                            payload, reported_node_id
                        ):
                            logger.debug(
                                "Suppressing streamed Modal executed output for component=%s node_id=%s because it does not belong to this mapped/static payload.",
                                payload.get("component_id"),
                                reported_node_id,
                            )
                            continue
                        logger.debug(
                            "Forwarding streamed Modal executed output for component=%s node_id=%s.",
                            payload.get("component_id"),
                            reported_node_id,
                        )
                        _emit_local_executed_output(
                            prompt_id=prompt_id,
                            client_id=client_id,
                            node_id=str(reported_node_id),
                            display_node_id=(
                                str(stream_event["display_node_id"])
                                if stream_event.get("display_node_id") is not None
                                else None
                            ),
                            output_payload=deserialize_value(
                                stream_event.get("output")
                            ),
                        )
                    continue
                if event_type == "preview":
                    reported_node_id = stream_event.get("node_id")
                    image_bytes = deserialize_value(stream_event.get("image_bytes"))
                    if reported_node_id is not None and isinstance(image_bytes, bytes):
                        if not _should_forward_suppressed_stream_event(
                            payload, reported_node_id
                        ):
                            logger.debug(
                                "Suppressing streamed Modal preview image for component=%s node_id=%s because it does not belong to this mapped/static payload.",
                                payload.get("component_id"),
                                reported_node_id,
                            )
                            continue
                        logger.debug(
                            "Forwarding streamed Modal preview image for component=%s node_id=%s.",
                            payload.get("component_id"),
                            reported_node_id,
                        )
                        _emit_local_preview_image(
                            prompt_id=prompt_id,
                            client_id=client_id,
                            node_id=str(reported_node_id),
                            display_node_id=(
                                str(stream_event["display_node_id"])
                                if stream_event.get("display_node_id") is not None
                                else None
                            ),
                            parent_node_id=(
                                str(stream_event["parent_node_id"])
                                if stream_event.get("parent_node_id") is not None
                                else None
                            ),
                            real_node_id=(
                                str(stream_event["real_node_id"])
                                if stream_event.get("real_node_id") is not None
                                else None
                            ),
                            image_type=str(stream_event.get("image_type", "PNG")),
                            image_bytes=image_bytes,
                            max_size=(
                                int(stream_event["max_size"])
                                if stream_event.get("max_size") is not None
                                else None
                            ),
                        )
                    continue
                if event_type == "boundary_output":
                    preview_target_node_ids = [
                        str(node_id)
                        for node_id in stream_event.get("preview_target_node_ids", [])
                        if str(node_id)
                    ]
                    if preview_target_node_ids:
                        logger.debug(
                            "Forwarding streamed Modal boundary output previews for component=%s source_node=%s targets=%s.",
                            payload.get("component_id"),
                            stream_event.get("node_id"),
                            preview_target_node_ids,
                        )
                        _emit_local_preview_boundary_output(
                            prompt_id=prompt_id,
                            client_id=client_id,
                            preview_target_node_ids=preview_target_node_ids,
                            image_value=deserialize_value(stream_event.get("value")),
                        )
                    continue
                if event_type == "node_cached":
                    progress_metadata = _progress_stream_event_metadata(stream_event)
                    filter_node_id = (
                        progress_metadata["filter_node_id"]
                        if progress_metadata is not None
                        else None
                    )
                    if (
                        suppress_status_stream
                        and not _should_forward_suppressed_stream_event(
                            payload, filter_node_id
                        )
                    ):
                        logger.debug(
                            "Suppressing streamed Modal cached-node marker for component=%s node_id=%s real_node_id=%s because it does not belong to this mapped/static payload.",
                            payload.get("component_id"),
                            stream_event.get("node_id"),
                            stream_event.get("real_node_id"),
                        )
                        continue
                    reported_node_id = (
                        progress_metadata["node_id"]
                        if progress_metadata is not None
                        else None
                    )
                    if reported_node_id is not None:
                        display_node_id = (
                            progress_metadata["display_node_id"]
                            if progress_metadata is not None
                            else str(reported_node_id)
                        )
                        real_node_id = (
                            progress_metadata["real_node_id"]
                            if progress_metadata is not None
                            else None
                        )
                        logger.debug(
                            "Forwarding streamed Modal cached-node marker for component=%s node_id=%s real_node_id=%s.",
                            payload.get("component_id"),
                            reported_node_id,
                            real_node_id,
                        )
                        _emit_local_modal_progress(
                            prompt_id=prompt_id,
                            client_id=client_id,
                            node_id=str(reported_node_id),
                            value=0.0,
                            max_value=1.0,
                            display_node_id=display_node_id,
                            real_node_id=real_node_id,
                            cached_hit=True,
                            **_remote_execution_identity(
                                payload, active_remote_task_id
                            ),
                        )
                    continue
                logger.info(
                    "Forwarding streamed Modal progress for component=%s phase=%s active_node_id=%s.",
                    payload.get("component_id"),
                    stream_event.get("phase"),
                    stream_event.get("active_node_id"),
                )
                if suppress_status_stream:
                    remote_phase = str(stream_event.get("phase", "executing"))
                    if remote_phase in {
                        "execution_success",
                        "execution_error",
                        "execution_interrupted",
                    }:
                        for (
                            progress_metadata
                        ) in suppressed_progress_node_metadata.values():
                            _emit_local_modal_progress(
                                prompt_id=prompt_id,
                                client_id=client_id,
                                node_id=str(progress_metadata["node_id"]),
                                value=0.0,
                                max_value=1.0,
                                display_node_id=progress_metadata["display_node_id"],
                                real_node_id=progress_metadata["real_node_id"],
                                clear=True,
                            )
                        suppressed_progress_node_metadata.clear()
                    continue
                remote_phase = str(stream_event.get("phase", "executing"))
                if remote_phase == "execution_success":
                    _emit_local_modal_status(
                        prompt_id=prompt_id,
                        client_id=client_id,
                        phase="finalizing",
                        node_ids=node_ids,
                        modal_gpu=modal_gpu,
                        status_message=(
                            "Receiving remote outputs from "
                            f"{_remote_execution_destination(payload)}"
                        ),
                        **_remote_execution_identity(payload, active_remote_task_id),
                    )
                    continue
                _emit_local_modal_status(
                    prompt_id=prompt_id,
                    client_id=client_id,
                    phase=remote_phase,
                    node_ids=node_ids,
                    modal_gpu=modal_gpu,
                    active_node_id=(
                        str(stream_event["active_node_id"])
                        if stream_event.get("active_node_id") is not None
                        else None
                    ),
                    completed_ancestor_node_ids=_remote_prompt_ancestor_node_ids(
                        payload,
                        str(stream_event["active_node_id"])
                        if stream_event.get("active_node_id") is not None
                        else None,
                    )
                    or None,
                    active_node_class_type=(
                        str(stream_event["active_node_class_type"])
                        if stream_event.get("active_node_class_type") is not None
                        else None
                    ),
                    active_node_role=(
                        str(stream_event["active_node_role"])
                        if stream_event.get("active_node_role") is not None
                        else None
                    ),
                    **_remote_execution_identity(payload, active_remote_task_id),
                )
                continue
            if event_kind == "result":
                candidate_outputs = stream_event.get("outputs")
                candidate_bytes = (
                    len(candidate_outputs)
                    if isinstance(candidate_outputs, bytes | bytearray)
                    else -1
                )
                logger.info(
                    "Received streamed Modal result component=%s prompt_id=%s invocation_id=%s "
                    "result_bytes=%d stream_elapsed_seconds=%.3f seconds_since_previous_event=%.3f "
                    "event_count=%d progress_event_count=%d.",
                    component_id,
                    prompt_id or "none",
                    invocation_id,
                    candidate_bytes,
                    event_received_at - stream_started_at,
                    seconds_since_previous_event,
                    event_count,
                    progress_event_count,
                )
                try:
                    result_payload = coerce_serialized_node_outputs(candidate_outputs)
                except TypeError as exc:
                    raise ModalRemoteInvocationError(
                        "Modal streamed payload result did not include transport-safe outputs."
                    ) from exc
                should_close_stream = True
                break
            logger.debug(
                "Ignoring unexpected streamed Modal event kind=%s for component=%s.",
                event_kind,
                payload.get("component_id"),
            )
    finally:
        if should_close_stream:
            close_started_at = time.monotonic()
            logger.info(
                "Starting local Modal result stream close component=%s prompt_id=%s invocation_id=%s.",
                component_id,
                prompt_id or "none",
                invocation_id,
            )
            _close_remote_payload_stream(stream_events)
            logger.info(
                "Finished local Modal result stream close in %.3fs component=%s prompt_id=%s "
                "invocation_id=%s.",
                time.monotonic() - close_started_at,
                component_id,
                prompt_id or "none",
                invocation_id,
            )
        if active_remote_log_task_id is not None:
            _release_remote_container_log_stream(active_remote_log_task_id)

    if result_payload is None:
        raise ModalRemoteInvocationError(
            f"Modal streamed payload for component={payload.get('component_id')!r} did not yield a final result."
        )
    logger.info(
        "Finished local Modal stream consumption in %.3fs component=%s prompt_id=%s invocation_id=%s "
        "result_bytes=%d event_count=%d progress_event_count=%d.",
        time.monotonic() - stream_started_at,
        component_id,
        prompt_id or "none",
        invocation_id,
        len(result_payload),
        event_count,
        progress_event_count,
    )
    return bytes(result_payload)


def _mapped_execution_parallelism(item_count: int) -> int:
    """Return the local worker width used to schedule mapped Modal item executions."""
    settings = get_settings()
    configured_limit = settings.max_inflight_calls
    if settings.max_containers is not None:
        configured_limit = min(configured_limit, settings.max_containers)
    return max(1, min(item_count, configured_limit))


def _mapped_phase_definition(
    payload: dict[str, Any], phase_key: str
) -> dict[str, Any] | None:
    """Return one explicit mapped phase definition when queue-time planning provided it."""
    phase_payload = payload.get(phase_key)
    if isinstance(phase_payload, dict):
        return phase_payload
    return None


def _shared_subgraph_payload_fields(payload: dict[str, Any]) -> dict[str, Any]:
    """Return the payload fields shared by every explicit mapped phase."""
    shared_fields = {
        "prompt_id": payload.get("prompt_id"),
        "extra_data": copy.deepcopy(payload.get("extra_data") or {}),
        "requires_volume_reload": bool(payload.get("requires_volume_reload", True)),
        "volume_reload_marker": payload.get("volume_reload_marker"),
        "uploaded_volume_paths": list(payload.get("uploaded_volume_paths", [])),
        "terminate_container_on_error": bool(
            payload.get("terminate_container_on_error", True)
        ),
        "custom_nodes_bundle": payload.get("custom_nodes_bundle"),
    }
    snapshot_profile_key = payload.get("snapshot_profile_key")
    if isinstance(snapshot_profile_key, str) and snapshot_profile_key.strip():
        shared_fields["snapshot_profile_key"] = snapshot_profile_key.strip()
    remote_session = payload.get("remote_session")
    if remote_session is not None:
        shared_fields["remote_session"] = copy.deepcopy(remote_session)
    if bool(payload.get("clear_remote_session")):
        shared_fields["clear_remote_session"] = True
    return shared_fields


def _build_phase_subgraph_payload(
    payload: dict[str, Any],
    phase_key: str,
    component_id: str,
    *,
    suppress_status_stream: bool = False,
    lane_id: str | None = None,
    item_index: int | None = None,
) -> dict[str, Any]:
    """Return one explicit static or mapped subgraph payload."""
    phase_definition = _mapped_phase_definition(payload, phase_key)
    if phase_definition is None:
        raise KeyError(f"Mapped payload is missing phase definition {phase_key!r}.")

    phase_payload = {
        "payload_kind": "subgraph",
        "component_id": component_id,
        **_shared_subgraph_payload_fields(payload),
        "component_node_ids": [
            str(node_id)
            for node_id in phase_definition.get("component_node_ids", [])
            if str(node_id)
        ],
        "subgraph_prompt": copy.deepcopy(phase_definition.get("subgraph_prompt", {})),
        "boundary_inputs": copy.deepcopy(phase_definition.get("boundary_inputs", [])),
        "boundary_outputs": copy.deepcopy(phase_definition.get("boundary_outputs", [])),
        "execute_node_ids": [
            str(node_id)
            for node_id in phase_definition.get("execute_node_ids", [])
            if str(node_id)
        ],
    }
    if suppress_status_stream:
        phase_payload["suppress_status_stream"] = True
    if lane_id is not None:
        phase_payload["mapped_progress_lane_id"] = str(lane_id)
        phase_payload["mapped_progress_display_node_id"] = str(
            payload.get("component_id", "modal-subgraph")
        )
    if item_index is not None:
        phase_payload["map_item_index"] = int(item_index)
    return phase_payload


def _split_phase_outputs(
    phase_outputs: tuple[Any, ...],
    boundary_outputs: list[dict[str, Any]],
    internal_output_names: set[str],
) -> tuple[dict[str, Any], tuple[Any, ...]]:
    """Split one phase result tuple into bridge values and external outputs."""
    internal_outputs: dict[str, Any] = {}
    external_outputs: list[Any] = []
    for boundary_output, output_value in zip(
        boundary_outputs, phase_outputs, strict=True
    ):
        output_name = str(boundary_output.get("proxy_output_name") or "")
        if output_name in internal_output_names:
            internal_outputs[output_name] = output_value
            continue
        external_outputs.append(output_value)
    return internal_outputs, tuple(external_outputs)


def _execute_mapped_subgraph_payload(
    payload: dict[str, Any],
    hydrated_inputs: dict[str, Any],
    node_mapping: dict[str, type[Any]] | None = None,
) -> tuple[Any, ...]:
    """Execute one mapped payload locally using explicit static and mapped phases."""
    mapped_input = payload.get("mapped_input") or {}
    mapped_input_name = str(mapped_input.get("proxy_input_name") or "")
    if not mapped_input_name:
        raise ModalRemoteInvocationError(
            "Mapped remote payloads must define mapped_input.proxy_input_name."
        )
    if mapped_input_name not in hydrated_inputs:
        raise KeyError(
            f"Mapped remote payload input {mapped_input_name!r} was not provided."
        )

    mapped_items = split_mapped_value(
        hydrated_inputs[mapped_input_name],
        str(mapped_input.get("io_type", "*")),
    )
    if not mapped_items:
        raise ValueError("Mapped remote execution requires at least one input item.")

    broadcast_inputs = dict(hydrated_inputs)
    broadcast_inputs.pop(mapped_input_name, None)
    static_to_mapped_boundaries = list(payload.get("static_to_mapped_boundaries", []))
    bridge_output_names = {
        str(boundary_spec.get("proxy_name") or "")
        for boundary_spec in static_to_mapped_boundaries
        if str(boundary_spec.get("proxy_name") or "")
    }

    static_outputs: tuple[Any, ...] = ()
    static_phase_payload: dict[str, Any] | None = None
    if payload.get("static_phase") is not None:
        static_phase_payload = _build_phase_subgraph_payload(
            payload,
            "static_phase",
            f"{payload.get('component_id', 'modal-subgraph')}::static",
            suppress_status_stream=True,
        )
        static_phase_payload.pop("clear_remote_session", None)
    elif payload.get("static_execute_node_ids"):
        static_phase_payload = _build_static_mapped_payload(payload)

    if static_phase_payload is not None:
        if static_phase_payload.get("execute_node_ids"):
            logger.info(
                "Executing static mapped phase for component=%s with execute nodes=%s.",
                payload.get("component_id"),
                static_phase_payload.get("execute_node_ids", []),
            )
            static_phase_outputs = _execute_subgraph_prompt(
                static_phase_payload,
                dict(broadcast_inputs),
                node_mapping,
            )
            bridge_inputs, static_outputs = _split_phase_outputs(
                static_phase_outputs,
                list(static_phase_payload.get("boundary_outputs", [])),
                bridge_output_names,
            )
            broadcast_inputs.update(bridge_inputs)

    total_items = len(mapped_items)
    _emit_local_mapped_progress(payload, 0, total_items)
    per_item_outputs: list[tuple[Any, ...]] = []
    for item_index, item_value in enumerate(mapped_items):
        if _local_processing_interrupted():
            _raise_local_interrupt()
        if payload.get("mapped_phase") is not None:
            item_payload = _build_phase_subgraph_payload(
                payload,
                "mapped_phase",
                f"{payload.get('component_id', 'modal-subgraph')}::item:{item_index}",
                suppress_status_stream=True,
                lane_id="0",
                item_index=item_index,
            )
        else:
            item_payload = _build_mapped_item_payload(payload, item_index, 0)
        item_inputs = dict(broadcast_inputs)
        item_inputs[mapped_input_name] = item_value
        logger.info(
            "Executing mapped item %d/%d for component=%s with execute nodes=%s.",
            item_index + 1,
            total_items,
            payload.get("component_id"),
            item_payload.get("execute_node_ids", []),
        )
        per_item_outputs.append(
            _execute_subgraph_prompt(
                item_payload,
                item_inputs,
                node_mapping,
            )
        )
        _emit_local_mapped_progress(payload, item_index + 1, total_items)

    if payload.get("mapped_phase") is not None:
        mapped_phase_payload = _build_phase_subgraph_payload(
            payload,
            "mapped_phase",
            f"{payload.get('component_id', 'modal-subgraph')}::mapped",
        )
    else:
        mapped_phase_payload = {
            **payload,
            "boundary_outputs": [
                boundary_output
                for boundary_output in payload.get("boundary_outputs", [])
                if _is_mapped_boundary_output(boundary_output, payload)
            ],
        }
    mapped_outputs = _aggregate_mapped_outputs(
        per_item_outputs,
        {
            **payload,
            "boundary_outputs": list(mapped_phase_payload.get("boundary_outputs", [])),
        },
    )
    return _merge_static_and_mapped_outputs(
        static_outputs=static_outputs,
        mapped_outputs=mapped_outputs,
        payload=payload,
    )


def _build_mapped_item_payload(
    payload: dict[str, Any],
    item_index: int,
    lane_index: int,
) -> dict[str, Any]:
    """Return one per-item subgraph payload derived from a mapped remote component payload."""
    if _mapped_phase_definition(payload, "mapped_phase") is not None:
        return _build_phase_subgraph_payload(
            payload,
            "mapped_phase",
            f"{payload.get('component_id', 'modal-subgraph')}::item:{item_index}",
            suppress_status_stream=True,
            lane_id=str(lane_index),
            item_index=item_index,
        )
    item_payload = copy.deepcopy(payload)
    item_payload["payload_kind"] = "subgraph"
    item_payload[
        "component_id"
    ] = f"{payload.get('component_id', 'modal-subgraph')}::item:{item_index}"
    item_payload["mapped_input"] = None
    item_payload["suppress_status_stream"] = True
    item_payload["map_item_index"] = item_index
    item_payload["mapped_progress_lane_id"] = str(lane_index)
    item_payload["mapped_progress_display_node_id"] = str(
        payload.get("component_id", "modal-subgraph")
    )
    item_payload.pop("clear_remote_session", None)
    item_payload["execute_node_ids"] = list(
        payload.get("mapped_execute_node_ids") or payload.get("execute_node_ids", [])
    )
    item_payload["boundary_outputs"] = [
        copy.deepcopy(boundary_output)
        for boundary_output in payload.get("boundary_outputs", [])
        if _is_mapped_boundary_output(boundary_output, payload)
    ]
    return item_payload


def _aggregate_mapped_outputs(
    per_item_outputs: list[tuple[Any, ...]],
    payload: dict[str, Any],
) -> tuple[Any, ...]:
    """Reassemble ordered per-item outputs from mapped execution into one proxy result tuple."""
    if not per_item_outputs:
        raise ValueError("Mapped execution produced no per-item outputs to aggregate.")

    output_count = len(per_item_outputs[0])
    if any(len(item_outputs) != output_count for item_outputs in per_item_outputs):
        raise RemoteSubgraphExecutionError(
            "Mapped remote execution produced inconsistent output arity."
        )

    boundary_outputs = list(payload.get("boundary_outputs", []))
    aggregated_outputs: list[Any] = []
    for output_index in range(output_count):
        boundary_output = (
            boundary_outputs[output_index]
            if output_index < len(boundary_outputs)
            else {}
        )
        aggregated_outputs.append(
            join_mapped_values_for_scheduler(
                [item_outputs[output_index] for item_outputs in per_item_outputs],
                io_type=str(boundary_output.get("io_type", "*")),
                is_list=bool(boundary_output.get("is_list", False)),
                scheduler_is_list=bool(boundary_output.get("scheduler_is_list", False)),
            )
        )
    return tuple(aggregated_outputs)


def _build_remote_session_cleanup_payload(
    payload: dict[str, Any]
) -> dict[str, Any] | None:
    """Return a dedicated cleanup payload that clears one shared remote session once."""
    remote_session = payload.get("remote_session")
    if not bool(
        payload.get("clear_remote_session")
    ) or not is_remote_session_handle_payload(remote_session):
        return None
    return {
        "payload_kind": "subgraph",
        "component_id": f"{payload.get('component_id', 'modal-subgraph')}::cleanup",
        **_shared_subgraph_payload_fields(payload),
        "component_node_ids": [],
        "subgraph_prompt": {},
        "boundary_inputs": [],
        "boundary_outputs": [],
        "execute_node_ids": [],
        "remote_session": copy.deepcopy(remote_session),
        "clear_remote_session": True,
        "suppress_status_stream": True,
        "terminate_container_on_error": False,
    }


def _is_mapped_boundary_output(
    boundary_output: dict[str, Any], payload: dict[str, Any]
) -> bool:
    """Return whether one boundary output belongs to the mapped per-item branch."""
    mapped_output = boundary_output.get("mapped_output")
    if mapped_output is not None:
        return bool(mapped_output)
    return bool(payload.get("mapped_input")) and not bool(
        payload.get("static_execute_node_ids")
    )


def _build_static_mapped_payload(payload: dict[str, Any]) -> dict[str, Any]:
    """Return the one-time static subgraph payload for a hybrid mapped component."""
    if _mapped_phase_definition(payload, "static_phase") is not None:
        return _build_phase_subgraph_payload(
            payload,
            "static_phase",
            f"{payload.get('component_id', 'modal-subgraph')}::static",
            suppress_status_stream=True,
        )
    static_payload = copy.deepcopy(payload)
    static_payload["payload_kind"] = "subgraph"
    static_payload[
        "component_id"
    ] = f"{payload.get('component_id', 'modal-subgraph')}::static"
    static_payload["mapped_input"] = None
    static_payload["suppress_status_stream"] = True
    static_payload.pop("clear_remote_session", None)
    static_payload["execute_node_ids"] = list(
        payload.get("static_execute_node_ids") or []
    )
    static_payload["boundary_outputs"] = [
        copy.deepcopy(boundary_output)
        for boundary_output in payload.get("boundary_outputs", [])
        if not _is_mapped_boundary_output(boundary_output, payload)
    ]
    return static_payload


def _merge_static_and_mapped_outputs(
    *,
    static_outputs: tuple[Any, ...],
    mapped_outputs: tuple[Any, ...],
    payload: dict[str, Any],
) -> tuple[Any, ...]:
    """Reassemble one hybrid mapped component's static and mapped outputs in original order."""
    combined_outputs: list[Any] = []
    static_output_index = 0
    mapped_output_index = 0

    for boundary_output in payload.get("boundary_outputs", []):
        if _is_mapped_boundary_output(boundary_output, payload):
            if mapped_output_index >= len(mapped_outputs):
                raise RemoteSubgraphExecutionError(
                    "Mapped remote execution returned fewer mapped outputs than expected."
                )
            combined_outputs.append(mapped_outputs[mapped_output_index])
            mapped_output_index += 1
            continue
        if static_output_index >= len(static_outputs):
            raise RemoteSubgraphExecutionError(
                "Mapped remote execution returned fewer static outputs than expected."
            )
        combined_outputs.append(static_outputs[static_output_index])
        static_output_index += 1

    if static_output_index != len(static_outputs) or mapped_output_index != len(
        mapped_outputs
    ):
        raise RemoteSubgraphExecutionError(
            "Mapped remote execution produced extra outputs that did not match the declared boundary outputs."
        )
    return tuple(combined_outputs)


def _emit_local_mapped_progress(
    payload: dict[str, Any],
    completed_items: int,
    total_items: int,
) -> None:
    """Emit one aggregate mapped-execution progress update for the component representative node."""
    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    extra_data = payload.get("extra_data") or {}
    client_id = (
        str(extra_data.get("client_id"))
        if extra_data.get("client_id") is not None
        else None
    )
    display_node_id = str(
        payload.get("mapped_progress_display_node_id")
        or payload.get("component_id")
        or ""
    )
    if not prompt_id or not client_id or not display_node_id:
        return
    _emit_local_modal_progress(
        prompt_id=prompt_id,
        client_id=client_id,
        node_id=display_node_id,
        value=float(completed_items),
        max_value=float(total_items),
        display_node_id=display_node_id,
        aggregate_only=True,
    )


def _mapped_progress_owner_component_id(payload: dict[str, Any]) -> str | None:
    """Return the stable component id used to track one mapped worker lane locally."""
    owner_component_id = payload.get(
        "mapped_progress_display_node_id", payload.get("component_id")
    )
    if owner_component_id is None:
        return None
    owner_component = str(owner_component_id)
    return owner_component or None


def _remember_mapped_lane_node_id(
    payload: dict[str, Any], lane_id: str, node_id: str
) -> None:
    """Remember the last real node id that emitted progress for one mapped worker lane."""
    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    owner_component_id = _mapped_progress_owner_component_id(payload)
    if not prompt_id or not owner_component_id or not node_id:
        return
    with _MAPPED_PROGRESS_NODE_IDS_LOCK:
        _MAPPED_PROGRESS_NODE_IDS[(prompt_id, owner_component_id, lane_id)] = node_id


def _pop_mapped_lane_node_id(payload: dict[str, Any], lane_id: str) -> str | None:
    """Forget and return the last real node id that emitted progress for one mapped worker lane."""
    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    owner_component_id = _mapped_progress_owner_component_id(payload)
    if not prompt_id or not owner_component_id:
        return None
    with _MAPPED_PROGRESS_NODE_IDS_LOCK:
        return _MAPPED_PROGRESS_NODE_IDS.pop(
            (prompt_id, owner_component_id, lane_id), None
        )


def _clear_local_mapped_lane_progress(
    payload: dict[str, Any],
    lane_index: int,
    item_index: int,
) -> None:
    """Remove one mapped worker lane from the local node overlay."""
    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    extra_data = payload.get("extra_data") or {}
    client_id = (
        str(extra_data.get("client_id"))
        if extra_data.get("client_id") is not None
        else None
    )
    lane_id = str(lane_index)
    display_node_id = _pop_mapped_lane_node_id(payload, lane_id) or str(
        payload.get("component_id") or ""
    )
    if not prompt_id or not client_id or not display_node_id:
        return
    _emit_local_modal_progress(
        prompt_id=prompt_id,
        client_id=client_id,
        node_id=display_node_id,
        value=0.0,
        max_value=1.0,
        display_node_id=display_node_id,
        lane_id=lane_id,
        clear=True,
        item_index=item_index,
    )


def _emit_local_mapped_lane_progress_start(
    payload: dict[str, Any],
    lane_index: int,
    item_index: int | None = None,
) -> None:
    """Create or reset one mapped worker lane before remote progress begins arriving."""
    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    extra_data = payload.get("extra_data") or {}
    client_id = (
        str(extra_data.get("client_id"))
        if extra_data.get("client_id") is not None
        else None
    )
    display_node_id = str(payload.get("component_id") or "")
    if not prompt_id or not client_id or not display_node_id:
        return
    _emit_local_modal_progress(
        prompt_id=prompt_id,
        client_id=client_id,
        node_id=display_node_id,
        value=0.0,
        max_value=1.0,
        display_node_id=display_node_id,
        lane_id=str(lane_index),
        item_index=item_index,
        setup_only=True,
    )


def _implicit_batch_preserving_targets(
    payload: dict[str, Any],
    boundary_input: dict[str, Any],
) -> list[str]:
    """Return target sockets that must receive one complete tensor batch."""
    prompt = payload.get("subgraph_prompt", {})
    if not isinstance(prompt, dict):
        return []

    preserving_targets: list[str] = []
    for target in boundary_input.get("targets", []):
        node_id = str(target.get("node_id") or "")
        input_name = str(target.get("input_name") or "")
        prompt_node = prompt.get(node_id)
        if not node_id or not input_name or not isinstance(prompt_node, dict):
            continue
        class_type = str(prompt_node.get("class_type") or "")
        if (class_type, input_name) in _IMPLICIT_BATCH_PRESERVING_TARGETS:
            preserving_targets.append(f"{node_id}.{input_name}")
    return sorted(preserving_targets)


def _split_batch_boundary_inputs(
    payload: dict[str, Any],
    hydrated_inputs: dict[str, Any],
) -> tuple[dict[str, list[Any]], int] | None:
    """Return zipped per-item boundary inputs when an ordinary subgraph receives batched values."""
    implicitly_batchable_scalar_io_types = frozenset(
        {"BOOLEAN", "FLOAT", "INT", "STRING"}
    )
    implicitly_batchable_transport_io_types = frozenset(
        {"IMAGE", "LATENT", "MASK", "NOISE", "SIGMAS"}
    )
    split_inputs: dict[str, list[Any]] = {}

    def is_session_ref_list(value: Any) -> bool:
        """Return whether `value` is a non-empty list of remote session refs."""
        return (
            isinstance(value, list)
            and len(value) > 0
            and all(
                is_remote_session_value_ref_payload(item)
                or is_remote_session_bridge_ref_payload(item)
                for item in value
            )
        )

    def unwrap_singleton_session_ref_list(value: Any) -> Any:
        """Unwrap Comfy's output-list wrapper around mapped remote session refs."""
        if (
            isinstance(value, list)
            and len(value) == 1
            and is_session_ref_list(value[0])
        ):
            return value[0]
        return value

    for boundary_input in payload.get("boundary_inputs", []):
        proxy_input_name = str(boundary_input.get("proxy_input_name") or "")
        if not proxy_input_name or proxy_input_name not in hydrated_inputs:
            continue
        input_value = unwrap_singleton_session_ref_list(
            hydrated_inputs[proxy_input_name]
        )
        io_type = _implicit_batch_boundary_effective_io_type(
            payload=payload,
            boundary_input=boundary_input,
            input_value=input_value,
        )
        preserving_targets = _implicit_batch_preserving_targets(
            payload,
            boundary_input,
        )
        if preserving_targets:
            logger.info(
                "Skipping implicit batch split for boundary input %s io_type=%s "
                "because target sockets consume the complete tensor batch: %s.",
                proxy_input_name,
                io_type,
                preserving_targets,
            )
            continue
        input_is_session_ref_list = is_session_ref_list(input_value)
        if (
            isinstance(input_value, list)
            and not is_mapped_output_value(input_value)
            and not input_is_session_ref_list
            and io_type
            not in (
                implicitly_batchable_scalar_io_types
                | implicitly_batchable_transport_io_types
            )
        ):
            logger.info(
                "Skipping implicit batch split for boundary input %s io_type=%s because list-backed non-scalar values stay broadcast.",
                proxy_input_name,
                io_type,
            )
            continue
        try:
            items = split_mapped_value(
                input_value,
                io_type,
            )
        except (TypeError, ValueError):
            continue
        if len(items) <= 1:
            continue
        split_inputs[proxy_input_name] = items

    if not split_inputs:
        return None

    item_counts = {input_name: len(items) for input_name, items in split_inputs.items()}
    unique_counts = set(item_counts.values())
    if len(unique_counts) != 1:
        raise ModalRemoteInvocationError(
            "Implicit Modal batch boundary inputs must all have the same item count. "
            f"Received counts: {item_counts!r}"
        )
    return split_inputs, next(iter(unique_counts))


def _is_latent_like_mapping(value: Any) -> bool:
    """Return whether one runtime value looks like a ComfyUI LATENT mapping."""
    return isinstance(value, Mapping) and "samples" in value


def _list_is_latent_like_batch(value: Any) -> bool:
    """Return whether one runtime value is a list of LATENT-like mappings."""
    return (
        isinstance(value, list)
        and len(value) > 0
        and all(_is_latent_like_mapping(item) for item in value)
    )


def _implicit_batch_boundary_target_input_types(
    payload: dict[str, Any],
    boundary_input: dict[str, Any],
) -> set[str]:
    """Return the declared input types of one boundary input's target sockets."""
    prompt = payload.get("subgraph_prompt", {})
    if not isinstance(prompt, dict):
        return set()

    try:
        resolved_node_mapping = _load_nodes_module().NODE_CLASS_MAPPINGS
    except ModuleNotFoundError:
        logger.debug(
            "Skipping target input type discovery for implicit batching because ComfyUI nodes are unavailable."
        )
        return set()

    target_input_types: set[str] = set()
    for target in boundary_input.get("targets", []):
        node_id = str(target.get("node_id") or "")
        input_name = str(target.get("input_name") or "")
        if not node_id or not input_name:
            continue
        prompt_node = prompt.get(node_id)
        if not isinstance(prompt_node, dict):
            continue
        node_class = resolved_node_mapping.get(str(prompt_node.get("class_type")))
        if node_class is None:
            continue
        prompt_inputs = prompt_node.get("inputs") or {}
        declared_type = _node_input_type_map(node_class, prompt_inputs).get(input_name)
        if isinstance(declared_type, str) and declared_type:
            target_input_types.add(declared_type)
    return target_input_types


def _implicit_batch_boundary_effective_io_type(
    *,
    payload: dict[str, Any],
    boundary_input: dict[str, Any],
    input_value: Any,
) -> str:
    """Return the best effective io_type for implicit batching of one boundary input."""
    declared_io_type = str(boundary_input.get("io_type", "*"))
    if declared_io_type != "*":
        return declared_io_type

    if _list_is_latent_like_batch(input_value):
        return "LATENT"

    target_input_types = _implicit_batch_boundary_target_input_types(
        payload, boundary_input
    )
    if len(target_input_types) == 1:
        return next(iter(target_input_types))
    return declared_io_type


def _implicit_batch_input_is_list_target_node_ids(
    payload: dict[str, Any],
    split_inputs: dict[str, list[Any]],
) -> list[str]:
    """Return split-boundary target nodes that must consume the full list in one execution."""
    prompt = payload.get("subgraph_prompt", {})
    if not isinstance(prompt, dict):
        return []

    try:
        resolved_node_mapping = _load_nodes_module().NODE_CLASS_MAPPINGS
    except ModuleNotFoundError:
        logger.debug(
            "Skipping INPUT_IS_LIST detection for implicit batching because ComfyUI nodes are unavailable."
        )
        return []
    target_node_ids: set[str] = set()
    for boundary_input in payload.get("boundary_inputs", []):
        proxy_input_name = str(boundary_input.get("proxy_input_name") or "")
        if proxy_input_name not in split_inputs:
            continue
        for target in boundary_input.get("targets", []):
            target_node_id = str(target.get("node_id") or "")
            if not target_node_id:
                continue
            prompt_node = prompt.get(target_node_id)
            if prompt_node is None:
                continue
            class_type = str(prompt_node.get("class_type"))
            node_class = resolved_node_mapping.get(class_type)
            if node_class is None:
                continue
            if bool(getattr(node_class, "INPUT_IS_LIST", False)):
                target_node_ids.add(target_node_id)
    return sorted(target_node_ids)


def _partition_implicit_batched_execute_nodes(
    payload: dict[str, Any],
    split_inputs: dict[str, list[Any]],
) -> tuple[list[str], list[str]]:
    """Split one implicitly batched subgraph into static and per-item execute targets."""
    prompt = payload.get("subgraph_prompt", {})
    if not isinstance(prompt, dict):
        execute_node_ids = [
            str(node_id) for node_id in payload.get("execute_node_ids", [])
        ]
        return [], execute_node_ids

    batched_target_node_ids: set[str] = set()
    for boundary_input in payload.get("boundary_inputs", []):
        proxy_input_name = str(boundary_input.get("proxy_input_name") or "")
        if proxy_input_name not in split_inputs:
            continue
        for target in boundary_input.get("targets", []):
            target_node_id = target.get("node_id")
            if target_node_id is not None:
                batched_target_node_ids.add(str(target_node_id))

    execute_node_ids = [str(node_id) for node_id in payload.get("execute_node_ids", [])]
    static_execute_node_ids: list[str] = []
    mapped_execute_node_ids: list[str] = []
    for execute_node_id in execute_node_ids:
        required_node_ids = set(
            _resolve_required_subgraph_nodes(
                prompt=prompt,
                execute_node_ids=[execute_node_id],
            )
        )
        if required_node_ids & batched_target_node_ids:
            mapped_execute_node_ids.append(execute_node_id)
            continue
        static_execute_node_ids.append(execute_node_id)

    if not mapped_execute_node_ids and execute_node_ids:
        logger.warning(
            "Implicitly batched Modal component=%s had batched inputs %s but no execute target depended on them; "
            "falling back to per-item execution for all execute nodes.",
            payload.get("component_id"),
            sorted(split_inputs),
        )
        return [], execute_node_ids

    logger.info(
        "Partitioned implicitly batched Modal component=%s into static execute nodes=%s and mapped execute nodes=%s.",
        payload.get("component_id"),
        static_execute_node_ids,
        mapped_execute_node_ids,
    )
    return static_execute_node_ids, mapped_execute_node_ids


def _annotate_implicit_batched_boundary_outputs(
    payload: dict[str, Any],
    mapped_execute_node_ids: list[str],
) -> list[dict[str, Any]]:
    """Mark which boundary outputs belong to the per-item branch of an implicitly batched subgraph."""
    prompt = payload.get("subgraph_prompt", {})
    if not isinstance(prompt, dict) or not mapped_execute_node_ids:
        return [
            copy.deepcopy(boundary_output)
            for boundary_output in payload.get("boundary_outputs", [])
        ]

    mapped_required_node_ids = set(
        _resolve_required_subgraph_nodes(
            prompt=prompt,
            execute_node_ids=[str(node_id) for node_id in mapped_execute_node_ids],
        )
    )
    annotated_outputs: list[dict[str, Any]] = []
    for boundary_output in payload.get("boundary_outputs", []):
        annotated_output = copy.deepcopy(boundary_output)
        annotated_output["mapped_output"] = (
            str(boundary_output.get("node_id")) in mapped_required_node_ids
        )
        annotated_outputs.append(annotated_output)
    return annotated_outputs


def _log_detached_mapped_lane_result(
    task: asyncio.Task[None],
    *,
    component_id: str,
    lane_index: int,
) -> None:
    """Consume and log any late failure from one detached mapped worker lane."""
    try:
        task.result()
    except asyncio.CancelledError:
        return
    except BaseException as exc:
        logger.warning(
            "Detached mapped worker lane=%d for component=%s finished with an unawaited failure: %s",
            lane_index,
            component_id,
            exc,
        )


async def _invoke_implicitly_mapped_subgraph_async(
    payload: dict[str, Any], kwargs_payload: bytes
) -> bytes:
    """Fan out one ordinary subgraph payload when batchable boundary inputs arrive zipped."""
    hydrated_inputs = deserialize_node_inputs(kwargs_payload)
    split_batch_inputs = _split_batch_boundary_inputs(payload, hydrated_inputs)
    if split_batch_inputs is None:
        raise ModalRemoteInvocationError(
            "Implicit mapped subgraph execution requires at least one batched boundary input."
        )

    split_inputs, total_items = split_batch_inputs
    input_is_list_target_node_ids = _implicit_batch_input_is_list_target_node_ids(
        payload,
        split_inputs,
    )
    if input_is_list_target_node_ids:
        logger.info(
            "Executing implicitly batched Modal component=%s as one ordinary subgraph because split boundary inputs target INPUT_IS_LIST nodes=%s.",
            payload.get("component_id"),
            input_is_list_target_node_ids,
        )
        return await invoke_remote_engine_async(
            payload,
            kwargs_payload,
            allow_implicit_mapping=False,
        )

    parallelism, refined_prompt_warmup_target = boost_mapped_component_warmup(
        payload,
        total_items=total_items,
        reason="implicit_mapped_component_exact_parallelism",
    )
    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    if prompt_id is not None:
        await _await_prompt_warmup_slots(
            prompt_id,
            list(range(refined_prompt_warmup_target)),
            _prompt_warmup_head_start_seconds(),
        )
    logger.info(
        "Scheduling implicitly mapped Modal component=%s for %d item(s) with local parallelism=%d prompt_warmup_target=%d across inputs=%s.",
        payload.get("component_id"),
        total_items,
        parallelism,
        refined_prompt_warmup_target,
        sorted(split_inputs),
    )
    _emit_local_mapped_progress(payload, 0, total_items)

    broadcast_inputs = {
        input_name: value
        for input_name, value in hydrated_inputs.items()
        if input_name not in split_inputs
    }
    (
        static_execute_node_ids,
        mapped_execute_node_ids,
    ) = _partition_implicit_batched_execute_nodes(
        payload,
        split_inputs,
    )
    hybrid_payload = copy.deepcopy(payload)
    hybrid_payload["static_execute_node_ids"] = static_execute_node_ids
    hybrid_payload["mapped_execute_node_ids"] = mapped_execute_node_ids
    hybrid_payload["boundary_outputs"] = _annotate_implicit_batched_boundary_outputs(
        payload,
        mapped_execute_node_ids,
    )
    cleanup_payload = _build_remote_session_cleanup_payload(hybrid_payload)
    execution_mode = os.getenv("COMFY_MODAL_EXECUTION_MODE", "local")
    use_seeded_remote_lanes = (
        execution_mode != "local"
        and modal is not None
        and _payload_remote_session_handle(hybrid_payload) is not None
        and _mapped_phase_definition(hybrid_payload, "static_phase") is not None
    )

    static_outputs: tuple[Any, ...] = ()
    lane_remote_engines: list[Any] = []
    seeded_lane_indices: set[int] = set()
    skip_cleanup_after_interrupt = False
    try:
        if use_seeded_remote_lanes:
            lane_remote_engines = [
                _lookup_deployed_remote_engine(
                    hybrid_payload,
                    affinity_key_override=_mapped_lane_affinity_key(
                        hybrid_payload, lane_index
                    ),
                )
                for lane_index in range(parallelism)
            ]
            logger.info(
                "Seeding %d mapped worker lane(s) for component=%s and allowing ready lanes to start item dispatch independently.",
                parallelism,
                payload.get("component_id"),
            )

            async def seed_lane(lane_index: int) -> None:
                """Run the static bridge-producing phase once on one mapped worker lane."""
                lane_seed_payload = _build_phase_subgraph_payload(
                    hybrid_payload,
                    "static_phase",
                    f"{payload.get('component_id', 'modal-subgraph')}::seed:{lane_index}",
                    suppress_status_stream=True,
                )
                lane_seed_payload.pop("clear_remote_session", None)
                await _invoke_bound_remote_engine_async(
                    lane_remote_engines[lane_index],
                    lane_seed_payload,
                    serialize_node_inputs(broadcast_inputs),
                )
                seeded_lane_indices.add(lane_index)

        elif static_execute_node_ids:
            static_response = await invoke_remote_engine_async(
                _build_static_mapped_payload(hybrid_payload),
                serialize_node_inputs(broadcast_inputs),
            )
            static_outputs = deserialize_node_outputs(static_response)
        else:
            lane_remote_engines = []

        per_item_outputs: list[tuple[Any, ...] | None] = [None] * total_items
        completed_items = 0
        all_items_completed = asyncio.Event()
        stop_dispatch_requested = asyncio.Event()
        worker_failure: asyncio.Future[
            BaseException
        ] = asyncio.get_running_loop().create_future()
        item_queue: asyncio.Queue[int | None] = asyncio.Queue()
        for item_index in range(total_items):
            item_queue.put_nowait(item_index)
        for _ in range(parallelism):
            item_queue.put_nowait(None)

        def request_interrupt_stop() -> None:
            """Stop queued mapped item dispatch and suppress cleanup after interruption."""
            nonlocal skip_cleanup_after_interrupt
            skip_cleanup_after_interrupt = True
            stop_dispatch_requested.set()

        def raise_if_local_interrupted() -> None:
            """Raise the native ComfyUI interrupt after marking mapped dispatch as stopped."""
            if _local_processing_interrupted():
                request_interrupt_stop()
                _raise_local_interrupt()

        async def run_worker(lane_index: int) -> None:
            """Execute queued implicit mapped items through one stable local worker lane."""
            nonlocal completed_items
            try:
                raise_if_local_interrupted()
                if use_seeded_remote_lanes:
                    _emit_local_mapped_lane_progress_start(payload, lane_index)
                    await seed_lane(lane_index)
                while True:
                    if stop_dispatch_requested.is_set():
                        return
                    item_index = await item_queue.get()
                    if item_index is None:
                        return
                    if stop_dispatch_requested.is_set():
                        return
                    raise_if_local_interrupted()
                    try:
                        item_payload = _build_mapped_item_payload(
                            hybrid_payload, item_index, lane_index
                        )
                        item_inputs = dict(broadcast_inputs)
                        for input_name, items in split_inputs.items():
                            item_inputs[input_name] = items[item_index]
                        if use_seeded_remote_lanes:
                            item_response = await _invoke_bound_remote_engine_async(
                                lane_remote_engines[lane_index],
                                item_payload,
                                serialize_node_inputs(item_inputs),
                            )
                        else:
                            item_response = await invoke_remote_engine_async(
                                item_payload,
                                serialize_node_inputs(item_inputs),
                            )
                        raise_if_local_interrupted()
                        per_item_outputs[item_index] = deserialize_node_outputs(
                            item_response
                        )
                        completed_items += 1
                        _emit_local_mapped_progress(
                            payload, completed_items, total_items
                        )
                        if completed_items >= total_items:
                            all_items_completed.set()
                    finally:
                        _clear_local_mapped_lane_progress(
                            payload, lane_index, item_index
                        )
            except BaseException as exc:
                if all_items_completed.is_set():
                    logger.info(
                        "Ignoring late mapped worker failure after component=%s already completed all items lane=%d: %s",
                        payload.get("component_id"),
                        lane_index,
                        exc,
                    )
                    return
                stop_dispatch_requested.set()
                if _exception_indicates_interruption(exc):
                    request_interrupt_stop()
                if not worker_failure.done():
                    worker_failure.set_result(exc)
                raise

        tasks = [
            asyncio.create_task(run_worker(lane_index))
            for lane_index in range(parallelism)
        ]
        for lane_index, task in enumerate(tasks):
            task.add_done_callback(
                lambda completed_task, lane_index=lane_index: _log_detached_mapped_lane_result(
                    completed_task,
                    component_id=str(payload.get("component_id", "modal-subgraph")),
                    lane_index=lane_index,
                )
            )
        completion_task = asyncio.create_task(all_items_completed.wait())
        try:
            while not all_items_completed.is_set():
                wait_set: set[asyncio.Task[Any] | asyncio.Future[BaseException]] = {
                    completion_task,
                    worker_failure,
                    *tasks,
                }
                done, _ = await asyncio.wait(
                    wait_set, return_when=asyncio.FIRST_COMPLETED
                )
                if completion_task in done or all_items_completed.is_set():
                    break
                if worker_failure in done:
                    stop_dispatch_requested.set()
                    raise worker_failure.result()
                for task in done:
                    if task in tasks:
                        task_exc = task.exception()
                        if task_exc is not None:
                            stop_dispatch_requested.set()
                            raise task_exc
            if not all(item_outputs is not None for item_outputs in per_item_outputs):
                await asyncio.gather(*tasks)
        except BaseException as exc:
            stop_dispatch_requested.set()
            if _exception_indicates_interruption(exc):
                skip_cleanup_after_interrupt = True
            for task in tasks:
                task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            raise
        finally:
            completion_task.cancel()
            await asyncio.gather(completion_task, return_exceptions=True)

        return serialize_node_outputs(
            _merge_static_and_mapped_outputs(
                static_outputs=static_outputs,
                mapped_outputs=_aggregate_mapped_outputs(
                    [
                        item_outputs
                        for item_outputs in per_item_outputs
                        if item_outputs is not None
                    ],
                    {
                        **hybrid_payload,
                        "boundary_outputs": [
                            boundary_output
                            for boundary_output in hybrid_payload.get(
                                "boundary_outputs", []
                            )
                            if _is_mapped_boundary_output(
                                boundary_output, hybrid_payload
                            )
                        ],
                    },
                ),
                payload=hybrid_payload,
            )
        )
    finally:
        if cleanup_payload is not None and not skip_cleanup_after_interrupt:
            if use_seeded_remote_lanes and lane_remote_engines:
                await asyncio.gather(
                    *(
                        _invoke_bound_remote_engine_async(
                            lane_remote_engines[lane_index],
                            {
                                **cleanup_payload,
                                "component_id": (
                                    f"{payload.get('component_id', 'modal-subgraph')}::cleanup:{lane_index}"
                                ),
                            },
                            serialize_node_inputs({}),
                        )
                        for lane_index in sorted(seeded_lane_indices)
                    )
                )
            else:
                await invoke_remote_engine_async(
                    cleanup_payload,
                    serialize_node_inputs({}),
                )
        elif cleanup_payload is not None:
            logger.info(
                "Skipping remote-session cleanup for implicitly mapped Modal component=%s because execution was interrupted.",
                payload.get("component_id"),
            )


async def _invoke_mapped_remote_engine_async(
    payload: dict[str, Any], kwargs_payload: bytes
) -> bytes:
    """Execute one mapped payload locally using explicit static and mapped phases."""
    hydrated_inputs = deserialize_node_inputs(kwargs_payload)
    return await asyncio.to_thread(
        lambda: serialize_node_outputs(
            _execute_mapped_subgraph_payload(
                payload,
                hydrated_inputs,
            )
        )
    )


def _lookup_deployed_remote_engine(
    payload: dict[str, Any],
    *,
    affinity_key_override: str | None = None,
    protocol_probe: bool = False,
) -> Any:
    """Look up the deployed runtime, optionally omitting new parameters for probing."""
    if modal is None:
        raise ModalRemoteInvocationError("Modal SDK is unavailable.")

    settings = _settings_for_payload(payload)
    deployment_app_name = modal_deployment_app_name(settings)
    snapshot_profile_key = _prepare_snapshot_profile_fields(payload)
    gpu_snapshot_enabled = (
        False
        if protocol_probe
        else _select_gpu_snapshot_for_profile(payload, snapshot_profile_key)
    )
    worker_affinity_key = affinity_key_override or _remote_worker_affinity_key(payload)
    logger.info(
        "Attempting deployed Modal invocation for app=%s class=%s component=%s worker_affinity=%s snapshot_profile=%s gpu_snapshot_enabled=%s.",
        deployment_app_name,
        "RemoteEngine",
        payload.get("component_id"),
        worker_affinity_key,
        snapshot_profile_key or None,
        gpu_snapshot_enabled,
    )
    remote_cls = modal.Cls.from_name(deployment_app_name, "RemoteEngine")
    if bool(payload.get("remote_local_gap_pool")) and hasattr(remote_cls, "with_options"):
        local_gap_scaledown_seconds = max(
            int(settings.scaledown_window_seconds),
            int(settings.local_gap_keepalive_seconds),
        )
        remote_cls = remote_cls.with_options(
            scaledown_window=local_gap_scaledown_seconds,
        )
        logger.info(
            "Using local-gap Modal pool component=%s worker_affinity=%s scaledown_window=%ds.",
            payload.get("component_id"),
            worker_affinity_key,
            local_gap_scaledown_seconds,
        )
    remote_engine_kwargs: dict[str, Any] = {
        "gpu_snapshot_enabled": gpu_snapshot_enabled,
    }
    if snapshot_profile_key:
        remote_engine_kwargs["snapshot_profile_key"] = snapshot_profile_key
    if not protocol_probe:
        remote_engine_kwargs["worker_affinity_key"] = worker_affinity_key
    else:
        logger.info(
            "Probing deployed RemoteEngine protocol without affinity parameters component=%s.",
            payload.get("component_id"),
        )
    remote_engine = remote_cls(**remote_engine_kwargs)
    with _MODAL_AUTO_DEPLOY_LOCK:
        runtime_is_known_current = (
            _modal_runtime_cache_key(payload) in _MODAL_REMOTE_APP_VERSION_OK
        )
    if runtime_is_known_current:
        _ensure_llm_profiles_staged(payload, deployment_app_name)
    return remote_engine


def _emit_local_llm_staging_progress(
    payload: Mapping[str, Any],
    stage_event: Mapping[str, Any],
) -> None:
    """Render one CPU ModelStager update on the target component's node bar."""
    prompt_id = (
        str(payload["prompt_id"]) if payload.get("prompt_id") is not None else None
    )
    extra_data = payload.get("extra_data") or {}
    client_id = (
        str(extra_data["client_id"])
        if isinstance(extra_data, Mapping) and extra_data.get("client_id") is not None
        else None
    )
    component_id = str(payload.get("component_id") or "")
    if not component_id:
        return
    maximum = stage_event.get("max")
    _emit_local_modal_progress(
        prompt_id=prompt_id,
        client_id=client_id,
        node_id=component_id,
        value=float(stage_event.get("value") or 0.0),
        max_value=float(maximum) if maximum is not None else 1.0,
        stage=str(stage_event.get("stage") or "staging"),
        message=str(stage_event.get("message") or "Staging LLM snapshot"),
        unit=(
            str(stage_event["unit"])
            if stage_event.get("unit") is not None
            else None
        ),
        indeterminate=bool(stage_event.get("indeterminate", False)),
        pre_gpu=True,
    )


def _ensure_llm_profiles_staged(
    payload: dict[str, Any],
    deployment_app_name: str,
) -> None:
    """Stage every LLM profile in a payload on a CPU worker before GPU dispatch."""
    if modal is None:
        raise ModalRemoteInvocationError("Modal SDK is unavailable.")
    model_references = llm_model_references_from_payload(payload)
    if not model_references:
        return
    with _STAGED_LLM_PROFILES_LOCK:
        if not _STAGED_LLM_PROFILES:
            _STAGED_LLM_PROFILE_RESULTS.clear()
        missing_model_references = [
            reference
            for reference in model_references
            if (deployment_app_name, reference) not in _STAGED_LLM_PROFILE_RESULTS
        ]
        if missing_model_references:
            _emit_local_remote_startup_status(
                payload,
                phase="llm_staging",
                status_message=(
                    "Inspecting and staging LLM on CPU; no GPU is allocated yet"
                ),
            )
            logger.info(
                "Dispatching CPU model resolution/staging app=%s models=%s "
                "before GPU component=%s.",
                deployment_app_name,
                missing_model_references,
                payload.get("component_id"),
            )
            stager_cls = modal.Cls.from_name(deployment_app_name, "ModelStager")
            stager = stager_cls()
            stage_results: list[dict[str, Any]] = []
            stage_stream = getattr(stager, "stage_profiles_stream", None)
            remote_generator = getattr(stage_stream, "remote_gen", None)
            if callable(remote_generator):
                for stage_event in remote_generator(missing_model_references):
                    if not isinstance(stage_event, Mapping):
                        continue
                    if stage_event.get("kind") == "result":
                        candidate_results = stage_event.get("results")
                        if isinstance(candidate_results, list):
                            stage_results = candidate_results
                        continue
                    if stage_event.get("kind") != "progress":
                        continue
                    _emit_local_llm_staging_progress(payload, stage_event)
            else:
                stage_results = stager.stage_profiles.remote(
                    missing_model_references
                )
            confirmed_references: set[str] = set()
            for stage_result in stage_results:
                if not isinstance(stage_result, Mapping):
                    continue
                requested_reference = str(
                    stage_result.get("requested_reference")
                    or stage_result.get("profile_id")
                    or ""
                )
                profile_id = str(stage_result.get("profile_id") or "")
                revision = str(stage_result.get("revision") or "")
                if not revision and requested_reference == profile_id:
                    try:
                        revision = get_llm_profile(profile_id).revision
                    except ValueError:
                        pass
                if not requested_reference or not profile_id or not revision:
                    continue
                normalized_result = dict(stage_result)
                normalized_result["requested_reference"] = requested_reference
                normalized_result["profile_id"] = profile_id
                normalized_result["revision"] = revision
                _STAGED_LLM_PROFILE_RESULTS[
                    (deployment_app_name, requested_reference)
                ] = normalized_result
                _STAGED_LLM_PROFILE_RESULTS[
                    (deployment_app_name, profile_id)
                ] = normalized_result
                _STAGED_LLM_PROFILES.add((deployment_app_name, profile_id, revision))
                confirmed_references.add(requested_reference)
            missing_results = set(missing_model_references) - confirmed_references
            if missing_results:
                raise ModalRemoteInvocationError(
                    f"Modal ModelStager did not confirm models {sorted(missing_results)}."
                )
            downloaded_gib = sum(
                float(result.get("artifact_bytes") or 0) / 1024**3
                for result in stage_results
                if isinstance(result, Mapping) and result.get("downloaded")
            )
            _emit_local_remote_startup_status(
                payload,
                phase="llm_staged",
                status_message=(
                    f"LLM staging complete ({downloaded_gib:.1f} GiB downloaded); "
                    "starting GPU worker"
                ),
            )
        resolved_results = {
            reference: _STAGED_LLM_PROFILE_RESULTS[(deployment_app_name, reference)]
            for reference in model_references
        }
    profile_ids_by_reference = {
        reference: str(result["profile_id"])
        for reference, result in resolved_results.items()
    }
    rewrite_llm_model_references(payload, profile_ids_by_reference)
    revisions = ",".join(
        f"{result['profile_id']}:{result['revision']}"
        for result in sorted(
            resolved_results.values(),
            key=lambda value: str(value["profile_id"]),
        )
    )
    payload["requires_volume_reload"] = True
    payload["volume_reload_marker"] = hashlib.sha256(
        f"llm-profiles:{revisions}".encode("utf-8")
    ).hexdigest()
    logger.info(
        "Modal LLM models are resolved and staged for component=%s profiles=%s "
        "reload_marker=%s.",
        payload.get("component_id"),
        sorted(profile_ids_by_reference.values()),
        payload["volume_reload_marker"],
    )


def _rewrite_staged_llm_kwargs_payload(
    kwargs_payload: bytes,
    deployment_app_name: str,
) -> bytes:
    """Replace a direct node input model reference with its staged profile ID."""
    hydrated_inputs = deserialize_node_inputs(kwargs_payload)
    if not isinstance(hydrated_inputs, Mapping):
        return kwargs_payload
    model_reference = hydrated_inputs.get("model_profile")
    if not isinstance(model_reference, str) or not model_reference.strip():
        return kwargs_payload
    normalized_reference = model_reference.strip()
    with _STAGED_LLM_PROFILES_LOCK:
        stage_result = _STAGED_LLM_PROFILE_RESULTS.get(
            (deployment_app_name, normalized_reference)
        )
    if stage_result is None:
        return kwargs_payload
    profile_id = str(stage_result.get("profile_id") or "").strip()
    if not profile_id or profile_id == normalized_reference:
        return kwargs_payload
    rewritten_inputs = dict(hydrated_inputs)
    rewritten_inputs["model_profile"] = profile_id
    logger.info(
        "Rewrote direct Modal LLM input model=%s to generated profile=%s.",
        normalized_reference,
        profile_id,
    )
    return serialize_node_inputs(rewritten_inputs)


async def _invoke_bound_remote_engine_async(
    remote_engine: Any,
    payload: dict[str, Any],
    kwargs_payload: bytes,
) -> bytes:
    """Invoke one pre-bound remote engine handle asynchronously with local interrupt mirroring."""
    logger.info(
        "Dispatching async Modal remote invocation via bound engine for component=%s payload_kind=%s.",
        payload.get("component_id"),
        payload.get("payload_kind"),
    )
    cancellation_event = threading.Event()
    future = _REMOTE_MODAL_CALL_EXECUTOR.submit(
        _invoke_remote_engine_payload_with_recovery,
        remote_engine,
        dict(payload),
        kwargs_payload,
        cancellation_event,
    )
    wrapped_future = asyncio.wrap_future(future)
    interrupt_sent = False
    cancellation_started_at: float | None = None
    try:
        while True:
            try:
                response = await asyncio.wait_for(
                    asyncio.shield(wrapped_future), timeout=0.1
                )
                break
            except asyncio.TimeoutError:
                (
                    interrupt_sent,
                    cancellation_started_at,
                ) = await _handle_modal_wait_cancellation_async(
                    payload,
                    cancellation_event,
                    interrupt_sent=interrupt_sent,
                    cancellation_started_at=cancellation_started_at,
                )
                continue
    except asyncio.CancelledError:
        cancellation_event.set()
        prompt_id = (
            str(payload.get("prompt_id"))
            if payload.get("prompt_id") is not None
            else None
        )
        if prompt_id is not None:
            await request_remote_modal_prompt_interrupt_async(prompt_id)
        raise
    except Exception:
        if cancellation_event.is_set() or _local_processing_interrupted():
            logger.info(
                "Reraising async Modal failure as a local interrupt for component=%s after cancellation.",
                payload.get("component_id"),
            )
            _raise_local_interrupt()
        logger.exception(
            "Async Modal remote invocation via bound engine failed for component=%s.",
            payload.get("component_id"),
        )
        raise
    if cancellation_event.is_set() or _local_processing_interrupted():
        logger.info(
            "Async bound-engine invocation for component=%s finished after interruption; raising local interrupt.",
            payload.get("component_id"),
        )
        _raise_local_interrupt()
    logger.info(
        "Async Modal remote invocation via bound engine completed for component=%s.",
        payload.get("component_id"),
    )
    return response


def _modal_environment_name() -> str | None:
    """Return the active Modal environment name when explicitly configured."""
    environment_name = os.getenv("MODAL_ENVIRONMENT")
    if environment_name is None:
        return None
    normalized = environment_name.strip()
    return normalized or None


def _lookup_modal_session_bridge_store() -> Any:
    """Return the shared Modal Dict that contains durable bridge records."""
    if modal is None or not hasattr(modal, "Dict"):
        raise RuntimeError(
            "Modal SDK is required to materialize a remote bridge locally."
        )
    settings = get_settings()
    cache_key = (settings.session_bridge_dict_name, _modal_environment_name())
    with _MODAL_SESSION_BRIDGE_DICTS_LOCK:
        cached_store = _MODAL_SESSION_BRIDGE_DICTS.get(cache_key)
        if cached_store is not None:
            return cached_store
    bridge_store = modal.Dict.from_name(
        settings.session_bridge_dict_name,
        environment_name=cache_key[1],
        create_if_missing=True,
    )
    with _MODAL_SESSION_BRIDGE_DICTS_LOCK:
        _MODAL_SESSION_BRIDGE_DICTS[cache_key] = bridge_store
    return bridge_store


def _lookup_modal_durable_volume() -> Any:
    """Return the shared Modal Volume that contains durable bridge objects."""
    if modal is None or not hasattr(modal, "Volume"):
        raise RuntimeError(
            "Modal SDK is required to download a remote bridge object locally."
        )
    settings = get_settings()
    cache_key = (settings.volume_name, _modal_environment_name())
    with _MODAL_DURABLE_VOLUMES_LOCK:
        cached_volume = _MODAL_DURABLE_VOLUMES.get(cache_key)
        if cached_volume is not None:
            return cached_volume
    volume = modal.Volume.from_name(
        settings.volume_name,
        environment_name=cache_key[1],
        create_if_missing=True,
    )
    with _MODAL_DURABLE_VOLUMES_LOCK:
        _MODAL_DURABLE_VOLUMES[cache_key] = volume
    return volume


def _modal_deploy_cache_key(
    payload: Mapping[str, Any] | None = None,
) -> tuple[str, str | None]:
    """Return the cache key for auto-deployed Modal apps."""
    settings = _settings_for_payload(payload) if payload is not None else get_settings()
    return (modal_deployment_app_name(settings), _modal_environment_name())


@lru_cache(maxsize=8)
def _remote_runtime_identity_for_settings(
    settings: ModalSyncSettings,
) -> RemoteRuntimeIdentity:
    """Build and cache the local runtime identity for one resolved settings object."""
    return build_remote_runtime_identity(
        repo_root=Path(__file__).resolve().parents[1],
        comfyui_root=settings.comfyui_root,
        custom_nodes_dir=settings.custom_nodes_dir,
        settings=settings,
    )


def _expected_remote_runtime_fingerprint(
    payload: Mapping[str, Any] | None = None
) -> str:
    """Return the exact deployment fingerprint required by this local client."""
    settings = _settings_for_payload(payload) if payload is not None else get_settings()
    return _remote_runtime_identity_for_settings(settings).fingerprint


def _modal_runtime_cache_key(
    payload: Mapping[str, Any] | None = None,
) -> tuple[str, str | None, str]:
    """Return the version cache key for the current app, environment, and runtime."""
    settings = _settings_for_payload(payload) if payload is not None else get_settings()
    return (
        modal_deployment_app_name(settings),
        _modal_environment_name(),
        _expected_remote_runtime_fingerprint(payload),
    )


def _call_modal_method(method: Any, *args: Any, **kwargs: Any) -> Any:
    """Invoke a Modal method handle or an in-process test double."""
    remote_method = getattr(method, "remote", None)
    if callable(remote_method):
        return remote_method(*args, **kwargs)
    return method(*args, **kwargs)


def _remote_engine_runtime_version(remote_engine: Any) -> dict[str, Any] | None:
    """Return runtime version metadata from a deployed engine when available."""
    version_method = getattr(remote_engine, "runtime_version", None)
    if version_method is None:
        return None
    version_payload = _call_modal_method(version_method)
    if not isinstance(version_payload, dict):
        return None
    return version_payload


def _remote_runtime_version_from_cpu_stager(
    payload: Mapping[str, Any],
) -> dict[str, Any] | None:
    """Read deployment identity from the CPU stager when that method is available."""
    if modal is None:
        return None
    settings = _settings_for_payload(payload)
    deployment_app_name = modal_deployment_app_name(settings)
    stager_cls = modal.Cls.from_name(deployment_app_name, "ModelStager")
    stager = stager_cls()
    version_method = getattr(stager, "runtime_version", None)
    if version_method is None:
        return None
    try:
        version_payload = _call_modal_method(version_method)
    except AttributeError:
        return None
    except _modal_lookup_error_types() as exc:
        if _is_missing_modal_deployment_error(exc):
            raise
        logger.info(
            "CPU ModelStager runtime identity is unavailable for app=%s; "
            "falling back to the legacy GPU protocol probe: %s",
            deployment_app_name,
            exc,
        )
        return None
    if not isinstance(version_payload, dict):
        return None
    return version_payload


def _runtime_fingerprint_from_payload(
    version_payload: dict[str, Any] | None
) -> str | None:
    """Return a normalized runtime fingerprint from remote version metadata."""
    if version_payload is None:
        return None
    runtime_fingerprint = version_payload.get("runtime_fingerprint")
    if not isinstance(runtime_fingerprint, str):
        return None
    normalized = runtime_fingerprint.strip()
    return normalized or None


def _is_runtime_version_payload_current(
    version_payload: dict[str, Any] | None,
    payload: Mapping[str, Any] | None = None,
) -> bool:
    """Return whether remote version metadata exactly matches this local runtime."""
    if version_payload is None:
        return False
    protocol_version = version_payload.get("protocol_version")
    if isinstance(protocol_version, bool) or not isinstance(protocol_version, int):
        return False
    return (
        protocol_version == _REMOTE_APP_PROTOCOL_VERSION
        and _runtime_fingerprint_from_payload(version_payload)
        == _expected_remote_runtime_fingerprint(payload)
    )


def _is_remote_engine_runtime_current(
    remote_engine: Any,
    payload: Mapping[str, Any] | None = None,
) -> bool:
    """Return whether a deployed engine exactly matches this local runtime."""
    return _is_runtime_version_payload_current(
        _remote_engine_runtime_version(remote_engine),
        payload,
    )


def _stop_modal_app_via_sdk(app_name: str) -> bool:
    """Try to stop a Modal app through the SDK if this SDK version exposes app stopping."""
    if modal is None:
        return False
    try:
        experimental_namespace = importlib.import_module("modal.experimental")
    except ModuleNotFoundError as exc:
        if exc.name not in {"modal", "modal.experimental"}:
            raise
    else:
        stop_app = getattr(experimental_namespace, "stop_app", None)
        if callable(stop_app):
            try:
                _call_modal_method(
                    stop_app,
                    app_name,
                    environment_name=_modal_environment_name(),
                )
            except _modal_lookup_error_types():
                return True
            return True

    app_namespace = getattr(modal, "App", None)
    app_lookup = getattr(app_namespace, "lookup", None)
    if not callable(app_lookup):
        return False
    try:
        lookup_signature = inspect.signature(app_lookup)
    except (TypeError, ValueError):
        lookup_signature = None
    lookup_kwargs: dict[str, Any] = {}
    if lookup_signature is not None:
        if "create_if_missing" in lookup_signature.parameters:
            lookup_kwargs["create_if_missing"] = False
        elif "create" in lookup_signature.parameters:
            lookup_kwargs["create"] = False
    try:
        app_handle = app_lookup(app_name, **lookup_kwargs)
    except _modal_lookup_error_types():
        return True
    stop_method = getattr(app_handle, "stop", None)
    if not callable(stop_method):
        return False
    _call_modal_method(stop_method)
    return True


def _stop_modal_app_via_cli(app_name: str) -> bool:
    """Try to stop a Modal app non-interactively through the Modal CLI."""
    modal_cli = shutil.which("modal")
    if modal_cli is None:
        return False
    command = [modal_cli, "app", "stop", app_name, "--yes"]
    environment_name = _modal_environment_name()
    if environment_name is not None:
        command.extend(("--env", environment_name))
    try:
        completed = subprocess.run(
            command,
            check=False,
            capture_output=True,
            text=True,
            timeout=_MODAL_APP_STOP_TIMEOUT_SECONDS,
        )
    except subprocess.TimeoutExpired:
        logger.warning(
            "Modal CLI app stop timed out for app=%s environment=%s after %.1fs.",
            app_name,
            environment_name or "<default>",
            _MODAL_APP_STOP_TIMEOUT_SECONDS,
        )
        return False
    if completed.returncode != 0:
        logger.warning(
            "Modal CLI app stop failed for app=%s exit_code=%s stderr=%s",
            app_name,
            completed.returncode,
            completed.stderr.strip(),
        )
        return False
    return True


def _stop_modal_app_for_replacement(app_name: str) -> None:
    """Stop an out-of-date Modal app before replacing it with a fresh deployment."""
    if _stop_modal_app_via_sdk(app_name):
        logger.warning(
            "Stopped out-of-date Modal app %s through the Modal SDK.", app_name
        )
        return
    if _stop_modal_app_via_cli(app_name):
        logger.warning(
            "Stopped out-of-date Modal app %s through the Modal CLI.", app_name
        )
        return
    raise ModalRemoteInvocationError(
        f"Deployed Modal app {app_name!r} is out of date, but Modal-Sync could not stop it automatically. "
        f"Stop it manually with `modal app stop {app_name}` and retry."
    )


def _mark_modal_deploy_state_not_ready(deploy_key: tuple[str, str | None]) -> None:
    """Invalidate the local in-process deploy readiness cache for one app."""
    with _MODAL_AUTO_DEPLOY_LOCK:
        deploy_state = _MODAL_AUTO_DEPLOY_STATES.get(deploy_key)
    if deploy_state is None:
        return
    with deploy_state.condition:
        deploy_state.ready = False
        deploy_state.last_error = None
        deploy_state.condition.notify_all()


def _replace_outdated_modal_app(
    payload: dict[str, Any],
    remote_engine: Any,
    *,
    version_payload: dict[str, Any] | None = None,
) -> Any:
    """Stop and auto-deploy a replacement for an incompatible deployed app."""
    settings = _settings_for_payload(payload)
    deployment_app_name = modal_deployment_app_name(settings)
    deploy_key = _modal_deploy_cache_key(payload)
    runtime_cache_key = _modal_runtime_cache_key(payload)
    if version_payload is None:
        version_payload = _remote_engine_runtime_version(remote_engine)
    protocol_version = (
        version_payload.get("protocol_version")
        if isinstance(version_payload, dict)
        else None
    )
    remote_fingerprint = _runtime_fingerprint_from_payload(version_payload)
    local_fingerprint = _expected_remote_runtime_fingerprint(payload)
    logger.warning(
        "Deployed Modal app %s is out of date for component=%s remote_protocol=%s local_protocol=%s remote_fingerprint=%s local_fingerprint=%s; stopping and replacing it.",
        deployment_app_name,
        payload.get("component_id"),
        protocol_version,
        _REMOTE_APP_PROTOCOL_VERSION,
        remote_fingerprint,
        local_fingerprint,
    )
    with _MODAL_AUTO_DEPLOY_LOCK:
        _MODAL_REMOTE_APP_VERSION_OK.discard(runtime_cache_key)
    _mark_modal_deploy_state_not_ready(deploy_key)
    _stop_modal_app_for_replacement(deployment_app_name)
    stale_error = ModalRemoteAppOutOfDateError(
        f"Modal app {deployment_app_name!r} runtime identity does not match the local client "
        f"(remote protocol={protocol_version!r}, local protocol={_REMOTE_APP_PROTOCOL_VERSION}, "
        f"remote fingerprint={remote_fingerprint!r}, local fingerprint={local_fingerprint!r})."
    )
    replacement_engine = _auto_deploy_modal_app(payload, stale_error)
    with _MODAL_AUTO_DEPLOY_LOCK:
        _MODAL_REMOTE_APP_VERSION_OK.add(runtime_cache_key)
    return replacement_engine


def _ensure_remote_engine_protocol_current(
    remote_engine: Any, payload: dict[str, Any]
) -> Any:
    """Return a compatible remote engine, replacing the deployed app when allowed."""
    settings = _settings_for_payload(payload)
    runtime_cache_key = _modal_runtime_cache_key(payload)
    with _MODAL_AUTO_DEPLOY_LOCK:
        runtime_is_known_current = runtime_cache_key in _MODAL_REMOTE_APP_VERSION_OK
    if runtime_is_known_current:
        logger.info(
            "Rebinding deployed RemoteEngine after cached protocol validation "
            "component=%s worker_affinity=%s.",
            payload.get("component_id"),
            _remote_worker_affinity_key(payload),
        )
        return _lookup_deployed_remote_engine(payload)
    version_payload = _remote_engine_runtime_version(remote_engine)
    if _is_runtime_version_payload_current(version_payload, payload):
        with _MODAL_AUTO_DEPLOY_LOCK:
            _MODAL_REMOTE_APP_VERSION_OK.add(runtime_cache_key)
        return _lookup_deployed_remote_engine(payload)
    if not settings.auto_deploy:
        raise ModalRemoteInvocationError(
            "Deployed Modal app runtime fingerprint is out of date and "
            "COMFY_MODAL_AUTO_DEPLOY=false prevents automatic replacement."
        )
    return _replace_outdated_modal_app(
        payload,
        remote_engine,
        version_payload=version_payload,
    )


def _lookup_protocol_current_remote_engine(payload: dict[str, Any]) -> Any:
    """Return an affinity-bound engine without allocating a cached protocol probe."""
    runtime_cache_key = _modal_runtime_cache_key(payload)
    with _MODAL_AUTO_DEPLOY_LOCK:
        runtime_is_known_current = runtime_cache_key in _MODAL_REMOTE_APP_VERSION_OK
    if runtime_is_known_current:
        logger.info(
            "Using cached Modal protocol validation without creating a parameterless "
            "probe component=%s worker_affinity=%s.",
            payload.get("component_id"),
            _remote_worker_affinity_key(payload),
        )
        return _lookup_deployed_remote_engine(payload)

    version_payload = _remote_runtime_version_from_cpu_stager(payload)
    if version_payload is not None:
        if _is_runtime_version_payload_current(version_payload, payload):
            with _MODAL_AUTO_DEPLOY_LOCK:
                _MODAL_REMOTE_APP_VERSION_OK.add(runtime_cache_key)
            logger.info(
                "Validated Modal runtime through CPU ModelStager without allocating "
                "a GPU protocol probe component=%s worker_affinity=%s.",
                payload.get("component_id"),
                _remote_worker_affinity_key(payload),
            )
            return _lookup_deployed_remote_engine(payload)
        settings = _settings_for_payload(payload)
        if not settings.auto_deploy:
            raise ModalRemoteInvocationError(
                "Deployed Modal app runtime fingerprint is out of date and "
                "COMFY_MODAL_AUTO_DEPLOY=false prevents automatic replacement."
            )
        return _replace_outdated_modal_app(
            payload,
            None,
            version_payload=version_payload,
        )

    logger.info(
        "CPU Modal runtime validation method is unavailable for component=%s; "
        "using the legacy parameterless GPU protocol probe.",
        payload.get("component_id"),
    )
    return _ensure_remote_engine_protocol_current(
        _lookup_deployed_remote_engine(payload, protocol_probe=True),
        payload,
    )


def _modal_auto_deploy_state(
    deploy_key: tuple[str, str | None],
) -> _ModalAutoDeployState:
    """Return the shared auto-deploy state bucket for one Modal app/environment."""
    with _MODAL_AUTO_DEPLOY_LOCK:
        state = _MODAL_AUTO_DEPLOY_STATES.get(deploy_key)
        if state is None:
            state = _ModalAutoDeployState()
            _MODAL_AUTO_DEPLOY_STATES[deploy_key] = state
        return state


def _lookup_deployed_remote_engine_with_retry(
    payload: dict[str, Any],
    *,
    timeout_seconds: float = 15.0,
    initial_delay_seconds: float = 0.25,
    max_delay_seconds: float = 2.0,
) -> Any:
    """Poll deployed Modal lookup until the freshly deployed app becomes discoverable."""
    lookup_error_types = _modal_lookup_error_types()
    if not lookup_error_types:
        return _lookup_deployed_remote_engine(payload)

    deadline = time.monotonic() + max(0.0, timeout_seconds)
    delay_seconds = max(0.0, initial_delay_seconds)
    last_error: BaseException | None = None
    while True:
        try:
            return _lookup_deployed_remote_engine(payload)
        except lookup_error_types as exc:
            last_error = exc
            if not _is_missing_modal_deployment_error(exc):
                raise
            remaining_seconds = deadline - time.monotonic()
            if remaining_seconds <= 0.0:
                raise
            time.sleep(min(delay_seconds, remaining_seconds))
            delay_seconds = min(
                max_delay_seconds, max(delay_seconds * 2.0, initial_delay_seconds)
            )
    if last_error is not None:
        raise last_error
    raise ModalRemoteInvocationError(
        "Deployed Modal lookup retry loop exited unexpectedly."
    )


def _warmup_prompt_id(warmup_request: dict[str, Any]) -> str | None:
    """Return the prompt id that scopes proactive warmup state."""
    prompt_id = warmup_request.get("prompt_id")
    if prompt_id is None:
        return None
    normalized_prompt_id = str(prompt_id).strip()
    return normalized_prompt_id or None


def _ensure_prompt_warmup_state(prompt_id: str) -> _PromptWarmupState:
    """Return the cached warmup state bucket for one prompt."""
    global _PROMPT_WARMUP_STATE_ORDER

    state = _PROMPT_WARMUP_STATES.get(prompt_id)
    if state is not None:
        return state

    state = _PromptWarmupState()
    _PROMPT_WARMUP_STATES[prompt_id] = state
    if _PROMPT_WARMUP_STATE_ORDER is None:
        _PROMPT_WARMUP_STATE_ORDER = queue.SimpleQueue()
    _PROMPT_WARMUP_STATE_ORDER.put(prompt_id)
    while len(_PROMPT_WARMUP_STATES) > _PROMPT_WARMUP_STATE_CACHE_LIMIT:
        expired_prompt_id = _PROMPT_WARMUP_STATE_ORDER.get()
        _PROMPT_WARMUP_STATES.pop(expired_prompt_id, None)
    return state


def _clamp_prompt_warmup_target(warmup_target: int) -> int:
    """Clamp one proactive warmup target to the configured Modal container cap."""
    normalized_target = max(0, int(warmup_target))
    max_containers = get_settings().max_containers
    if max_containers is not None:
        return min(normalized_target, max_containers)
    return normalized_target


def _iter_loader_prewarm_prompt_payloads(
    payload: dict[str, Any]
) -> Iterator[dict[str, Any]]:
    """Yield subgraph-like payload fragments that may contain root loader nodes."""
    split_proxy_payloads = payload.get("split_proxy_payloads")
    if isinstance(split_proxy_payloads, dict):
        for phase_payload in split_proxy_payloads.values():
            if isinstance(phase_payload, dict):
                yield phase_payload
        return
    if isinstance(split_proxy_payloads, list):
        for phase_payload in split_proxy_payloads:
            if isinstance(phase_payload, dict):
                yield phase_payload
        return
    if isinstance(payload.get("subgraph_prompt"), dict):
        yield payload


def _is_root_literal_loader_node(prompt_node: Mapping[str, Any]) -> bool:
    """Return whether one prompt node is a supported loader with only literal inputs."""
    class_type = str(prompt_node.get("class_type") or "")
    if class_type not in _ROOT_LOADER_PREWARM_CLASS_TYPES:
        return False
    inputs = prompt_node.get("inputs")
    if not isinstance(inputs, Mapping):
        return False
    return not any(_is_link(input_value) for input_value in inputs.values())


def _loader_prewarm_plan_signature(class_type: str, inputs: Mapping[str, Any]) -> str:
    """Return a stable signature for one synthetic loader prewarm plan."""
    return json.dumps(
        {
            "class_type": class_type,
            "inputs": copy.deepcopy(dict(inputs)),
        },
        sort_keys=True,
        default=str,
    )


def _loader_snapshot_profile_key(loader_prewarm_plans: list[dict[str, Any]]) -> str:
    """Return the stable snapshot-profile key for one set of loader prewarm plans."""
    if not loader_prewarm_plans:
        return ""

    profile_payload = {
        "plan_signatures": sorted(
            str(plan.get("signature") or "")
            for plan in loader_prewarm_plans
            if str(plan.get("signature") or "")
        )
    }
    if not profile_payload["plan_signatures"]:
        return ""
    profile_digest = hashlib.sha256(
        json.dumps(profile_payload, sort_keys=True).encode("utf-8")
    ).hexdigest()
    return f"loader-profile:{profile_digest}"


def _normalize_loader_snapshot_profile_record(
    loader_prewarm_plans: list[dict[str, Any]],
) -> dict[str, Any] | None:
    """Return the normalized snapshot-profile record for one loader-plan set."""
    snapshot_profile_key = _loader_snapshot_profile_key(loader_prewarm_plans)
    if not snapshot_profile_key:
        return None
    return {
        "snapshot_profile_key": snapshot_profile_key,
        "loader_prewarm_plans": copy.deepcopy(loader_prewarm_plans),
        "snapshot_policy": {
            "selected_variant": None,
            "measurements": {"snapshot": [], "direct": []},
        },
    }


def _store_loader_snapshot_profile(loader_prewarm_plans: list[dict[str, Any]]) -> str:
    """Persist one loader snapshot profile so `snap=True` can retrieve it later."""
    settings = get_settings()
    if not settings.enable_gpu_memory_snapshot or modal is None:
        return ""

    normalized_record = _normalize_loader_snapshot_profile_record(loader_prewarm_plans)
    if normalized_record is None:
        return ""

    snapshot_profile_key = str(normalized_record["snapshot_profile_key"])
    with _SNAPSHOT_PROFILE_RECORDS_LOCK:
        cached_record = _SNAPSHOT_PROFILE_RECORDS.get(snapshot_profile_key)
        if cached_record is not None:
            return snapshot_profile_key
        snapshot_profiles = modal.Dict.from_name(
            settings.snapshot_profile_dict_name,
            create_if_missing=True,
        )
        stored_record = snapshot_profiles.get(snapshot_profile_key)
        if isinstance(stored_record, dict):
            merged_record = copy.deepcopy(stored_record)
            merged_record["snapshot_profile_key"] = snapshot_profile_key
            merged_record["loader_prewarm_plans"] = copy.deepcopy(
                loader_prewarm_plans
            )
        else:
            merged_record = normalized_record
        snapshot_profiles[snapshot_profile_key] = merged_record
        _SNAPSHOT_PROFILE_RECORDS[snapshot_profile_key] = merged_record
    return snapshot_profile_key


def _prepare_snapshot_profile_fields(payload: dict[str, Any]) -> str:
    """Derive, persist, and attach the loader profile used by one request."""
    loader_prewarm_plans = _build_loader_prewarm_plans(payload)
    stored_profile_key = _store_loader_snapshot_profile(loader_prewarm_plans)
    existing_profile_key = str(payload.get("snapshot_profile_key") or "").strip()
    snapshot_profile_key = stored_profile_key or existing_profile_key
    if snapshot_profile_key:
        payload["snapshot_profile_key"] = snapshot_profile_key
    return snapshot_profile_key


def _snapshot_profile_store_for_payload(payload: Mapping[str, Any]) -> Any | None:
    """Return the shared profile store when Modal and snapshot policy are enabled."""
    if modal is None:
        return None
    settings = _settings_for_payload(payload)
    return modal.Dict.from_name(
        settings.snapshot_profile_dict_name,
        create_if_missing=True,
    )


def _snapshot_policy_record(
    payload: Mapping[str, Any], snapshot_profile_key: str
) -> dict[str, Any] | None:
    """Load one profile record without discarding locally cached measurements."""
    with _SNAPSHOT_PROFILE_RECORDS_LOCK:
        cached = _SNAPSHOT_PROFILE_RECORDS.get(snapshot_profile_key)
    store = _snapshot_profile_store_for_payload(payload)
    if store is None:
        return copy.deepcopy(cached) if cached is not None else None
    stored = store.get(snapshot_profile_key)
    if not isinstance(stored, dict):
        return copy.deepcopy(cached) if cached is not None else None
    with _SNAPSHOT_PROFILE_RECORDS_LOCK:
        _SNAPSHOT_PROFILE_RECORDS[snapshot_profile_key] = copy.deepcopy(stored)
    return copy.deepcopy(stored)


def _select_gpu_snapshot_for_profile(
    payload: dict[str, Any], snapshot_profile_key: str
) -> bool:
    """Choose snapshot or direct startup from persisted per-profile measurements."""
    settings = _settings_for_payload(payload)
    if not settings.enable_gpu_memory_snapshot or not snapshot_profile_key:
        payload["gpu_snapshot_variant"] = "direct"
        return False
    requested_variant = str(payload.get("gpu_snapshot_variant") or "").strip()
    if requested_variant in {"snapshot", "direct"}:
        return requested_variant == "snapshot"

    record = _snapshot_policy_record(payload, snapshot_profile_key) or {}
    policy = record.get("snapshot_policy")
    if not isinstance(policy, Mapping):
        policy = {}
    selected_variant = policy.get("selected_variant")
    if selected_variant not in {"snapshot", "direct"}:
        measurements = policy.get("measurements")
        if not isinstance(measurements, Mapping):
            measurements = {}
        snapshot_samples = measurements.get("snapshot")
        direct_samples = measurements.get("direct")
        snapshot_count = len(snapshot_samples) if isinstance(snapshot_samples, list) else 0
        direct_count = len(direct_samples) if isinstance(direct_samples, list) else 0
        selected_variant = "snapshot" if snapshot_count <= direct_count else "direct"
    payload["gpu_snapshot_variant"] = selected_variant
    logger.info(
        "Selected Modal startup variant=%s snapshot_profile=%s component=%s.",
        selected_variant,
        snapshot_profile_key,
        payload.get("component_id"),
    )
    return selected_variant == "snapshot"


def _record_snapshot_warmup_measurement(
    payload: Mapping[str, Any], elapsed_seconds: float
) -> None:
    """Persist one warmup latency and select the faster profile after both variants run."""
    snapshot_profile_key = str(payload.get("snapshot_profile_key") or "").strip()
    variant = str(payload.get("gpu_snapshot_variant") or "").strip()
    if not snapshot_profile_key or variant not in {"snapshot", "direct"}:
        return
    store = _snapshot_profile_store_for_payload(payload)
    if store is None:
        return
    record = _snapshot_policy_record(payload, snapshot_profile_key) or {
        "snapshot_profile_key": snapshot_profile_key,
        "loader_prewarm_plans": [],
    }
    policy = record.get("snapshot_policy")
    if not isinstance(policy, dict):
        policy = {}
    measurements = policy.get("measurements")
    if not isinstance(measurements, dict):
        measurements = {"snapshot": [], "direct": []}
    for candidate in ("snapshot", "direct"):
        samples = measurements.get(candidate)
        if not isinstance(samples, list):
            measurements[candidate] = []
    measurements[variant].append(float(elapsed_seconds))
    measurements[variant] = measurements[variant][-5:]
    if measurements["snapshot"] and measurements["direct"]:
        policy["selected_variant"] = min(
            ("snapshot", "direct"),
            key=lambda candidate: statistics.median(measurements[candidate]),
        )
    else:
        policy["selected_variant"] = None
    policy["measurements"] = measurements
    record["snapshot_policy"] = policy
    store[snapshot_profile_key] = record
    with _SNAPSHOT_PROFILE_RECORDS_LOCK:
        _SNAPSHOT_PROFILE_RECORDS[snapshot_profile_key] = copy.deepcopy(record)
    logger.info(
        "Recorded Modal startup measurement snapshot_profile=%s variant=%s elapsed=%.3fs selected=%s.",
        snapshot_profile_key,
        variant,
        elapsed_seconds,
        policy["selected_variant"],
    )


def _build_loader_prewarm_plans(payload: dict[str, Any]) -> list[dict[str, Any]]:
    """Return synthetic one-node subgraph plans for supported root loader nodes."""
    if not get_settings().enable_loader_prewarm:
        return []

    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    seen_signatures: set[str] = set()
    plans: list[dict[str, Any]] = []
    for prompt_payload in _iter_loader_prewarm_prompt_payloads(payload):
        subgraph_prompt = prompt_payload.get("subgraph_prompt")
        if not isinstance(subgraph_prompt, Mapping):
            continue
        for node_id, prompt_node in subgraph_prompt.items():
            if not isinstance(prompt_node, Mapping) or not _is_root_literal_loader_node(
                prompt_node
            ):
                continue
            class_type = str(prompt_node.get("class_type") or "")
            inputs = prompt_node.get("inputs")
            if not isinstance(inputs, Mapping):
                continue
            signature = _loader_prewarm_plan_signature(class_type, inputs)
            if signature in seen_signatures:
                continue
            seen_signatures.add(signature)
            plans.append(
                {
                    "signature": signature,
                    "node_id": str(node_id),
                    "class_type": class_type,
                    "prompt_id": prompt_id,
                    "subgraph_prompt": {
                        str(node_id): copy.deepcopy(dict(prompt_node)),
                    },
                    "execute_node_ids": [str(node_id)],
                }
            )
    return plans


def _iter_executable_subgraph_nodes(
    payload: Mapping[str, Any],
) -> Iterator[Mapping[str, Any]]:
    """Yield only prompt nodes required by this payload's execution targets."""
    for prompt_payload in _iter_loader_prewarm_prompt_payloads(dict(payload)):
        raw_prompt = prompt_payload.get("subgraph_prompt")
        if not isinstance(raw_prompt, Mapping):
            continue
        prompt = {
            str(node_id): dict(prompt_node)
            for node_id, prompt_node in raw_prompt.items()
            if isinstance(prompt_node, Mapping)
        }
        execute_node_ids: list[str] = []
        for field_name in (
            "execute_node_ids",
            "mapped_execute_node_ids",
            "static_execute_node_ids",
        ):
            field_value = prompt_payload.get(field_name)
            if isinstance(field_value, (list, tuple)):
                execute_node_ids.extend(str(node_id) for node_id in field_value)
        required_node_ids = _resolve_required_subgraph_nodes(
            prompt,
            list(dict.fromkeys(execute_node_ids)),
        )
        for node_id in required_node_ids:
            yield prompt[node_id]


def _build_llm_prewarm_plans(payload: Mapping[str, Any]) -> list[dict[str, Any]]:
    """Return LLM warmups found in executable subgraph dependency closures."""
    plans: list[dict[str, Any]] = []
    seen_profiles: set[str] = set()
    for prompt_node in _iter_executable_subgraph_nodes(payload):
        if prompt_node.get("class_type") != "ModalLLM":
            continue
        inputs = prompt_node.get("inputs")
        if not isinstance(inputs, Mapping):
            continue
        model_profile = inputs.get("model_profile")
        if not isinstance(model_profile, str) or not model_profile.strip():
            continue
        normalized_profile = model_profile.strip()
        if normalized_profile in seen_profiles:
            continue
        seen_profiles.add(normalized_profile)
        signature_payload = {
            "model_profile": normalized_profile,
            "representative_request_count": 3,
        }
        plans.append(
            {
                **signature_payload,
                "signature": hashlib.sha256(
                    json.dumps(signature_payload, sort_keys=True).encode("utf-8")
                ).hexdigest(),
                "prompt_node": copy.deepcopy(dict(prompt_node)),
            }
        )
    return plans


def _build_prompt_warmup_request(payload: dict[str, Any]) -> dict[str, Any] | None:
    """Extract the prompt-scoped warmup-relevant fields from one payload."""
    settings = _settings_for_payload(payload)
    loader_prewarm_plans = _build_loader_prewarm_plans(payload)
    llm_prewarm_plans = _build_llm_prewarm_plans(payload)
    snapshot_profile_key = _store_loader_snapshot_profile(loader_prewarm_plans)
    if settings.enable_gpu_memory_snapshot and not snapshot_profile_key and not llm_prewarm_plans:
        logger.info(
            "Skipping proactive Modal warmup request for component=%s because GPU snapshots are enabled and no loader snapshot profile was derived.",
            payload.get("component_id"),
        )
        return None
    return {
        "prompt_id": (
            str(payload.get("prompt_id"))
            if payload.get("prompt_id") is not None
            else None
        ),
        "component_id": str(
            payload.get(
                "mapped_progress_display_node_id",
                payload.get("component_id", "modal-warmup"),
            )
        ),
        "modal_gpu": settings.modal_gpu,
        "remote_worker_affinity_group": payload.get(
            "remote_worker_affinity_group"
        ),
        "remote_local_gap_pool": bool(payload.get("remote_local_gap_pool")),
        "requires_volume_reload": bool(payload.get("requires_volume_reload", True)),
        "volume_reload_marker": payload.get("volume_reload_marker"),
        "uploaded_volume_paths": list(payload.get("uploaded_volume_paths", [])),
        "custom_nodes_bundle": payload.get("custom_nodes_bundle"),
        "loader_prewarm_plans": loader_prewarm_plans,
        "llm_prewarm_plans": llm_prewarm_plans,
        "snapshot_profile_key": snapshot_profile_key,
    }


def _component_parallelism_metadata(payload: dict[str, Any]) -> dict[str, Any]:
    """Return the prompt-level parallelism metadata attached by the queue rewrite."""
    extra_data = payload.get("extra_data") or {}
    modal_metadata = extra_data.get("modal") or {}
    if not isinstance(modal_metadata, dict):
        return {}
    return modal_metadata


def _prompt_parallelism_target(
    payload: dict[str, Any],
    *,
    exact_component_parallelism: dict[str, int] | None = None,
) -> int:
    """Return the current best-effort prompt-wide warmup target for one payload."""
    metadata = _component_parallelism_metadata(payload)
    execution_stages = metadata.get("component_execution_stages")
    if not isinstance(execution_stages, list):
        return 1

    mapped_component_ids = {
        str(component_id)
        for component_id in metadata.get("mapped_component_ids", [])
        if str(component_id)
    }
    exact_parallelism = exact_component_parallelism or {}
    stage_parallelism = 0
    for stage in execution_stages:
        if not isinstance(stage, list):
            continue
        current_stage_parallelism = 0
        for component_id_value in stage:
            component_id = str(component_id_value)
            if component_id in exact_parallelism:
                current_stage_parallelism += max(
                    1, int(exact_parallelism[component_id])
                )
            elif component_id in mapped_component_ids:
                current_stage_parallelism += 1
            else:
                current_stage_parallelism += 1
        stage_parallelism = max(stage_parallelism, current_stage_parallelism)

    fallback_parallelism = metadata.get("estimated_max_parallel_requests")
    if stage_parallelism <= 0 and isinstance(fallback_parallelism, int):
        stage_parallelism = max(stage_parallelism, fallback_parallelism)
    return _clamp_prompt_warmup_target(max(1, stage_parallelism))


def _register_exact_component_parallelism(
    payload: dict[str, Any], component_parallelism: int
) -> int:
    """Record exact mapped-component parallelism and return the refined prompt-wide warmup target."""
    prompt_id = (
        str(payload.get("prompt_id")) if payload.get("prompt_id") is not None else None
    )
    component_id = (
        str(payload.get("component_id"))
        if payload.get("component_id") is not None
        else None
    )
    if not prompt_id or not component_id:
        return _clamp_prompt_warmup_target(component_parallelism)

    with _PROMPT_WARMUP_STATES_LOCK:
        warmup_state = _ensure_prompt_warmup_state(prompt_id)
        warmup_state.exact_component_parallelism[component_id] = max(
            1, int(component_parallelism)
        )
        exact_parallelism = dict(warmup_state.exact_component_parallelism)
    return _prompt_parallelism_target(
        payload, exact_component_parallelism=exact_parallelism
    )


def _warmup_slot_payload(
    warmup_request: dict[str, Any], slot_index: int
) -> dict[str, Any]:
    """Return the remote warmup payload for one desired container slot."""
    slot_payload = copy.deepcopy(warmup_request)
    component_id = str(
        slot_payload.get("component_id")
        or slot_payload.get("prompt_id")
        or "modal-warmup"
    )
    slot_payload["component_id"] = f"{component_id}::warmup:{slot_index}"
    slot_payload["warmup_slot_index"] = int(slot_index)
    slot_payload["warmup_only"] = True
    return slot_payload


def _prompt_warmup_head_start_seconds() -> float:
    """Return the bounded delay the local scheduler may wait for exact warmup slots to start."""
    return max(0.0, float(get_settings().proactive_warmup_head_start_seconds))


def _invoke_remote_engine_warmup(
    remote_engine: Any, warmup_request: dict[str, Any]
) -> Any:
    """Ask one prepared remote engine instance to warm a container for a prompt."""
    warmup_method = getattr(remote_engine, "warmup_for_request", None)
    if warmup_method is None:
        logger.warning(
            "Remote warmup method is unavailable for prompt=%s component=%s; skipping proactive warmup.",
            warmup_request.get("prompt_id"),
            warmup_request.get("component_id"),
        )
        return None
    started_at = time.perf_counter()
    if hasattr(warmup_method, "remote"):
        result = warmup_method.remote(warmup_request)
    else:
        result = warmup_method(warmup_request)
    _record_snapshot_warmup_measurement(
        warmup_request,
        time.perf_counter() - started_at,
    )
    return result


def _invoke_remote_engine_warmup_with_recovery(
    remote_engine: Any,
    warmup_request: dict[str, Any],
) -> Any:
    """Retry one warmup call after auto-deploy when a stale deployed handle vanishes."""
    lookup_error_types = _modal_lookup_error_types()
    settings = _settings_for_payload(warmup_request)
    try:
        return _invoke_remote_engine_warmup(remote_engine, warmup_request)
    except lookup_error_types as exc:
        if not settings.auto_deploy or not _is_missing_modal_deployment_error(exc):
            raise
        logger.warning(
            "Modal warmup invocation failed for component=%s because the deployed app was missing at call time: %s. Recreating the app and retrying.",
            warmup_request.get("component_id"),
            exc,
        )
        recovered_remote_engine = _auto_deploy_modal_app(warmup_request, exc)
        _ensure_llm_profiles_staged(
            warmup_request,
            modal_deployment_app_name(settings),
        )
        return _invoke_remote_engine_warmup(recovered_remote_engine, warmup_request)


def _invoke_modal_warmup_blocking(warmup_request: dict[str, Any]) -> Any:
    """Warm one Modal container slot using deployed or ephemeral app state."""
    if modal is None:
        return None

    lookup_error_types = _modal_lookup_error_types()
    settings = _settings_for_payload(warmup_request)
    deployment_app_name = modal_deployment_app_name(settings)
    if lookup_error_types:
        try:
            remote_engine = _lookup_protocol_current_remote_engine(warmup_request)
            _ensure_llm_profiles_staged(warmup_request, deployment_app_name)
            return _invoke_remote_engine_warmup_with_recovery(
                remote_engine, warmup_request
            )
        except lookup_error_types as exc:
            if settings.auto_deploy:
                remote_engine = _auto_deploy_modal_app(warmup_request, exc)
                try:
                    _ensure_llm_profiles_staged(warmup_request, deployment_app_name)
                    return _invoke_remote_engine_warmup_with_recovery(
                        remote_engine,
                        warmup_request,
                    )
                except lookup_error_types as retry_exc:
                    exc = retry_exc
            if not settings.allow_ephemeral_fallback:
                raise ModalRemoteInvocationError(
                    "Proactive Modal warmup requires a deployed Modal app or a successful first-run auto-deploy. "
                    f"Lookup failed for app={deployment_app_name!r}: {exc}."
                ) from exc
    else:
        remote_engine = _lookup_protocol_current_remote_engine(warmup_request)
        _ensure_llm_profiles_staged(warmup_request, deployment_app_name)
        return _invoke_remote_engine_warmup_with_recovery(remote_engine, warmup_request)

    with _modal_cloud_settings_override(settings):
        cloud_module = _load_modal_cloud_module()
    cloud_app = getattr(cloud_module, "app", None)
    cloud_remote_engine = getattr(cloud_module, "RemoteEngine", None)
    if cloud_app is None or cloud_remote_engine is None:
        raise ModalRemoteInvocationError(
            "Stable Modal cloud entry module did not expose app and RemoteEngine."
        )
    run_context = cloud_app.run() if hasattr(cloud_app, "run") else nullcontext()
    with run_context:
        remote_engine = cloud_remote_engine()
        return _invoke_remote_engine_warmup_with_recovery(remote_engine, warmup_request)


def _run_prompt_warmup_slot(
    prompt_id: str,
    slot_index: int,
    warmup_request: dict[str, Any],
    reason: str,
) -> None:
    """Execute one proactive warmup slot and release it for retry on failure."""
    try:
        logger.info(
            "Starting proactive Modal warmup for prompt=%s slot=%d component=%s reason=%s.",
            prompt_id,
            slot_index,
            warmup_request.get("component_id"),
            reason,
        )
        _invoke_modal_warmup_blocking(_warmup_slot_payload(warmup_request, slot_index))
    except Exception:
        with _PROMPT_WARMUP_STATES_LOCK:
            warmup_state = _PROMPT_WARMUP_STATES.get(prompt_id)
            if warmup_state is not None:
                warmup_state.scheduled_slots.discard(slot_index)
        logger.exception(
            "Proactive Modal warmup failed for prompt=%s slot=%d component=%s.",
            prompt_id,
            slot_index,
            warmup_request.get("component_id"),
        )


def _track_prompt_warmup_future(
    prompt_id: str,
    slot_index: int,
    future: Future[Any],
) -> None:
    """Track one in-flight warmup future so mapped execution can await short exact warmup bursts."""
    with _PROMPT_WARMUP_STATES_LOCK:
        warmup_state = _PROMPT_WARMUP_STATES.get(prompt_id)
        if warmup_state is not None:
            warmup_state.slot_futures[slot_index] = future

    def _clear_tracked_future(completed_future: Future[Any]) -> None:
        """Drop one completed future from the prompt warmup state."""
        with _PROMPT_WARMUP_STATES_LOCK:
            warmup_state = _PROMPT_WARMUP_STATES.get(prompt_id)
            if warmup_state is None:
                return
            tracked_future = warmup_state.slot_futures.get(slot_index)
            if tracked_future is completed_future:
                warmup_state.slot_futures.pop(slot_index, None)

    future.add_done_callback(_clear_tracked_future)


async def _await_prompt_warmup_slots(
    prompt_id: str,
    slot_indices: list[int],
    timeout_seconds: float,
) -> int:
    """Wait briefly for any in-flight prompt warmup slots to finish."""
    if timeout_seconds <= 0.0 or not slot_indices:
        return 0

    with _PROMPT_WARMUP_STATES_LOCK:
        warmup_state = _PROMPT_WARMUP_STATES.get(prompt_id)
        if warmup_state is None:
            return 0
        pending_slot_futures = [
            warmup_state.slot_futures[slot_index]
            for slot_index in slot_indices
            if slot_index in warmup_state.slot_futures
            and not warmup_state.slot_futures[slot_index].done()
        ]

    if not pending_slot_futures:
        return 0

    wrapped_futures = [
        asyncio.wrap_future(slot_future) for slot_future in pending_slot_futures
    ]
    done_futures, pending_futures = await asyncio.wait(
        wrapped_futures,
        timeout=timeout_seconds,
        return_when=asyncio.ALL_COMPLETED,
    )
    if done_futures:
        logger.info(
            "Prompt=%s warmup head-start completed %d/%d slot(s) before mapped dispatch.",
            prompt_id,
            len(done_futures),
            len(wrapped_futures),
        )
    elif pending_futures:
        logger.info(
            "Prompt=%s warmup head-start timed out after %.3fs with %d slot(s) still warming.",
            prompt_id,
            timeout_seconds,
            len(pending_futures),
        )
    return len(done_futures)


def ensure_remote_warm_capacity(
    warmup_request: dict[str, Any] | None,
    *,
    warmup_target: int,
    reason: str,
) -> int:
    """Best-effort background warmup so enough Modal containers are ready for one prompt."""
    settings = get_settings()
    if not settings.enable_proactive_warmup:
        return 0
    if settings.execution_mode == "local" or modal is None:
        return 0
    if warmup_request is None:
        return 0

    prompt_id = _warmup_prompt_id(warmup_request)
    if prompt_id is None:
        return 0

    clamped_target = _clamp_prompt_warmup_target(warmup_target)
    if clamped_target <= 0:
        return 0

    with _PROMPT_WARMUP_STATES_LOCK:
        warmup_state = _ensure_prompt_warmup_state(prompt_id)
        missing_slots = [
            slot_index
            for slot_index in range(clamped_target)
            if slot_index not in warmup_state.scheduled_slots
        ]
        for slot_index in missing_slots:
            warmup_state.scheduled_slots.add(slot_index)

    if not missing_slots:
        return clamped_target

    logger.info(
        "Scheduling proactive Modal warmup for prompt=%s target=%d missing_slots=%s component=%s reason=%s.",
        prompt_id,
        clamped_target,
        missing_slots,
        warmup_request.get("component_id"),
        reason,
    )
    for slot_index in missing_slots:
        future = _REMOTE_MODAL_WARMUP_EXECUTOR.submit(
            _run_prompt_warmup_slot,
            prompt_id,
            slot_index,
            copy.deepcopy(warmup_request),
            reason,
        )
        _track_prompt_warmup_future(prompt_id, slot_index, future)
    return clamped_target


def _run_speculative_affinity_prewarm(
    prompt_id: str,
    warmup_identity: str,
    warmup_request: dict[str, Any],
    reason: str,
) -> None:
    """Prepare one future affinity worker and make a failed attempt retryable."""
    affinity_key = _remote_worker_affinity_key(warmup_request)
    try:
        logger.info(
            "Starting speculative Modal prewarm prompt=%s affinity=%s component=%s reason=%s.",
            prompt_id,
            affinity_key,
            warmup_request.get("component_id"),
            reason,
        )
        _invoke_modal_warmup_blocking(_warmup_slot_payload(warmup_request, 0))
        logger.info(
            "Completed speculative Modal prewarm prompt=%s affinity=%s component=%s.",
            prompt_id,
            affinity_key,
            warmup_request.get("component_id"),
        )
    except Exception:
        with _PROMPT_WARMUP_STATES_LOCK:
            warmup_state = _PROMPT_WARMUP_STATES.get(prompt_id)
            if warmup_state is not None:
                warmup_state.scheduled_speculative_affinities.discard(
                    warmup_identity
                )
        logger.exception(
            "Speculative Modal prewarm failed prompt=%s affinity=%s component=%s.",
            prompt_id,
            affinity_key,
            warmup_request.get("component_id"),
        )


def _schedule_speculative_affinity_prewarm(
    payload: Mapping[str, Any],
    *,
    reason: str,
) -> bool:
    """Schedule one planner-selected future worker without blocking current execution."""
    target_payload = payload.get(_SPECULATIVE_PREWARM_TARGET_KEY)
    if not isinstance(target_payload, Mapping):
        return False
    if (
        payload.get("execution_provider") != "modal"
        or target_payload.get("execution_provider") != "modal"
        or payload.get("execution_environment_id")
        != target_payload.get("execution_environment_id")
    ):
        logger.debug(
            "Skipping speculative Modal prewarm across execution environments "
            "source=%s target=%s.",
            payload.get("execution_environment_id"),
            target_payload.get("execution_environment_id"),
        )
        return False

    normalized_target_payload = copy.deepcopy(dict(target_payload))
    settings = _settings_for_payload(normalized_target_payload)
    if (
        not settings.enable_proactive_warmup
        or settings.execution_mode == "local"
        or modal is None
        or (settings.max_containers is not None and settings.max_containers < 2)
    ):
        return False

    warmup_request = _build_prompt_warmup_request(normalized_target_payload)
    if warmup_request is None:
        return False
    prompt_id = _warmup_prompt_id(warmup_request)
    if prompt_id is None:
        return False

    current_affinity_key = _remote_worker_affinity_key(dict(payload))
    target_affinity_key = _remote_worker_affinity_key(warmup_request)
    if target_affinity_key == current_affinity_key:
        return False
    snapshot_profile_key = _prepare_snapshot_profile_fields(warmup_request)
    _select_gpu_snapshot_for_profile(warmup_request, snapshot_profile_key)
    warmup_identity = _speculative_warmup_identity(warmup_request)

    with _PROMPT_WARMUP_STATES_LOCK:
        warmup_state = _ensure_prompt_warmup_state(prompt_id)
        if warmup_identity in warmup_state.scheduled_speculative_affinities:
            return False
        warmup_state.scheduled_speculative_affinities.add(warmup_identity)

    logger.info(
        "Scheduling speculative Modal prewarm prompt=%s current_component=%s current_affinity=%s target_component=%s target_affinity=%s reason=%s.",
        prompt_id,
        payload.get("component_id"),
        current_affinity_key,
        warmup_request.get("component_id"),
        target_affinity_key,
        reason,
    )
    future = _REMOTE_MODAL_WARMUP_EXECUTOR.submit(
        _run_speculative_affinity_prewarm,
        prompt_id,
        warmup_identity,
        copy.deepcopy(warmup_request),
        reason,
    )
    with _PROMPT_WARMUP_STATES_LOCK:
        warmup_state = _PROMPT_WARMUP_STATES.get(prompt_id)
        if warmup_state is not None:
            warmup_state.speculative_affinity_futures[warmup_identity] = future

    def _clear_speculative_future(completed_future: Future[Any]) -> None:
        """Drop a completed speculative future while retaining successful dedupe state."""
        with _PROMPT_WARMUP_STATES_LOCK:
            warmup_state = _PROMPT_WARMUP_STATES.get(prompt_id)
            if warmup_state is None:
                return
            tracked_future = warmup_state.speculative_affinity_futures.get(
                warmup_identity
            )
            if tracked_future is completed_future:
                warmup_state.speculative_affinity_futures.pop(
                    warmup_identity, None
                )

    future.add_done_callback(_clear_speculative_future)
    return True


def _speculative_warmup_identity(payload: Mapping[str, Any]) -> str:
    """Return the exact deployed-worker identity shared by warmup and dispatch."""
    settings = _settings_for_payload(payload)
    identity = {
        "app": modal_deployment_app_name(settings),
        "gpu": settings.modal_gpu,
        "affinity": _remote_worker_affinity_key(dict(payload)),
        "snapshot_profile": str(payload.get("snapshot_profile_key") or ""),
        "snapshot_variant": str(payload.get("gpu_snapshot_variant") or "direct"),
    }
    return json.dumps(identity, sort_keys=True, separators=(",", ":"))


def _await_matching_speculative_prewarm(
    payload: dict[str, Any],
    cancellation_event: threading.Event | None,
) -> None:
    """Join an active speculative warmup instead of racing a duplicate container."""
    prompt_id = _warmup_prompt_id(payload)
    if prompt_id is None:
        return
    snapshot_profile_key = _prepare_snapshot_profile_fields(payload)
    _select_gpu_snapshot_for_profile(payload, snapshot_profile_key)
    warmup_identity = _speculative_warmup_identity(payload)
    with _PROMPT_WARMUP_STATES_LOCK:
        warmup_state = _PROMPT_WARMUP_STATES.get(prompt_id)
        future = (
            warmup_state.speculative_affinity_futures.get(warmup_identity)
            if warmup_state is not None
            else None
        )
    if future is None or future.done():
        return
    started_at = time.perf_counter()
    logger.info(
        "Joining active speculative Modal prewarm before dispatch prompt=%s component=%s identity=%s.",
        prompt_id,
        payload.get("component_id"),
        warmup_identity,
    )
    while not future.done():
        if cancellation_event is not None and cancellation_event.is_set():
            raise ModalRemoteInvocationError(
                "Remote dispatch was cancelled while waiting for speculative warmup."
            )
        try:
            future.result(timeout=0.1)
        except FutureTimeoutError:
            continue
    future.result()
    logger.info(
        "Joined speculative Modal prewarm before dispatch prompt=%s component=%s wait_seconds=%.3f.",
        prompt_id,
        payload.get("component_id"),
        time.perf_counter() - started_at,
    )


def _schedule_post_deploy_runtime_seed(payload: Mapping[str, Any]) -> bool:
    """Seed the just-deployed profile and expose its future to the first dispatch."""
    warmup_request = _build_prompt_warmup_request(copy.deepcopy(dict(payload)))
    if warmup_request is None:
        return False
    prompt_id = _warmup_prompt_id(warmup_request)
    if prompt_id is None:
        return False
    snapshot_profile_key = _prepare_snapshot_profile_fields(warmup_request)
    _select_gpu_snapshot_for_profile(warmup_request, snapshot_profile_key)
    warmup_identity = _speculative_warmup_identity(warmup_request)
    with _PROMPT_WARMUP_STATES_LOCK:
        warmup_state = _ensure_prompt_warmup_state(prompt_id)
        if warmup_identity in warmup_state.scheduled_speculative_affinities:
            return False
        warmup_state.scheduled_speculative_affinities.add(warmup_identity)
        future = _REMOTE_MODAL_WARMUP_EXECUTOR.submit(
            _run_speculative_affinity_prewarm,
            prompt_id,
            warmup_identity,
            copy.deepcopy(warmup_request),
            "post_deploy_runtime_seed",
        )
        warmup_state.speculative_affinity_futures[warmup_identity] = future

    def clear_seed_future(completed_future: Future[Any]) -> None:
        """Forget the completed seed future while retaining successful dedupe state."""
        with _PROMPT_WARMUP_STATES_LOCK:
            current_state = _PROMPT_WARMUP_STATES.get(prompt_id)
            if current_state is None:
                return
            if (
                current_state.speculative_affinity_futures.get(warmup_identity)
                is completed_future
            ):
                current_state.speculative_affinity_futures.pop(
                    warmup_identity, None
                )

    future.add_done_callback(clear_seed_future)
    logger.info(
        "Scheduled automatic post-deployment Modal runtime seed prompt=%s component=%s identity=%s.",
        prompt_id,
        warmup_request.get("component_id"),
        warmup_identity,
    )
    return True


def _local_gap_keepalive_key(payload: Mapping[str, Any]) -> tuple[str, int] | None:
    """Return the prompt and affinity-slot key for a local-gap payload."""
    prompt_id = _warmup_prompt_id(dict(payload))
    if prompt_id is None:
        return None
    return (prompt_id, _component_pool_slot_index(dict(payload)))


def _stop_local_gap_keepalive(payload: Mapping[str, Any], *, reason: str) -> bool:
    """Stop a previously scheduled affinity-slot keepalive, if one exists."""
    keepalive_key = _local_gap_keepalive_key(payload)
    if keepalive_key is None:
        return False
    with _LOCAL_GAP_KEEPALIVES_LOCK:
        keepalive_state = _LOCAL_GAP_KEEPALIVES.pop(keepalive_key, None)
    if keepalive_state is None:
        return False
    keepalive_state.stop_event.set()
    logger.info(
        "Stopping remote local-gap keepalive prompt=%s slot=%d after component=%s reason=%s.",
        keepalive_key[0],
        keepalive_key[1],
        keepalive_state.component_id,
        reason,
    )
    return True


def _invoke_remote_local_gap_keepalive(payload: dict[str, Any]) -> Any:
    """Send one lightweight activity pulse to the payload's affinity slot."""
    remote_engine = _lookup_deployed_remote_engine(payload)
    keepalive_method = getattr(remote_engine, "keepalive_for_local_gap", None)
    if keepalive_method is None:
        raise ModalRemoteInvocationError(
            "The deployed RemoteEngine does not expose keepalive_for_local_gap."
        )
    return _call_modal_method(keepalive_method, payload)


def _run_local_gap_keepalive(
    payload: dict[str, Any],
    stop_event: threading.Event,
    duration_seconds: float,
    interval_seconds: float,
) -> None:
    """Pulse one remote affinity slot until local execution finishes or the budget expires."""
    deadline = time.monotonic() + duration_seconds
    keepalive_count = 0
    while not stop_event.is_set():
        remaining_seconds = deadline - time.monotonic()
        if remaining_seconds <= 0.0:
            break
        if stop_event.wait(min(interval_seconds, remaining_seconds)):
            break
        keepalive_count += 1
        pulse_started_at = time.perf_counter()
        _invoke_remote_local_gap_keepalive(payload)
        logger.info(
            "Completed remote local-gap keepalive pulse=%d prompt=%s component=%s in %.3fs.",
            keepalive_count,
            payload.get("prompt_id"),
            payload.get("component_id"),
            time.perf_counter() - pulse_started_at,
        )
    logger.info(
        "Remote local-gap keepalive loop finished prompt=%s component=%s pulses=%d stopped=%s.",
        payload.get("prompt_id"),
        payload.get("component_id"),
        keepalive_count,
        stop_event.is_set(),
    )


def _start_local_gap_keepalive(payload: Mapping[str, Any]) -> bool:
    """Retain one remote affinity slot while a downstream local gap executes."""
    if (
        payload.get("execution_provider") != "modal"
        or not bool(payload.get("keepalive_after_remote_component"))
    ):
        return False
    settings = _settings_for_payload(payload)
    duration_seconds = max(0.0, float(settings.local_gap_keepalive_seconds))
    if duration_seconds <= 0.0 or settings.execution_mode == "local" or modal is None:
        return False
    keepalive_key = _local_gap_keepalive_key(payload)
    if keepalive_key is None:
        return False

    _stop_local_gap_keepalive(payload, reason="replacement")
    stop_event = threading.Event()
    keepalive_payload = dict(payload)
    keepalive_payload["warmup_slot_index"] = keepalive_key[1]
    keepalive_payload["component_id"] = (
        f"{payload.get('component_id', 'modal-component')}::local-gap-keepalive"
    )
    interval_seconds = min(
        duration_seconds,
        max(1.0, float(settings.local_gap_keepalive_interval_seconds)),
    )
    future = _REMOTE_MODAL_KEEPALIVE_EXECUTOR.submit(
        _run_local_gap_keepalive,
        keepalive_payload,
        stop_event,
        duration_seconds,
        interval_seconds,
    )
    keepalive_state = _LocalGapKeepaliveState(
        component_id=str(payload.get("component_id") or "modal-component"),
        stop_event=stop_event,
        future=future,
    )
    with _LOCAL_GAP_KEEPALIVES_LOCK:
        _LOCAL_GAP_KEEPALIVES[keepalive_key] = keepalive_state

    def clear_completed_keepalive(completed_future: Future[Any]) -> None:
        """Forget a finished keepalive and report any best-effort failure."""
        with _LOCAL_GAP_KEEPALIVES_LOCK:
            current_state = _LOCAL_GAP_KEEPALIVES.get(keepalive_key)
            if current_state is keepalive_state:
                _LOCAL_GAP_KEEPALIVES.pop(keepalive_key, None)
        failure = completed_future.exception()
        if failure is not None:
            logger.warning(
                "Remote local-gap keepalive failed prompt=%s slot=%d component=%s: %s",
                keepalive_key[0],
                keepalive_key[1],
                keepalive_state.component_id,
                failure,
            )

    future.add_done_callback(clear_completed_keepalive)
    logger.info(
        "Started remote local-gap keepalive prompt=%s slot=%d component=%s duration=%.1fs interval=%.1fs.",
        keepalive_key[0],
        keepalive_key[1],
        keepalive_state.component_id,
        duration_seconds,
        interval_seconds,
    )
    return True


def boost_mapped_component_warmup(
    payload: dict[str, Any],
    *,
    total_items: int,
    reason: str,
) -> tuple[int, int]:
    """Record exact mapped fan-out and top up prompt warmup for the resulting lane count."""
    parallelism = _mapped_execution_parallelism(total_items)
    refined_prompt_warmup_target = _register_exact_component_parallelism(
        payload, parallelism
    )
    ensure_remote_warm_capacity(
        _build_prompt_warmup_request(payload),
        warmup_target=refined_prompt_warmup_target,
        reason=reason,
    )
    logger.info(
        "Boosted exact mapped warmup for component=%s total_items=%d local_parallelism=%d prompt_warmup_target=%d reason=%s.",
        payload.get("component_id"),
        total_items,
        parallelism,
        refined_prompt_warmup_target,
        reason,
    )
    return parallelism, refined_prompt_warmup_target


def _auto_deploy_modal_app(payload: dict[str, Any], lookup_error: BaseException) -> Any:
    """Deploy the stable Modal cloud app once and wait until deployed lookup becomes ready."""
    if modal is None:
        raise ModalRemoteInvocationError("Modal SDK is unavailable.")

    settings = _settings_for_payload(payload)
    deployment_app_name = modal_deployment_app_name(settings)
    deploy_key = _modal_deploy_cache_key(payload)
    deploy_state = _modal_auto_deploy_state(deploy_key)
    with _modal_cloud_settings_override(settings):
        cloud_module = _load_modal_cloud_module()
    cloud_app = getattr(cloud_module, "app", None)
    if cloud_app is None:
        raise ModalRemoteInvocationError(
            "Stable Modal cloud entry module did not expose a deployable app."
        )

    while True:
        with deploy_state.condition:
            if (
                not _is_missing_modal_deployment_error(lookup_error)
                and deploy_state.ready
            ):
                logger.info(
                    "Auto-deploy already completed for app=%s env=%s; reusing cached deployment state.",
                    deployment_app_name,
                    deploy_key[1] or "<default>",
                )
                return _lookup_deployed_remote_engine(payload)
            if _is_missing_modal_deployment_error(lookup_error) and deploy_state.ready:
                logger.warning(
                    "Discarding stale auto-deploy ready state for app=%s env=%s after missing deployment lookup failure: %s",
                    deployment_app_name,
                    deploy_key[1] or "<default>",
                    lookup_error,
                )
                deploy_state.ready = False
            if deploy_state.deploy_in_progress:
                logger.info(
                    "Waiting for in-flight auto-deploy readiness for app=%s env=%s component=%s.",
                    deployment_app_name,
                    deploy_key[1] or "<default>",
                    payload.get("component_id"),
                )
                _emit_local_remote_startup_status(
                    payload,
                    phase="setup",
                    status_message="Waiting for Modal app rebuild",
                )
                deploy_state.condition.wait()
                if deploy_state.ready:
                    _emit_local_remote_dispatch_status(payload)
                    return _lookup_deployed_remote_engine(payload)
                if deploy_state.last_error is not None:
                    logger.warning(
                        "Retrying Modal auto-deploy for app=%s env=%s after previous readiness failure: %s",
                        deployment_app_name,
                        deploy_key[1] or "<default>",
                        deploy_state.last_error,
                    )
                continue
            deploy_state.deploy_in_progress = True
            deploy_state.last_error = None
            break

    _emit_local_remote_startup_status(
        payload,
        phase="setup",
        status_message="Rebuilding Modal app",
    )
    try:
        logger.warning(
            "Deployed Modal app lookup failed for app=%s component=%s: %s. "
            "Attempting first-run auto-deploy from the custom node.",
            deployment_app_name,
            payload.get("component_id"),
            lookup_error,
        )
        output_context = (
            modal.enable_output() if hasattr(modal, "enable_output") else nullcontext()
        )
        deploy_started_at = time.perf_counter()
        with output_context:
            cloud_app.deploy(
                name=deployment_app_name,
                environment_name=deploy_key[1],
            )
        logger.info(
            "Auto-deployed Modal app %s for env=%s in %.3fs; waiting for deployed lookup readiness.",
            deployment_app_name,
            deploy_key[1] or "<default>",
            time.perf_counter() - deploy_started_at,
        )
        _ensure_llm_profiles_staged(payload, deployment_app_name)
        remote_engine = _lookup_deployed_remote_engine_with_retry(payload)
        if not _is_remote_engine_runtime_current(remote_engine, payload):
            raise ModalRemoteInvocationError(
                f"Auto-deployed Modal app {deployment_app_name!r} did not report expected protocol "
                f"{_REMOTE_APP_PROTOCOL_VERSION} and fingerprint "
                f"{_expected_remote_runtime_fingerprint(payload)!r}."
            )
    except BaseException as exc:
        with deploy_state.condition:
            deploy_state.deploy_in_progress = False
            deploy_state.last_error = exc
            deploy_state.condition.notify_all()
        raise

    with deploy_state.condition:
        deploy_state.deploy_in_progress = False
        deploy_state.ready = True
        deploy_state.last_error = None
        deploy_state.condition.notify_all()
    with _MODAL_AUTO_DEPLOY_LOCK:
        _MODAL_REMOTE_APP_VERSION_OK.add(_modal_runtime_cache_key(payload))
    logger.info(
        "Deployed Modal app %s for env=%s is now lookup-ready.",
        deployment_app_name,
        deploy_key[1] or "<default>",
    )
    if _schedule_post_deploy_runtime_seed(payload):
        _await_matching_speculative_prewarm(payload, None)
    _emit_local_remote_dispatch_status(payload)
    return remote_engine


def _build_remote_interrupt_callback(
    _remote_engine: Any, payload: dict[str, Any]
) -> Callable[[], Any] | None:
    """Return a callable that requests interruption for one active Modal payload."""
    interrupt_store = _lookup_modal_interrupt_store()
    if interrupt_store is None:
        return None
    return lambda: _request_remote_interrupt(payload)


def _invoke_remote_engine_payload(
    remote_engine: Any,
    payload: dict[str, Any],
    kwargs_payload: bytes,
    cancellation_event: threading.Event | None,
) -> bytes:
    """Invoke one prepared remote engine instance with optional progress streaming."""
    _install_modal_cloud_exception_compatibility_module()
    if cancellation_event is not None and cancellation_event.is_set():
        _request_remote_interrupt(payload)
        raise ModalRemoteInvocationError(
            "Remote Modal payload dispatch was cancelled before the remote call started."
        )

    stream_method = getattr(remote_engine, "execute_payload_stream", None)
    interrupt_remote_call = _build_remote_interrupt_callback(remote_engine, payload)
    if _should_stream_remote_progress(payload) and hasattr(stream_method, "remote_gen"):
        logger.info(
            "Using streamed Modal progress path for component=%s via execute_payload_stream.remote_gen(...).",
            payload.get("component_id"),
        )
        return _invoke_remote_call_with_interrupts(
            payload=payload,
            invoke_remote_call=lambda: _consume_remote_payload_stream(
                payload,
                stream_method.remote_gen(payload, kwargs_payload),
            ),
            interrupt_remote_call=interrupt_remote_call,
            cancellation_event=cancellation_event,
        )

    if _should_stream_remote_progress(payload):
        logger.warning(
            "Streamed Modal progress is unavailable for component=%s; falling back to execute_payload.remote(...).",
            payload.get("component_id"),
        )
    return _invoke_remote_call_with_interrupts(
        payload=payload,
        invoke_remote_call=lambda: remote_engine.execute_payload.remote(
            payload, kwargs_payload
        ),
        interrupt_remote_call=interrupt_remote_call,
        cancellation_event=cancellation_event,
    )


def _fresh_llm_memory_recovery_affinity_key(payload: Mapping[str, Any]) -> str:
    """Return a one-use affinity identity that cannot select the dirty worker."""
    return (
        f"{_remote_worker_affinity_key(dict(payload))}:llm-memory-recovery:"
        f"{uuid.uuid4().hex[:12]}"
    )


def _retry_exhausted_llm_memory_on_fresh_worker(
    *,
    payload: dict[str, Any],
    kwargs_payload: bytes,
    cancellation_event: threading.Event | None,
    error: BaseException,
) -> bytes:
    """Retry one exhausted post-eviction admission on a distinct Modal worker."""
    recovery_payload = dict(payload)
    if exhausted_recovery_used_vllm_throughput(error):
        recovery_payload[LLM_FORCE_VLLM_THROUGHPUT_PAYLOAD_KEY] = True
    recovery_affinity_key = _fresh_llm_memory_recovery_affinity_key(payload)
    logger.warning(
        "Modal LLM memory recovery timed out for component=%s; retrying "
        "invocation_id=%s once on fresh worker_affinity=%s.",
        payload.get("component_id"),
        payload["invocation_id"],
        recovery_affinity_key,
    )
    _emit_local_remote_startup_status(
        recovery_payload,
        phase="starting",
        status_message="Retrying LLM on a fresh Modal worker",
    )
    recovered_remote_engine = _lookup_deployed_remote_engine(
        recovery_payload,
        affinity_key_override=recovery_affinity_key,
    )
    return _invoke_remote_engine_payload(
        recovered_remote_engine,
        recovery_payload,
        kwargs_payload,
        cancellation_event,
    )


def _invoke_remote_engine_payload_with_recovery(
    remote_engine: Any,
    payload: dict[str, Any],
    kwargs_payload: bytes,
    cancellation_event: threading.Event | None,
) -> bytes:
    """Recover missing deployments and exhausted LLM workers with one safe retry."""
    payload = dict(payload)
    settings = _settings_for_payload(payload)
    if llm_model_references_from_payload(payload):
        kwargs_payload = _rewrite_staged_llm_kwargs_payload(
            kwargs_payload,
            modal_deployment_app_name(settings),
        )
    payload.setdefault("capture_remote_outputs", True)
    payload.setdefault(
        "invocation_id",
        stable_remote_invocation_id(payload, kwargs_payload),
    )
    lookup_error_types = _modal_lookup_error_types()
    invocation_error_types = tuple(
        dict.fromkeys((*lookup_error_types, RemoteSubgraphExecutionError, RuntimeError))
    )
    try:
        response = _invoke_remote_engine_payload(
            remote_engine,
            payload,
            kwargs_payload,
            cancellation_event,
        )
    except invocation_error_types as exc:
        if is_llm_memory_recovery_exhausted(exc):
            response = _retry_exhausted_llm_memory_on_fresh_worker(
                payload=payload,
                kwargs_payload=kwargs_payload,
                cancellation_event=cancellation_event,
                error=exc,
            )
        elif not isinstance(exc, lookup_error_types):
            raise
        elif not settings.auto_deploy or not _is_missing_modal_deployment_error(exc):
            raise
        else:
            logger.warning(
                "Modal payload invocation failed for component=%s because the deployed app was missing at call time: %s. Recreating the app and retrying.",
                payload.get("component_id"),
                exc,
            )
            recovered_remote_engine = _auto_deploy_modal_app(payload, exc)
            _ensure_llm_profiles_staged(
                payload,
                modal_deployment_app_name(settings),
            )
            response = _invoke_remote_engine_payload(
                recovered_remote_engine,
                payload,
                kwargs_payload,
                cancellation_event,
            )
    return _materialize_remote_execution_result(response, settings=settings)


def _local_comfy_output_directory(settings: ModalSyncSettings) -> Path:
    """Return the effective output directory of the local ComfyUI process."""
    try:
        import folder_paths
    except ModuleNotFoundError as exc:
        if exc.name != "folder_paths":
            raise
    else:
        get_output_directory = getattr(folder_paths, "get_output_directory", None)
        if callable(get_output_directory):
            return Path(get_output_directory()).expanduser().resolve()
    if settings.comfyui_root is None:
        raise ModalRemoteInvocationError(
            "Modal returned ComfyUI output files, but the local ComfyUI output directory "
            "could not be resolved. Set COMFY_MODAL_COMFYUI_ROOT."
        )
    return (settings.comfyui_root / "output").resolve()


def _materialize_remote_execution_result(
    response: bytes | bytearray,
    *,
    settings: ModalSyncSettings | None = None,
) -> bytes:
    """Download bundled remote files and return the ordinary serialized node outputs."""
    result: RemoteExecutionResult = unpack_remote_execution_result(response)
    if not result.artifacts:
        return result.outputs
    resolved_settings = settings or get_settings()
    output_directory = _local_comfy_output_directory(resolved_settings)
    logger.info(
        "Materializing %d remote ComfyUI output file(s) into %s.",
        len(result.artifacts),
        output_directory,
    )
    materialized_paths = materialize_remote_output_artifacts(
        result,
        output_directory=output_directory,
        app_name=modal_deployment_app_name(resolved_settings),
    )
    logger.info(
        "Finished downloading %d remote ComfyUI output file(s): %s.",
        len(materialized_paths),
        [str(path) for path in materialized_paths],
    )
    return result.outputs


def _invoke_modal_payload_blocking(
    payload: dict[str, Any],
    kwargs_payload: bytes,
    cancellation_event: threading.Event | None = None,
) -> bytes:
    """Invoke the Modal runtime from a worker thread using deployed or ephemeral app state."""
    if modal is None:
        raise ModalRemoteInvocationError("Modal SDK is unavailable.")

    _await_matching_speculative_prewarm(payload, cancellation_event)
    lookup_error_types = _modal_lookup_error_types()
    settings = _settings_for_payload(payload)
    deployment_app_name = modal_deployment_app_name(settings)
    if lookup_error_types:
        try:
            remote_engine = _lookup_protocol_current_remote_engine(payload)
            _ensure_llm_profiles_staged(payload, deployment_app_name)
            logger.info(
                "Using deployed Modal app %s for component %s.",
                deployment_app_name,
                payload.get("component_id"),
            )
            return _invoke_remote_engine_payload_with_recovery(
                remote_engine,
                payload,
                kwargs_payload,
                cancellation_event,
            )
        except lookup_error_types as exc:
            missing_deployment = _is_missing_modal_deployment_error(exc)
            if settings.auto_deploy and missing_deployment:
                remote_engine = _auto_deploy_modal_app(payload, exc)
                try:
                    _ensure_llm_profiles_staged(payload, deployment_app_name)
                    logger.info(
                        "Using auto-deployed Modal app %s for component %s.",
                        deployment_app_name,
                        payload.get("component_id"),
                    )
                    return _invoke_remote_engine_payload_with_recovery(
                        remote_engine,
                        payload,
                        kwargs_payload,
                        cancellation_event,
                    )
                except lookup_error_types as retry_exc:
                    exc = retry_exc
                    missing_deployment = _is_missing_modal_deployment_error(exc)
            if not missing_deployment:
                raise
            if not settings.allow_ephemeral_fallback:
                raise ModalRemoteInvocationError(
                    "Remote execution requires a deployed Modal app or a successful first-run auto-deploy. "
                    f"Lookup failed for app={deployment_app_name!r} component={payload.get('component_id')!r}: {exc}. "
                    "Ensure Modal credentials are configured so the custom node can auto-deploy, "
                    "or set COMFY_MODAL_ALLOW_EPHEMERAL_FALLBACK=true to allow slow ephemeral app.run() fallback behavior."
                ) from exc
            logger.warning(
                "Deployed Modal app lookup failed for app=%s component=%s: %s. Falling back to ephemeral app.run(); this creates a temporary Modal app session, not a persistent deployment or endpoint.",
                deployment_app_name,
                payload.get("component_id"),
                exc,
            )
    else:
        remote_engine = _lookup_protocol_current_remote_engine(payload)
        _ensure_llm_profiles_staged(payload, deployment_app_name)
        logger.info(
            "Using deployed Modal app %s for component %s.",
            deployment_app_name,
            payload.get("component_id"),
        )
        return _invoke_remote_engine_payload_with_recovery(
            remote_engine,
            payload,
            kwargs_payload,
            cancellation_event,
        )

    if "app" not in globals() or "RemoteEngine" not in globals():
        logger.debug(
            "Local module Modal runtime objects are unavailable; loading stable cloud entry module."
        )

    with _modal_cloud_settings_override(settings):
        cloud_module = _load_modal_cloud_module()
    cloud_app = getattr(cloud_module, "app", None)
    cloud_remote_engine = getattr(cloud_module, "RemoteEngine", None)
    if cloud_app is None or cloud_remote_engine is None:
        raise ModalRemoteInvocationError(
            "Stable Modal cloud entry module did not expose app and RemoteEngine."
        )
    logger.info(
        "Starting ephemeral Modal app.run() for component %s. This does not create a persistent deployed app or web endpoint.",
        payload.get("component_id"),
    )
    run_context = cloud_app.run() if hasattr(cloud_app, "run") else nullcontext()
    with run_context:
        remote_engine = cloud_remote_engine()
        result = _invoke_remote_engine_payload_with_recovery(
            remote_engine,
            payload,
            kwargs_payload,
            cancellation_event,
        )
    logger.info(
        "Ephemeral Modal app.run() invocation completed for component %s.",
        payload.get("component_id"),
    )
    return result


def invoke_remote_engine(
    payload: dict[str, Any],
    kwargs_payload: bytes,
    *,
    allow_implicit_mapping: bool = True,
) -> bytes:
    """Invoke Modal when configured, or fall back to local in-process execution."""
    execution_mode = get_settings().execution_mode
    if payload.get("payload_kind") == "mapped_subgraph":
        if execution_mode == "local" or modal is None:
            hydrated_inputs = deserialize_node_inputs(kwargs_payload)
            return serialize_node_outputs(
                _execute_mapped_subgraph_payload(payload, hydrated_inputs)
            )
    if allow_implicit_mapping and payload.get("payload_kind") == "subgraph":
        hydrated_inputs = deserialize_node_inputs(kwargs_payload)
        if _split_batch_boundary_inputs(payload, hydrated_inputs) is not None:
            return asyncio.run(
                _invoke_implicitly_mapped_subgraph_async(payload, kwargs_payload)
            )
    if execution_mode == "local" or modal is None:
        if payload.get("payload_kind") == "subgraph":
            return execute_subgraph_locally(payload, kwargs_payload)
        return execute_node_locally(payload, kwargs_payload)

    logger.info(
        "Dispatching Modal remote invocation for component=%s payload_kind=%s.",
        payload.get("component_id"),
        payload.get("payload_kind"),
    )
    _emit_local_remote_dispatch_status(payload)
    cancellation_event = threading.Event()
    future = _REMOTE_MODAL_CALL_EXECUTOR.submit(
        _invoke_modal_payload_blocking,
        dict(payload),
        kwargs_payload,
        cancellation_event,
    )
    interrupt_sent = False
    cancellation_started_at: float | None = None
    try:
        while True:
            try:
                response = future.result(timeout=0.1)
                break
            except FutureTimeoutError:
                (
                    interrupt_sent,
                    cancellation_started_at,
                ) = _handle_modal_wait_cancellation(
                    payload,
                    cancellation_event,
                    interrupt_sent=interrupt_sent,
                    cancellation_started_at=cancellation_started_at,
                )
                continue
    except Exception:
        if cancellation_event.is_set() or _local_processing_interrupted():
            logger.info(
                "Reraising Modal failure as a local interrupt for component=%s after cancellation.",
                payload.get("component_id"),
            )
            _raise_local_interrupt()
        logger.exception(
            "Modal remote invocation failed for component=%s.",
            payload.get("component_id"),
        )
        raise
    if cancellation_event.is_set() or _local_processing_interrupted():
        logger.info(
            "Remote invocation for component=%s finished after interruption; raising local interrupt.",
            payload.get("component_id"),
        )
        _raise_local_interrupt()
    logger.info(
        "Modal remote invocation completed for component=%s.",
        payload.get("component_id"),
    )
    return response


async def invoke_remote_engine_async(
    payload: dict[str, Any],
    kwargs_payload: bytes,
    *,
    allow_implicit_mapping: bool = True,
) -> bytes:
    """Invoke Modal asynchronously so multiple proxy nodes can wait on remote work in parallel."""
    execution_mode = get_settings().execution_mode
    if (
        execution_mode != "local"
        and modal is not None
        and bool(payload.get("stop_local_gap_keepalive_before_remote_component"))
    ):
        _stop_local_gap_keepalive(payload, reason="next_remote_component_started")
    if payload.get("payload_kind") == "mapped_subgraph":
        if execution_mode == "local" or modal is None:
            return await _invoke_mapped_remote_engine_async(payload, kwargs_payload)
    if allow_implicit_mapping and payload.get("payload_kind") == "subgraph":
        hydrated_inputs = deserialize_node_inputs(kwargs_payload)
        if _split_batch_boundary_inputs(payload, hydrated_inputs) is not None:
            if execution_mode != "local" and modal is not None:
                await asyncio.to_thread(
                    _lookup_protocol_current_remote_engine,
                    payload,
                )
            return await _invoke_implicitly_mapped_subgraph_async(
                payload, kwargs_payload
            )
    if execution_mode == "local" or modal is None:
        return await asyncio.to_thread(
            invoke_remote_engine,
            payload,
            kwargs_payload,
            allow_implicit_mapping=allow_implicit_mapping,
        )

    logger.info(
        "Dispatching async Modal remote invocation for component=%s payload_kind=%s.",
        payload.get("component_id"),
        payload.get("payload_kind"),
    )
    _emit_local_remote_dispatch_status(payload)
    cancellation_event = threading.Event()
    future = _REMOTE_MODAL_CALL_EXECUTOR.submit(
        _invoke_modal_payload_blocking,
        dict(payload),
        kwargs_payload,
        cancellation_event,
    )
    wrapped_future = asyncio.wrap_future(future)
    interrupt_sent = False
    cancellation_started_at: float | None = None
    try:
        while True:
            try:
                response = await asyncio.wait_for(
                    asyncio.shield(wrapped_future), timeout=0.1
                )
                break
            except asyncio.TimeoutError:
                (
                    interrupt_sent,
                    cancellation_started_at,
                ) = await _handle_modal_wait_cancellation_async(
                    payload,
                    cancellation_event,
                    interrupt_sent=interrupt_sent,
                    cancellation_started_at=cancellation_started_at,
                )
                continue
    except asyncio.CancelledError:
        cancellation_event.set()
        prompt_id = (
            str(payload.get("prompt_id"))
            if payload.get("prompt_id") is not None
            else None
        )
        if prompt_id is not None:
            await request_remote_modal_prompt_interrupt_async(prompt_id)
        raise
    except Exception:
        if cancellation_event.is_set() or _local_processing_interrupted():
            logger.info(
                "Reraising async Modal failure as a local interrupt for component=%s after cancellation.",
                payload.get("component_id"),
            )
            _raise_local_interrupt()
        logger.exception(
            "Async Modal remote invocation failed for component=%s.",
            payload.get("component_id"),
        )
        raise
    if cancellation_event.is_set() or _local_processing_interrupted():
        logger.info(
            "Async remote invocation for component=%s finished after interruption; raising local interrupt.",
            payload.get("component_id"),
        )
        _raise_local_interrupt()
    logger.info(
        "Async Modal remote invocation completed for component=%s.",
        payload.get("component_id"),
    )
    _start_local_gap_keepalive(payload)
    return response


if modal is not None:  # pragma: no branch - simple import-time configuration.
    settings = get_settings()
    app = modal.App(modal_deployment_app_name(settings))
    vol = modal.Volume.from_name(settings.volume_name, create_if_missing=True)
    image = modal.Image.debian_slim().pip_install(
        "torch", "safetensors", "pillow", "numpy"
    )

    @app.cls(
        gpu=settings.modal_gpu,
        volumes={settings.remote_storage_root: vol},
        scaledown_window=60,
        image=image,
    )
    class RemoteEngine:
        """Modal runtime class that executes proxied ComfyUI payloads."""

        snapshot_profile_key: str = modal.parameter(default="")
        gpu_snapshot_enabled: bool = modal.parameter(default=False)
        worker_affinity_key: str = modal.parameter(default="worker-pool:slot:0")

        @modal.enter()
        def setup(self) -> None:
            """Prepare the container process for headless node execution."""
            logger.info(
                "RemoteEngine setup complete for snapshot_profile_key=%s gpu_snapshot_enabled=%s.",
                self.snapshot_profile_key or None,
                bool(self.gpu_snapshot_enabled),
            )

        @modal.method()
        def execute_payload(
            self, payload: dict[str, Any], kwargs_payload: bytes
        ) -> bytes:
            """Execute a proxied node or subgraph inside the Modal container."""
            if payload.get("payload_kind") == "mapped_subgraph":
                hydrated_inputs = deserialize_node_inputs(kwargs_payload)
                return serialize_node_outputs(
                    _execute_mapped_subgraph_payload(payload, hydrated_inputs)
                )
            if payload.get("payload_kind") == "subgraph":
                return execute_subgraph_locally(payload, kwargs_payload)
            return execute_node_locally(payload, kwargs_payload)

        @modal.method()
        def warmup_for_request(self, payload: dict[str, Any]) -> dict[str, Any]:
            """No-op local warmup entrypoint for the simplified Modal runtime."""
            return {"component_id": str(payload.get("component_id") or "modal-warmup")}

        @modal.method()
        def keepalive_for_local_gap(self, payload: dict[str, Any]) -> dict[str, Any]:
            """Return a lightweight keepalive acknowledgement for one affinity slot."""
            return {
                "component_id": str(payload.get("component_id") or "modal-keepalive"),
                "worker_affinity_key": self.worker_affinity_key,
            }

else:

    class RemoteEngine:
        """Local fallback runtime used when the Modal SDK is unavailable."""

        def __init__(
            self,
            snapshot_profile_key: str | None = None,
            gpu_snapshot_enabled: bool = False,
            worker_affinity_key: str = "worker-pool:slot:0",
        ) -> None:
            """Record the optional snapshot-profile key and GPU snapshot mode."""
            self.snapshot_profile_key = snapshot_profile_key
            self.gpu_snapshot_enabled = gpu_snapshot_enabled
            self.worker_affinity_key = worker_affinity_key

        def setup(self) -> None:
            """No-op setup for local fallback execution."""

        def execute_payload(
            self, payload: dict[str, Any], kwargs_payload: bytes
        ) -> bytes:
            """Execute the proxied node or subgraph locally."""
            if payload.get("payload_kind") == "mapped_subgraph":
                hydrated_inputs = deserialize_node_inputs(kwargs_payload)
                return serialize_node_outputs(
                    _execute_mapped_subgraph_payload(payload, hydrated_inputs)
                )
            if payload.get("payload_kind") == "subgraph":
                return execute_subgraph_locally(payload, kwargs_payload)
            return execute_node_locally(payload, kwargs_payload)

        def warmup_for_request(self, payload: dict[str, Any]) -> dict[str, Any]:
            """Return a local no-op warmup result when Modal is unavailable."""
            return {"component_id": str(payload.get("component_id") or "modal-warmup")}

        def keepalive_for_local_gap(self, payload: dict[str, Any]) -> dict[str, Any]:
            """Return a local keepalive acknowledgement when Modal is unavailable."""
            return {
                "component_id": str(payload.get("component_id") or "modal-keepalive"),
                "worker_affinity_key": self.worker_affinity_key,
            }