Files
ttulttul-ComfyUI-Modal/api_intercept.py

10133 lines
396 KiB
Python

"""Prompt interception and graph rewriting for Modal-backed execution."""
from __future__ import annotations
import asyncio
import copy
import hashlib
import json
import logging
import math
import os
import threading
import time
import uuid
from collections import defaultdict, deque
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field, replace
from pathlib import Path
from typing import Any, Callable, Iterable, Iterator, Mapping, Sequence
from aiohttp import web
from .modal_executor_node import (
MODAL_ARTIFACT_FINALIZER_MAX_COMPONENTS,
MODAL_ARTIFACT_FINALIZER_NODE_ID,
MODAL_COMPONENT_COMPLETION_OUTPUT_NAME,
MODAL_LOCAL_BRIDGE_MATERIALIZER_NODE_ID,
MODAL_MAP_INPUT_NODE_ID,
MODAL_PARALLEL_LOCAL_PASSTHROUGH_NODE_ID,
MODAL_PROMPT_ID_EXTRA_PNGINFO_KEY,
ensure_modal_artifact_finalizer_registered,
ensure_modal_component_proxy_node_registered,
ensure_modal_local_bridge_materializer_registered,
ensure_modal_parallel_local_passthrough_registered,
register_cache_friendly_proxy_payload,
register_modal_map_input_warmup_context,
registered_proxy_execution_payload,
update_registered_proxy_payload_fields,
)
from .session_state import RemoteSessionHandle
from .settings import (
ModalSyncSettings,
get_settings,
discover_comfyui_user_directory,
modal_gpu_from_workflow,
settings_for_modal_gpu,
)
from .execution_environments import (
ComponentResourceRequirements,
CostAwareEnvironmentScheduler,
EnvironmentCapabilities,
EnvironmentHealth,
EnvironmentSchedulingState,
ExecutionAssignment,
ExecutionPolicy,
ExecutionProvider,
GpuCapability,
NoCompatibleExecutionEnvironmentError,
WorkflowExecutionPreferences,
)
from .execution_history import ExecutionHistory
from .sync_engine import (
AssetSyncRequestCache,
MODEL_FILE_EXTENSIONS,
ModalAssetSyncEngine,
ModalVolumeBackend,
SyncCancelledError,
SyncedAsset,
begin_r2_writeback_prompt,
finish_r2_writeback_prompt,
modal,
resolve_model_path,
)
from .remote_hosts import RemoteExecutionConfig, RemoteHostRegistry, SshHostConfig
from .r2_cache import R2CacheClient, R2CacheError, R2StorageUsage
from .r2_credentials import (
R2CredentialError,
R2CredentialStore,
R2_KEYCHAIN_UNLOCK_REQUIRED_CODE,
request_macos_keychain_unlock,
)
from .cloudflare_oauth import setup_r2_oauth_routes
from .remote_configuration_nodes import (
REMOTE_CONFIGURATION_NODE_IDS,
REMOTE_EXECUTION_CONFIGURATOR_NODE_ID,
compile_remote_configuration_set,
)
from .remote_configurations import (
ModalRemoteConfiguration,
RemoteConfiguration,
RemoteConfigurationSet,
R2StorageBackingConfiguration,
SshRemoteConfiguration,
VastRemoteConfiguration,
)
from .ssh_docker import SshDockerController, SshDockerVolumeBackend
from .ssh_runtime import SshRuntimeManager
from .vast_config_node import extract_vast_profiles
from .vast_service import VastProfileQuote, VastSearchRequirements, VastService
from .vast_api import VastApiClient
from .vast_leases import VastLeaseManager, VastLeaseRegistry
logger = logging.getLogger(__name__)
SetupStatusCallback = Callable[[str, int | None, int | None], None]
EnvironmentSetupStatusCallback = Callable[
[str, str, int | None, int | None], None
]
ExecutionPlanStatusCallback = Callable[
[dict[str, dict[str, Any]], list[dict[str, Any]]], None
]
_ROUTE_REGISTERED = False
_MODAL_UI_EVENT_RETENTION_SECONDS = 2 * 60 * 60
_MODAL_UI_EVENT_LIMIT_PER_CLIENT = 512
_MODAL_UI_EVENTS_LOCK = threading.Lock()
_MODAL_UI_EVENTS_BY_CLIENT: dict[str, deque[dict[str, Any]]] = {}
_MODAL_INTERRUPT_QUEUE_BRIDGE_ATTR = "__comfy_modal_interrupt_queue_bridge_installed"
_REMOTE_PREPARATION_PROMPTS_ATTR = "__comfy_modal_remote_preparation_prompts"
_REMOTE_PREPARATION_CANCELLATIONS_ATTR = (
"__comfy_modal_remote_preparation_cancellations"
)
_REMOTE_PREPARATION_LOCK_ATTR = "__comfy_modal_remote_preparation_lock"
_REMOTE_ASSET_PREPARATION_MAX_WORKERS = 8
_R2_STORAGE_USAGE_ROUTE = "/remote/storage/r2/usage"
_R2_KEYCHAIN_UNLOCK_ROUTE = "/remote/storage/r2/keychain/unlock"
_R2_STORAGE_USAGE_CACHE_SECONDS = 5 * 60
_R2_STORAGE_USAGE_CACHE_LOCK = threading.Lock()
_R2_STORAGE_USAGE_CACHE: dict[
tuple[str, str, str], tuple[float, R2StorageUsage]
] = {}
_REMOTE_TOGGLE_WIDGET_NAME = "Run Remotely"
_LEGACY_REMOTE_TOGGLE_WIDGET_NAME = "Run on Modal"
_LOCAL_ONLY_REMOTE_CLASS_TYPES = frozenset(
{
"ModalEndpointChat",
"VastAILeaseConfiguration",
*REMOTE_CONFIGURATION_NODE_IDS,
MODAL_ARTIFACT_FINALIZER_NODE_ID,
MODAL_LOCAL_BRIDGE_MATERIALIZER_NODE_ID,
MODAL_MAP_INPUT_NODE_ID,
MODAL_PARALLEL_LOCAL_PASSTHROUGH_NODE_ID,
}
)
_TRANSPORTABLE_OUTPUT_TYPES = frozenset(
{
"*",
"AUDIO",
"BOOLEAN",
"FLOAT",
"IMAGE",
"INT",
"LATENT",
"MASK",
"SIGMAS",
"STRING",
"VIDEO",
}
)
_INEXPENSIVE_REMOTE_BOUNDARY_TYPES = frozenset(
{
"BOOLEAN",
"FLOAT",
"INT",
"STRING",
}
)
_ROOT_LOADER_PREWARM_CLASS_TYPES = frozenset(
{
"CheckpointLoaderSimple",
"UNETLoader",
"CLIPLoader",
"DualCLIPLoader",
}
)
_SPECULATIVE_PREWARM_TARGET_KEY = "speculative_remote_prewarm_target"
_SPECULATIVE_PREWARM_PAYLOAD_FIELDS = frozenset(
{
"component_id",
"prompt_id",
"modal_gpu",
"execution_provider",
"execution_environment_id",
"remote_worker_affinity_group",
"remote_local_gap_pool",
"subgraph_prompt",
"requires_volume_reload",
"volume_reload_marker",
"uploaded_volume_paths",
"custom_nodes_bundle",
"snapshot_profile_key",
}
)
_MODEL_VRAM_WEIGHT_MULTIPLIER = 1.20
_MODEL_VRAM_HEADROOM_BYTES = 4 * 1024**3
_MODEL_RAM_HEADROOM_BYTES = 4 * 1024**3
_REMOTE_REPLICA_NODE_PREFIX = "__ComfyModalReplica__"
_REPLICABLE_REMOTE_OBJECT_TYPES = frozenset(
{"CLIP", "MODEL", "NOISE", "SAMPLER", "VAE"}
)
_ADDITIVE_MODEL_CLASS_TOKENS = (
"adapter",
"controlnet",
"ipadapter",
"lora",
)
@dataclass(frozen=True)
class ComponentMemoryEstimate:
"""Describe conservative scheduler memory floors inferred from model assets."""
minimum_vram_bytes: int
minimum_ram_bytes: int
model_asset_count: int = 0
largest_model_bytes: int = 0
@dataclass(frozen=True)
class LinkedOutputRef:
"""Reference a node output slot within a prompt graph."""
node_id: str
output_index: int
@dataclass(frozen=True)
class InputTarget:
"""Describe a target input inside a remote component."""
node_id: str
input_name: str
@dataclass
class BoundaryInputSpec:
"""Describe one local-to-remote boundary value for a component."""
proxy_input_name: str
source: LinkedOutputRef
io_type: str
targets: list[InputTarget] = field(default_factory=list)
@dataclass
class BoundaryOutputSpec:
"""Describe one value exported across a remote component boundary."""
proxy_output_name: str
source: LinkedOutputRef
io_type: str
is_list: bool
preview_target_node_ids: list[str] = field(default_factory=list)
session_output: bool = False
session_consumer_node_ids: list[str] = field(default_factory=list)
local_materializer_node_id: str | None = None
local_materializer_consumer_node_ids: list[str] = field(default_factory=list)
@dataclass
class StaticToMappedBoundarySpec:
"""Describe one static-phase output injected into each mapped item run."""
proxy_name: str
source: LinkedOutputRef
io_type: str
is_list: bool
targets: list[InputTarget] = field(default_factory=list)
@dataclass(frozen=True)
class ProducedPhaseOutputSpec:
"""Describe one output published by an earlier split-proxy phase."""
proxy_output_name: str
source: LinkedOutputRef
io_type: str
is_list: bool
session_output: bool = False
@dataclass
class RemoteComponentPlan:
"""Execution and rewrite plan for one connected remote component."""
node_ids: list[str]
representative_node_id: str
boundary_inputs: list[BoundaryInputSpec]
boundary_outputs: list[BoundaryOutputSpec]
execute_node_ids: list[str]
contains_output_node: bool
mapped_boundary_input_name: str | None = None
mapped_boundary_input_io_type: str | None = None
mapped_boundary_source_node_id: str | None = None
static_node_ids: list[str] = field(default_factory=list)
mapped_node_ids: list[str] = field(default_factory=list)
mapped_execute_node_ids: list[str] = field(default_factory=list)
static_execute_node_ids: list[str] = field(default_factory=list)
static_to_mapped_boundaries: list[StaticToMappedBoundarySpec] = field(
default_factory=list
)
local_tap_node_ids: list[str] = field(default_factory=list)
local_tap_terminal_node_ids: list[str] = field(default_factory=list)
@dataclass
class RewriteSummary:
"""Summary of the prompt rewrite performed for a queue request."""
remote_node_ids: list[str] = field(default_factory=list)
remote_component_ids: list[str] = field(default_factory=list)
component_node_ids_by_representative: dict[str, list[str]] = field(
default_factory=dict
)
component_dependency_ids_by_representative: dict[str, list[str]] = field(
default_factory=dict
)
component_execution_stages: list[list[str]] = field(default_factory=list)
mapped_component_ids: list[str] = field(default_factory=list)
estimated_max_parallel_requests: int = 0
max_parallel_requests_upper_bound: int | None = None
requires_volume_reload: bool = False
volume_reload_marker: str | None = None
uploaded_volume_paths: list[str] = field(default_factory=list)
rewritten_node_id_map: dict[str, str] = field(default_factory=dict)
sandwiched_local_node_ids: list[str] = field(default_factory=list)
parallel_local_branch_node_ids: list[str] = field(default_factory=list)
synced_assets: list[SyncedAsset] = field(default_factory=list)
custom_nodes_bundle: SyncedAsset | None = None
artifact_finalizer_node_id: str | None = None
execution_assignments_by_representative: dict[str, ExecutionAssignment] = field(
default_factory=dict
)
execution_worker_indices_by_representative: dict[str, int] = field(
default_factory=dict
)
custom_nodes_bundles_by_environment: dict[str, SyncedAsset | None] = field(
default_factory=dict
)
remote_configurations: list[dict[str, Any]] = field(default_factory=list)
execution_locations_by_environment: dict[str, str] = field(
default_factory=dict
)
@dataclass
class ComponentExecutionPlan:
"""Hold assignments and prepared provider state for one prompt rewrite."""
assignments: dict[str, ExecutionAssignment]
configuration_set: RemoteConfigurationSet | None = None
configurations_by_id: dict[str, RemoteConfiguration] = field(
default_factory=dict
)
safe_configurations: list[dict[str, Any]] = field(default_factory=list)
ssh_hosts_by_id: dict[str, SshHostConfig] = field(default_factory=dict)
vast_service: VastService | None = None
vast_leases_by_environment: dict[str, Any] = field(default_factory=dict)
resolved_llm_profiles: dict[str, Any] = field(default_factory=dict)
@dataclass(frozen=True)
class _EnvironmentAssetPreparationResult:
"""Collect custom-node and prompt-asset results for one environment."""
environment_id: str
custom_nodes_bundle: SyncedAsset | None
component_prompts: dict[str, dict[str, Any]]
assets_by_component_id: dict[str, list[SyncedAsset]]
@dataclass(frozen=True)
class RemoteExpansionReason:
"""Describe why one upstream node had to join a remote component."""
node_id: str
class_type: str
required_by_node_id: str
required_by_class_type: str
output_index: int
io_type: str
@dataclass
class RemoteNodeAnalysis:
"""Structured dry-run result for context-menu remote expansion."""
requested_node_ids: list[str] = field(default_factory=list)
requested_workflow_node_paths: list[str] = field(default_factory=list)
current_remote_node_ids: list[str] = field(default_factory=list)
current_remote_workflow_node_paths: list[str] = field(default_factory=list)
resolved_remote_node_ids: list[str] = field(default_factory=list)
resolved_workflow_node_paths: list[str] = field(default_factory=list)
added_node_ids: list[str] = field(default_factory=list)
added_workflow_node_paths: list[str] = field(default_factory=list)
sandwiched_local_node_ids: list[str] = field(default_factory=list)
reasons: list[RemoteExpansionReason] = field(default_factory=list)
@dataclass(frozen=True)
class PromptGraphLink:
"""Describe one direct prompt dependency edge."""
source_node_id: str
source_output_index: int
target_node_id: str
target_input_name: str
source_class_type: str
target_class_type: str
class ModalPromptValidationError(ValueError):
"""Raised when a prompt cannot be executed with the current Modal transport."""
_MODAL_GPU_VRAM_GB: dict[str, float] = {
"T4": 16,
"L4": 24,
"A10": 24,
"L40S": 48,
"A100": 40,
"A100-40GB": 40,
"A100-80GB": 80,
"RTX-PRO-6000": 96,
"H100": 80,
"H100!": 80,
"H200": 141,
"B200": 180,
"B200+": 180,
"B300": 288,
}
# Modal public pricing verified 2026-08-22; CPU and memory charges are excluded.
_MODAL_GPU_COST_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,
}
_HBM_GPU_NAME_MARKERS = ("A100", "H100", "H200", "B200", "B300")
def _gpu_memory_kind(machine_type: str) -> str:
"""Return the best-known GPU memory technology for a machine label."""
normalized = machine_type.upper().replace(" ", "-")
return (
"HBM"
if any(marker in normalized for marker in _HBM_GPU_NAME_MARKERS)
else "VRAM"
)
def _hardware_payload(
*,
machine_type: str,
gpu_count: int,
gpu_memory_bytes_per_device: int,
gpu_memory_bytes_total: int,
ram_bytes: int = 0,
ram_available_bytes: int | None = None,
ram_capacity_label: str | None = None,
) -> dict[str, Any]:
"""Return compact credential-free target hardware metadata."""
payload: dict[str, Any] = {
"machine_type": machine_type or "Unknown GPU",
"gpu_count": max(0, gpu_count),
"gpu_memory_kind": _gpu_memory_kind(machine_type),
"gpu_memory_bytes_per_device": max(0, gpu_memory_bytes_per_device),
"gpu_memory_bytes_total": max(0, gpu_memory_bytes_total),
}
if ram_bytes > 0:
payload["ram_bytes"] = ram_bytes
if ram_available_bytes is not None and ram_available_bytes >= 0:
payload["ram_available_bytes"] = ram_available_bytes
if ram_capacity_label:
payload["ram_capacity_label"] = ram_capacity_label
return payload
def _capabilities_hardware_payload(
capabilities: EnvironmentCapabilities | None,
) -> dict[str, Any] | None:
"""Summarize a probed host's GPUs and system memory for the UI."""
if capabilities is None:
return None
gpu_names = tuple(dict.fromkeys(gpu.name for gpu in capabilities.gpus))
machine_type = " + ".join(gpu_names) if gpu_names else "CPU-only"
total_gpu_memory = sum(gpu.total_vram_bytes for gpu in capabilities.gpus)
per_device_memory = max(
(gpu.total_vram_bytes for gpu in capabilities.gpus),
default=0,
)
return _hardware_payload(
machine_type=machine_type,
gpu_count=len(capabilities.gpus),
gpu_memory_bytes_per_device=per_device_memory,
gpu_memory_bytes_total=total_gpu_memory,
ram_bytes=capabilities.total_ram_bytes,
ram_available_bytes=capabilities.available_ram_bytes,
)
def _modal_hardware_payload(gpu_type: str) -> dict[str, Any]:
"""Summarize known Modal GPU capacity without guessing dynamic host RAM."""
gpu_memory_bytes = int(_MODAL_GPU_VRAM_GB.get(gpu_type, 0.0) * 1024**3)
return _hardware_payload(
machine_type=gpu_type,
gpu_count=1,
gpu_memory_bytes_per_device=gpu_memory_bytes,
gpu_memory_bytes_total=gpu_memory_bytes,
ram_capacity_label="Provider managed",
)
def _vast_hardware_payload(resource: Any) -> dict[str, Any]:
"""Summarize a Vast offer or lease using its advertised capacities."""
gpu_count = max(
0,
int(
getattr(resource, "gpu_count", None)
or getattr(resource, "num_gpus", 0)
or 0
),
)
per_device_memory = max(
0,
int(getattr(resource, "gpu_ram_mb", 0) or 0) * 1024**2,
)
advertised_total_mb = int(
getattr(resource, "gpu_total_ram_mb", 0) or 0
)
total_gpu_memory = (
advertised_total_mb * 1024**2
if advertised_total_mb > 0
else per_device_memory * max(gpu_count, 1)
)
return _hardware_payload(
machine_type=str(
getattr(resource, "gpu_name", "Unknown GPU") or "Unknown GPU"
),
gpu_count=gpu_count,
gpu_memory_bytes_per_device=per_device_memory,
gpu_memory_bytes_total=total_gpu_memory,
ram_bytes=max(
0,
int(getattr(resource, "cpu_ram_mb", 0) or 0) * 1024**2,
),
)
def _modal_environment_state(settings: ModalSyncSettings) -> EnvironmentSchedulingState:
"""Return the scheduler-facing state of the selected Modal GPU target."""
modal_gpu = settings.modal_gpu
vram_bytes = int(_MODAL_GPU_VRAM_GB.get(modal_gpu, 0.0) * 1024**3)
capabilities = EnvironmentCapabilities(
architecture="x86_64",
operating_system="linux",
cpu_count=1,
total_ram_bytes=max(vram_bytes, 1),
available_ram_bytes=None,
available_disk_bytes=None,
docker_version="modal-managed",
docker_rootless=False,
nvidia_container_runtime=True,
gpus=(
GpuCapability(
uuid=f"modal-{modal_gpu.lower()}",
name=modal_gpu,
total_vram_bytes=vram_bytes,
),
),
)
return EnvironmentSchedulingState(
environment_id=f"modal:{modal_gpu}",
provider=ExecutionProvider.MODAL,
enabled=True,
health=EnvironmentHealth.READY,
cost_usd_per_second=_MODAL_GPU_COST_USD_PER_SECOND.get(modal_gpu),
capabilities=capabilities,
maximum_workers=settings.max_containers or 1,
)
def _modal_configuration_environment_state(
configuration: ModalRemoteConfiguration,
settings: ModalSyncSettings,
) -> EnvironmentSchedulingState:
"""Return one scheduler state for a workflow-declared Modal capacity pool."""
selected_settings = settings_for_modal_gpu(settings, configuration.gpu_type)
base_state = _modal_environment_state(selected_settings)
return replace(
base_state,
environment_id=(
f"modal:{configuration.configuration_id}:{configuration.gpu_type}"
),
configuration_id=configuration.configuration_id,
display_name=configuration.display_name,
maximum_workers=configuration.instance_count,
)
def _ssh_host_registry(settings: ModalSyncSettings) -> RemoteHostRegistry | None:
"""Return the persistent SSH host registry when a user directory exists."""
user_directory = discover_comfyui_user_directory(settings)
if user_directory is None:
return None
return RemoteHostRegistry.for_user_directory(user_directory)
def _configured_ssh_hosts(settings: ModalSyncSettings) -> tuple[SshHostConfig, ...]:
"""Return configured SSH hosts without probing during queue submission."""
registry = _ssh_host_registry(settings)
if registry is None:
return ()
return registry.load().hosts
def _schedulable_ssh_hosts(settings: ModalSyncSettings) -> tuple[SshHostConfig, ...]:
"""Probe every enabled host immediately before cost-aware scheduling."""
hosts = _configured_ssh_hosts(settings)
registry = _ssh_host_registry(settings)
if registry is None:
return hosts
probe_hosts = [host for host in hosts if host.enabled and not host.draining]
if not probe_hosts:
return hosts
with ThreadPoolExecutor(max_workers=min(8, len(probe_hosts))) as executor:
refreshed = {
host.environment_id: future.result()
for host, future in (
(host, executor.submit(_refresh_ssh_host, host, registry))
for host in probe_hosts
)
}
return tuple(refreshed.get(host.environment_id, host) for host in hosts)
def _refresh_ssh_host(
host: SshHostConfig,
registry: RemoteHostRegistry,
) -> SshHostConfig:
"""Probe and persist one host without aborting other candidates."""
try:
capabilities = SshDockerController(host).probe_capabilities()
return registry.update_probe_result(
host.environment_id,
capabilities=capabilities,
health=EnvironmentHealth.READY,
last_error=None,
)
except (OSError, RuntimeError, ValueError) as exc:
logger.warning(
"Remote environment probe failed environment=%s: %s",
host.environment_id,
exc,
)
try:
return registry.update_probe_result(
host.environment_id,
capabilities=host.capabilities,
health=EnvironmentHealth.UNAVAILABLE,
last_error=str(exc),
)
except (KeyError, OSError, ValueError):
return replace(
host,
health=EnvironmentHealth.UNAVAILABLE,
last_error=str(exc),
)
def _prompt_llm_model_references(prompt: Mapping[str, Any]) -> tuple[str, ...]:
"""Return every fixed Modal LLM model reference in a prompt."""
references: set[str] = set()
for prompt_node in prompt.values():
if not isinstance(prompt_node, Mapping):
continue
if str(prompt_node.get("class_type") or "") != "ModalLLM":
continue
inputs = prompt_node.get("inputs")
if not isinstance(inputs, Mapping):
continue
model_reference = inputs.get("model_profile")
if isinstance(model_reference, str) and model_reference.strip():
references.add(model_reference.strip())
return tuple(sorted(references))
def _resolve_prompt_llm_profiles(
prompt: Mapping[str, Any],
settings: ModalSyncSettings,
) -> dict[str, Any]:
"""Resolve LLM metadata before environment admission and cost ranking."""
from .llm_profiles import get_llm_profile
from .llm_resolver import resolve_model_profile
storage_root = Path(
getattr(settings, "local_storage_root", "/tmp/comfyui-modal-sync-storage")
)
profiles: dict[str, Any] = {}
for model_reference in _prompt_llm_model_references(prompt):
try:
profile = get_llm_profile(model_reference, storage_root=storage_root)
except ValueError as profile_error:
if model_reference.startswith("hf-"):
raise ModalPromptValidationError(str(profile_error)) from profile_error
try:
profile = resolve_model_profile(model_reference, storage_root).profile
except ValueError as resolution_error:
raise ModalPromptValidationError(
str(resolution_error)
) from resolution_error
profiles[model_reference] = profile
logger.info(
"Resolved planner LLM profile model=%s profile=%s weights_gib=%.2f "
"estimated_vram_gib=%.2f.",
model_reference,
profile.profile_id,
profile.artifact_bytes / 1024**3,
profile.estimated_vram_gb,
)
return profiles
def _component_profile_memory_estimate(
component: RemoteComponentPlan,
prompt: Mapping[str, Any],
preferences: WorkflowExecutionPreferences,
resolved_profiles: Mapping[str, Any],
) -> ComponentMemoryEstimate:
"""Estimate a component's RAM and VRAM floors from resolved LLM profiles."""
minimum_vram_bytes = preferences.minimum_vram_bytes
profiles: dict[str, Any] = {}
for node_id in component.node_ids:
prompt_node = prompt.get(node_id)
if not isinstance(prompt_node, Mapping):
continue
if str(prompt_node.get("class_type") or "") != "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
profile = resolved_profiles.get(model_profile.strip())
if profile is None:
continue
profiles[profile.profile_id] = profile
minimum_vram_bytes = max(
minimum_vram_bytes,
int(max(0.0, profile.estimated_vram_gb) * 1024**3),
)
artifact_sizes = [int(profile.artifact_bytes) for profile in profiles.values()]
return ComponentMemoryEstimate(
minimum_vram_bytes=minimum_vram_bytes,
minimum_ram_bytes=(sum(artifact_sizes) + _MODEL_RAM_HEADROOM_BYTES)
if artifact_sizes
else 0,
model_asset_count=len(artifact_sizes),
largest_model_bytes=max(artifact_sizes, default=0),
)
def _component_required_provider(
component: RemoteComponentPlan,
prompt: Mapping[str, Any],
resolved_profiles: Mapping[str, Any],
) -> ExecutionProvider | None:
"""Require SSH when a component uses an SSH-only resident backend."""
for node_id in component.node_ids:
prompt_node = prompt.get(node_id)
if not isinstance(prompt_node, Mapping):
continue
if str(prompt_node.get("class_type") or "") != "ModalLLM":
continue
inputs = prompt_node.get("inputs")
if not isinstance(inputs, Mapping):
continue
model_reference = inputs.get("model_profile")
if not isinstance(model_reference, str):
continue
profile = resolved_profiles.get(model_reference.strip())
if (
profile is not None
and getattr(profile, "backend", "") == "llama_cpp_server"
):
return ExecutionProvider.SSH_DOCKER
return None
def _prompt_node_required_provider(
prompt_node: Mapping[str, Any],
) -> ExecutionProvider | None:
"""Return the provider required by one curated node backend, when known."""
if str(prompt_node.get("class_type") or "") != "ModalLLM":
return None
inputs = prompt_node.get("inputs")
if not isinstance(inputs, Mapping):
return None
model_reference = inputs.get("model_profile")
if not isinstance(model_reference, str) or not model_reference.strip():
return None
from .llm_profiles import get_llm_profile
try:
profile = get_llm_profile(model_reference.strip())
except ValueError:
return None
if profile.backend == "llama_cpp_server":
return ExecutionProvider.SSH_DOCKER
return None
def _iter_prompt_string_values(value: object) -> Iterator[str]:
"""Yield nested prompt string values while ignoring graph links and scalars."""
if isinstance(value, str):
yield value
return
if isinstance(value, Mapping):
for child in value.values():
yield from _iter_prompt_string_values(child)
return
if isinstance(value, (list, tuple)) and not _is_link(value):
for child in value:
yield from _iter_prompt_string_values(child)
def _is_additive_model_node(class_type: str) -> bool:
"""Return whether a loader contributes weights alongside a primary model."""
normalized = class_type.casefold()
return any(token in normalized for token in _ADDITIVE_MODEL_CLASS_TOKENS)
def _component_model_asset_sizes(
component: RemoteComponentPlan,
prompt: Mapping[str, Any],
settings: ModalSyncSettings,
) -> tuple[tuple[int, ...], tuple[int, ...]]:
"""Return unique primary and additive model sizes referenced by a component."""
asset_roles: dict[Path, bool] = {}
for node_id in component.node_ids:
prompt_node = prompt.get(node_id)
if not isinstance(prompt_node, Mapping):
continue
inputs = prompt_node.get("inputs")
if not isinstance(inputs, Mapping):
continue
additive = _is_additive_model_node(str(prompt_node.get("class_type") or ""))
for value in _iter_prompt_string_values(inputs):
resolved_path = resolve_model_path(
value,
comfyui_root=getattr(settings, "comfyui_root", None),
extensions=MODEL_FILE_EXTENSIONS,
)
if resolved_path is None:
continue
asset_roles[resolved_path] = (
asset_roles.get(resolved_path, True) and additive
)
primary_sizes: list[int] = []
additive_sizes: list[int] = []
for asset_path, additive in asset_roles.items():
try:
size_bytes = asset_path.stat().st_size
except OSError as exc:
logger.warning(
"Unable to inspect model size for placement path=%s: %s",
asset_path,
exc,
)
continue
if size_bytes <= 0:
continue
target = additive_sizes if additive else primary_sizes
target.append(size_bytes)
return tuple(primary_sizes), tuple(additive_sizes)
def _component_memory_estimate(
component: RemoteComponentPlan,
prompt: Mapping[str, Any],
preferences: WorkflowExecutionPreferences,
settings: ModalSyncSettings,
resolved_llm_profiles: Mapping[str, Any] | None = None,
) -> ComponentMemoryEstimate:
"""Infer conservative RAM and VRAM floors from resident model weight sizes."""
profile_estimate = _component_profile_memory_estimate(
component,
prompt,
preferences,
resolved_llm_profiles
if resolved_llm_profiles is not None
else _resolve_prompt_llm_profiles(prompt, settings),
)
primary_sizes, additive_sizes = _component_model_asset_sizes(
component,
prompt,
settings,
)
all_sizes = primary_sizes + additive_sizes
if not all_sizes:
return profile_estimate
primary_peak_bytes = max(primary_sizes or all_sizes)
additive_bytes = sum(additive_sizes) if primary_sizes else 0
resident_weight_bytes = primary_peak_bytes + additive_bytes
model_vram_bytes = (
math.ceil(resident_weight_bytes * _MODEL_VRAM_WEIGHT_MULTIPLIER)
+ _MODEL_VRAM_HEADROOM_BYTES
)
return ComponentMemoryEstimate(
minimum_vram_bytes=max(
profile_estimate.minimum_vram_bytes,
model_vram_bytes,
),
minimum_ram_bytes=max(
profile_estimate.minimum_ram_bytes,
resident_weight_bytes + _MODEL_RAM_HEADROOM_BYTES,
),
model_asset_count=len(all_sizes) + profile_estimate.model_asset_count,
largest_model_bytes=max(
max(all_sizes),
profile_estimate.largest_model_bytes,
),
)
def _component_execution_signature(
component: RemoteComponentPlan,
prompt: Mapping[str, Any],
) -> str:
"""Return a stable workload signature for runtime-history estimates."""
cost_shaping_input_names = frozenset(
{
"batch_size",
"duration",
"frames",
"height",
"max_new_tokens",
"model_profile",
"num_frames",
"steps",
"width",
}
)
nodes: list[dict[str, Any]] = []
for node_id in sorted(component.node_ids):
prompt_node = prompt.get(node_id)
if not isinstance(prompt_node, Mapping):
continue
inputs = prompt_node.get("inputs")
input_mapping = inputs if isinstance(inputs, Mapping) else {}
nodes.append(
{
"class_type": str(prompt_node.get("class_type") or ""),
"inputs": {
name: input_mapping[name]
for name in sorted(cost_shaping_input_names & set(input_mapping))
if not _is_link(input_mapping[name])
},
}
)
serialized = json.dumps(nodes, separators=(",", ":"), sort_keys=True).encode(
"utf-8"
)
return hashlib.sha256(serialized).hexdigest()
def _execution_history(settings: ModalSyncSettings) -> ExecutionHistory | None:
"""Return local runtime history when the ComfyUI user directory is known."""
user_directory = discover_comfyui_user_directory(settings)
if user_directory is None:
return None
return ExecutionHistory.for_user_directory(user_directory)
def _maximum_capacity_state(
environment: EnvironmentSchedulingState,
) -> EnvironmentSchedulingState:
"""Return an optimistic state used only to identify reclaimable SSH capacity."""
capabilities = environment.capabilities
if environment.provider is not ExecutionProvider.SSH_DOCKER or capabilities is None:
return environment
return replace(
environment,
capabilities=replace(
capabilities,
available_ram_bytes=capabilities.total_ram_bytes,
gpus=tuple(
replace(gpu, free_vram_bytes=gpu.total_vram_bytes)
for gpu in capabilities.gpus
),
),
)
def _reprobe_reclaimed_ssh_host(
host: SshHostConfig,
settings: ModalSyncSettings,
) -> SshHostConfig:
"""Probe a reclaimed host and persist its new free-memory measurements."""
registry = _ssh_host_registry(settings)
if registry is not None:
return _refresh_ssh_host(host, registry)
try:
capabilities = SshDockerController(host).probe_capabilities()
except (OSError, RuntimeError, ValueError) as exc:
logger.warning(
"Remote environment re-probe failed after worker reclaim "
"environment=%s: %s",
host.environment_id,
exc,
)
return replace(
host,
health=EnvironmentHealth.UNAVAILABLE,
last_error=str(exc),
)
return replace(
host,
capabilities=capabilities,
health=EnvironmentHealth.READY,
last_error=None,
)
def _probe_workflow_ssh_configuration(
configuration: SshRemoteConfiguration,
) -> SshHostConfig:
"""Probe one workflow-declared SSH host without mutating the global registry."""
host = configuration.host
try:
capabilities = SshDockerController(host).probe_capabilities()
except (OSError, RuntimeError, ValueError) as exc:
logger.warning(
"Workflow SSH environment probe failed configuration=%s environment=%s: %s",
configuration.display_name,
host.environment_id,
exc,
)
return replace(
host,
capabilities=None,
health=EnvironmentHealth.UNAVAILABLE,
last_error=str(exc),
)
return replace(
host,
capabilities=capabilities,
health=EnvironmentHealth.READY,
last_error=None,
)
def _optional_scheduler_choice(
scheduler: CostAwareEnvironmentScheduler,
environments: list[EnvironmentSchedulingState],
requirements: ComponentResourceRequirements,
) -> tuple[
ExecutionAssignment | None,
NoCompatibleExecutionEnvironmentError | None,
]:
"""Return a scheduler choice and preserve a compatibility failure for fallback."""
try:
return scheduler.choose(environments, requirements), None
except NoCompatibleExecutionEnvironmentError as exc:
return None, exc
def _require_scheduler_choice(
assignment: ExecutionAssignment | None,
error: NoCompatibleExecutionEnvironmentError | None,
) -> ExecutionAssignment:
"""Return an existing scheduler choice or raise its compatibility detail."""
if assignment is not None:
return assignment
if error is not None:
raise error
raise RuntimeError("Scheduler failed without an incompatibility reason.")
def _remove_idle_ssh_workers_for_reclaim(host: SshHostConfig) -> tuple[str, ...]:
"""Recycle idle managed workers without disrupting fallback placement."""
try:
return SshDockerController(host).remove_idle_managed_workers()
except (OSError, RuntimeError, ValueError) as exc:
logger.warning(
"Could not reclaim managed SSH worker capacity environment=%s: %s",
host.environment_id,
exc,
)
return ()
def _reclaim_improves_assignment(
optimistic: ExecutionAssignment,
actual: ExecutionAssignment | None,
requirements: ComponentResourceRequirements,
) -> bool:
"""Return whether reclaim unlocks required or materially better placement."""
if actual is None:
return True
if optimistic.environment_id == actual.environment_id:
return False
preferred = requirements.preferred_environment_ids
if optimistic.environment_id in preferred:
if actual.environment_id not in preferred:
return True
return preferred.index(optimistic.environment_id) < preferred.index(
actual.environment_id
)
if optimistic.predicted_cost_usd is None:
return False
if actual.predicted_cost_usd is None:
return True
return optimistic.predicted_cost_usd < actual.predicted_cost_usd
def _choose_with_idle_ssh_worker_reclaim(
*,
scheduler: CostAwareEnvironmentScheduler,
environments: list[EnvironmentSchedulingState],
requirements: ComponentResourceRequirements,
ssh_hosts_by_id: dict[str, SshHostConfig],
settings: ModalSyncSettings,
) -> ExecutionAssignment:
"""Choose an environment, recycling a cheaper idle SSH worker when necessary."""
actual_assignment, actual_error = _optional_scheduler_choice(
scheduler,
environments,
requirements,
)
optimistic_assignment, _ = _optional_scheduler_choice(
scheduler,
[_maximum_capacity_state(environment) for environment in environments],
requirements,
)
if optimistic_assignment is None:
return _require_scheduler_choice(actual_assignment, actual_error)
if (
optimistic_assignment.provider is not ExecutionProvider.SSH_DOCKER
or not _reclaim_improves_assignment(
optimistic_assignment,
actual_assignment,
requirements,
)
):
return _require_scheduler_choice(actual_assignment, actual_error)
host = ssh_hosts_by_id.get(optimistic_assignment.environment_id)
if host is None:
return _require_scheduler_choice(actual_assignment, actual_error)
removed_workers = _remove_idle_ssh_workers_for_reclaim(host)
if not removed_workers:
return _require_scheduler_choice(actual_assignment, actual_error)
logger.info(
"Recycled idle managed SSH worker(s) environment=%s containers=%s so the "
"planner can re-evaluate full host capacity.",
host.environment_id,
removed_workers,
)
refreshed_host = _reprobe_reclaimed_ssh_host(host, settings)
ssh_hosts_by_id[host.environment_id] = refreshed_host
refreshed_environments = [
(
refreshed_host.scheduling_state()
if environment.environment_id == host.environment_id
else environment
)
for environment in environments
]
return scheduler.choose(refreshed_environments, requirements)
def _plan_component_execution_assignments(
*,
components: list[RemoteComponentPlan],
prompt: Mapping[str, Any],
workflow: Mapping[str, Any] | None,
settings: ModalSyncSettings,
status_callback: SetupStatusCallback | None = None,
resolved_llm_profiles: Mapping[str, Any] | None = None,
) -> dict[str, ExecutionAssignment]:
"""Assign components across Modal, SSH Docker, and workflow-declared Vast pools."""
preferences = WorkflowExecutionPreferences.from_workflow(workflow)
active_llm_profiles = (
dict(resolved_llm_profiles)
if resolved_llm_profiles is not None
else _resolve_prompt_llm_profiles(prompt, settings)
)
if preferences.policy is ExecutionPolicy.MODAL:
incompatible_component_ids = [
component.representative_node_id
for component in components
if _component_required_provider(
component,
prompt,
active_llm_profiles,
)
is ExecutionProvider.SSH_DOCKER
]
if incompatible_component_ids:
raise ModalPromptValidationError(
"Modal-only execution cannot run SSH-only component(s) "
f"{incompatible_component_ids}. Select Automatic or Self-hosted "
"execution for the workflow."
)
modal_state = _modal_environment_state(settings)
return {
component.representative_node_id: ExecutionAssignment(
environment_id=modal_state.environment_id,
provider=ExecutionProvider.MODAL,
predicted_cost_usd=None,
predicted_completion_seconds=0.0,
reasons=("workflow policy requires Modal",),
)
for component in components
}
try:
vast_profiles = extract_vast_profiles(prompt)
except ValueError as exc:
raise ModalPromptValidationError(str(exc)) from exc
vast_service: VastService | None = None
if preferences.policy in {ExecutionPolicy.VAST, ExecutionPolicy.AUTOMATIC}:
if preferences.policy is ExecutionPolicy.VAST and not vast_profiles:
raise ModalPromptValidationError(
"Vast.ai-only execution requires at least one disconnected "
"Vast.ai Lease Configuration node in the workflow."
)
if vast_profiles:
try:
vast_service = VastService.from_environment(
settings,
repo_root=Path(__file__).resolve().parent,
)
except (OSError, RuntimeError, ValueError) as exc:
if preferences.policy is ExecutionPolicy.VAST:
raise ModalPromptValidationError(str(exc)) from exc
logger.warning("Skipping Vast.ai automatic placement: %s", exc)
ssh_hosts = (
_schedulable_ssh_hosts(settings)
if preferences.policy in {ExecutionPolicy.SELF_HOSTED, ExecutionPolicy.AUTOMATIC}
else ()
)
ssh_hosts_by_id = {host.environment_id: host for host in ssh_hosts}
modal_state = (
_modal_environment_state(settings)
if preferences.policy is ExecutionPolicy.AUTOMATIC
else None
)
if not ssh_hosts and modal_state is None and vast_service is None:
raise ModalPromptValidationError(
"This workflow requests remote execution, but no compatible provider is configured."
)
scheduler = CostAwareEnvironmentScheduler()
history = _execution_history(settings)
component_signatures: dict[str, str] = {}
component_memory_estimates: dict[str, ComponentMemoryEstimate] = {}
component_required_providers: dict[str, ExecutionProvider | None] = {}
for component in components:
component_id = component.representative_node_id
component_signatures[component_id] = _component_execution_signature(
component,
prompt,
)
memory_estimate = _component_memory_estimate(
component,
prompt,
preferences,
settings,
active_llm_profiles,
)
component_memory_estimates[component_id] = memory_estimate
component_required_providers[component_id] = _component_required_provider(
component,
prompt,
active_llm_profiles,
)
if memory_estimate.model_asset_count:
logger.info(
"Estimated remote component=%s memory floor vram_gib=%.2f ram_gib=%.2f "
"from model_assets=%d largest_model_gib=%.2f.",
component_id,
memory_estimate.minimum_vram_bytes / 1024**3,
memory_estimate.minimum_ram_bytes / 1024**3,
memory_estimate.model_asset_count,
memory_estimate.largest_model_bytes / 1024**3,
)
if vast_service is not None:
try:
vast_service.prefetch_offers_sync(
vast_profiles,
tuple(
VastSearchRequirements(
minimum_vram_bytes=memory_estimate.minimum_vram_bytes,
minimum_ram_bytes=memory_estimate.minimum_ram_bytes,
)
for memory_estimate in component_memory_estimates.values()
),
)
except (OSError, RuntimeError, ValueError) as exc:
if preferences.policy is ExecutionPolicy.VAST:
raise ModalPromptValidationError(str(exc)) from exc
logger.warning(
"Unable to prefetch Vast.ai marketplace offers: %s",
exc,
)
assignments: dict[str, ExecutionAssignment] = {}
for component in components:
component_id = component.representative_node_id
environments = [
host.scheduling_state() for host in ssh_hosts_by_id.values()
]
if modal_state is not None:
environments.append(modal_state)
component_signature = component_signatures[component_id]
memory_estimate = component_memory_estimates[component_id]
vast_quote: VastProfileQuote | None = None
if vast_service is not None:
try:
vast_quote = vast_service.quote_best_profile_sync(
vast_profiles,
minimum_vram_bytes=memory_estimate.minimum_vram_bytes,
minimum_ram_bytes=memory_estimate.minimum_ram_bytes,
predicted_execution_seconds=60.0,
)
except (OSError, RuntimeError, ValueError) as exc:
if preferences.policy is ExecutionPolicy.VAST:
raise ModalPromptValidationError(str(exc)) from exc
logger.warning(
"No Vast.ai automatic candidate for component=%s: %s",
component.representative_node_id,
exc,
)
else:
environments.append(vast_service.scheduling_state(vast_quote))
environment_ids = [
environment.environment_id for environment in environments
]
historical_estimates = (
history.estimates(component_signature, environment_ids)
if history is not None
else {}
)
component_required_provider = component_required_providers[component_id]
if preferences.policy is ExecutionPolicy.VAST:
if component_required_provider is ExecutionProvider.SSH_DOCKER:
raise ModalPromptValidationError(
"Vast.ai-only execution cannot run an SSH-only component. "
"Select Automatic or Self-hosted execution for this workflow."
)
component_required_provider = ExecutionProvider.VAST
requirements = ComponentResourceRequirements(
minimum_vram_bytes=memory_estimate.minimum_vram_bytes,
minimum_ram_bytes=memory_estimate.minimum_ram_bytes,
gpu_required=True,
architecture="x86_64",
estimated_execution_seconds=60.0,
estimated_execution_seconds_by_environment={
environment_id: estimate.execution_seconds
for environment_id, estimate in historical_estimates.items()
},
preferred_environment_ids=preferences.preferred_environment_ids,
required_provider=component_required_provider,
)
try:
assignment = _choose_with_idle_ssh_worker_reclaim(
scheduler=scheduler,
environments=environments,
requirements=requirements,
ssh_hosts_by_id=ssh_hosts_by_id,
settings=settings,
)
except NoCompatibleExecutionEnvironmentError as exc:
raise ModalPromptValidationError(str(exc)) from exc
if assignment.provider is ExecutionProvider.VAST:
if vast_service is None or vast_quote is None:
raise ModalPromptValidationError(
"Vast.ai was selected without a current marketplace quote."
)
try:
if status_callback is None:
lease = vast_service.acquire_sync(vast_quote)
else:
lease = vast_service.acquire_sync(
vast_quote,
status_callback=lambda message: status_callback(
message,
None,
None,
),
)
except (OSError, RuntimeError, ValueError) as exc:
raise ModalPromptValidationError(
f"Unable to acquire Vast.ai capacity: {exc}"
) from exc
assignment = replace(
assignment,
environment_id=lease.environment_id,
predicted_cost_usd=vast_quote.predicted_incremental_cost_usd,
reasons=assignment.reasons
+ (
f"Vast profile {vast_quote.profile.profile_name}",
f"idle retention {lease.idle_retention_seconds / 3600:.1f}h",
),
)
assignments[component.representative_node_id] = assignment
logger.info(
"Assigned remote component=%s provider=%s environment=%s predicted_cost_usd=%s reasons=%s.",
component.representative_node_id,
assignment.provider.value,
assignment.environment_id,
assignment.predicted_cost_usd,
assignment.reasons,
)
return assignments
def _plan_component_execution(
*,
components: list[RemoteComponentPlan],
prompt: Mapping[str, Any],
workflow: Mapping[str, Any] | None,
settings: ModalSyncSettings,
status_callback: SetupStatusCallback | None = None,
environment_status_callback: EnvironmentSetupStatusCallback | None = None,
plan_callback: ExecutionPlanStatusCallback | None = None,
occupied_environment_ids: frozenset[str] = frozenset(),
) -> ComponentExecutionPlan:
"""Plan through connected configurations or synthesize the legacy behavior."""
try:
configuration_set = compile_remote_configuration_set(prompt)
except (TypeError, ValueError) as exc:
raise ModalPromptValidationError(str(exc)) from exc
resolved_llm_profiles = _resolve_prompt_llm_profiles(prompt, settings)
if configuration_set is None:
return ComponentExecutionPlan(
assignments=_plan_component_execution_assignments(
components=components,
prompt=prompt,
workflow=workflow,
settings=settings,
status_callback=status_callback,
resolved_llm_profiles=resolved_llm_profiles,
),
resolved_llm_profiles=resolved_llm_profiles,
)
return _plan_configured_component_execution(
components=components,
prompt=prompt,
workflow=workflow,
settings=settings,
configuration_set=configuration_set,
status_callback=status_callback,
environment_status_callback=environment_status_callback,
plan_callback=plan_callback,
occupied_environment_ids=occupied_environment_ids,
resolved_llm_profiles=resolved_llm_profiles,
)
def _plan_configured_component_execution(
*,
components: list[RemoteComponentPlan],
prompt: Mapping[str, Any],
workflow: Mapping[str, Any] | None,
settings: ModalSyncSettings,
configuration_set: RemoteConfigurationSet,
status_callback: SetupStatusCallback | None = None,
environment_status_callback: EnvironmentSetupStatusCallback | None = None,
plan_callback: ExecutionPlanStatusCallback | None = None,
occupied_environment_ids: frozenset[str] = frozenset(),
resolved_llm_profiles: Mapping[str, Any] | None = None,
) -> ComponentExecutionPlan:
"""Resolve and prepare a capacity-aware plan from connected configurations."""
configurations_by_id = {
configuration.configuration_id: configuration
for configuration in configuration_set.capacity_configurations
}
ssh_hosts_by_id = _probe_configured_ssh_hosts(configuration_set)
vast_service, vast_unavailable_reason = _configured_vast_service(
configuration_set,
settings,
)
requirements_by_component = _configured_component_requirements(
components=components,
prompt=prompt,
workflow=workflow,
settings=settings,
resolved_llm_profiles=resolved_llm_profiles,
)
_prefetch_configured_vast_offers(
configuration_set=configuration_set,
requirements_by_component=requirements_by_component,
vast_service=vast_service,
)
environments_by_component, vast_quotes = _configured_candidate_environments(
configuration_set=configuration_set,
requirements_by_component=requirements_by_component,
settings=settings,
ssh_hosts_by_id=ssh_hosts_by_id,
vast_service=vast_service,
vast_unavailable_reason=vast_unavailable_reason,
occupied_environment_ids=occupied_environment_ids,
)
if _reclaim_idle_configured_ssh_capacity(
configuration_set=configuration_set,
requirements_by_component=requirements_by_component,
ssh_hosts_by_id=ssh_hosts_by_id,
occupied_environment_ids=occupied_environment_ids,
):
environments_by_component, vast_quotes = _configured_candidate_environments(
configuration_set=configuration_set,
requirements_by_component=requirements_by_component,
settings=settings,
ssh_hosts_by_id=ssh_hosts_by_id,
vast_service=vast_service,
vast_unavailable_reason=vast_unavailable_reason,
occupied_environment_ids=occupied_environment_ids,
)
_apply_historical_execution_estimates(
components=components,
environments_by_component=environments_by_component,
requirements_by_component=requirements_by_component,
prompt=prompt,
settings=settings,
)
execution_stages = _component_execution_stages(
dict(prompt),
{
component.representative_node_id: set(component.node_ids)
for component in components
},
)
try:
assignments = CostAwareEnvironmentScheduler().plan(
execution_stages=execution_stages,
environments_by_component=environments_by_component,
requirements_by_component=requirements_by_component,
)
except NoCompatibleExecutionEnvironmentError as exc:
raise ModalPromptValidationError(str(exc)) from exc
for component_id, assignment in tuple(assignments.items()):
if assignment.environment_id not in occupied_environment_ids:
continue
assignments[component_id] = replace(
assignment,
reasons=assignment.reasons
+ ("capacity available after the earlier workflow completes",),
)
safe_configurations = _safe_remote_configuration_payload(configuration_set)
if plan_callback is not None:
plan_callback(
_planned_execution_assignments_payload(
assignments,
components,
configurations_by_id=configurations_by_id,
ssh_hosts_by_id=ssh_hosts_by_id,
vast_quotes=vast_quotes,
),
safe_configurations,
)
vast_leases = _prepare_selected_vast_capacity(
assignments=assignments,
configuration_set=configuration_set,
requirements_by_component=requirements_by_component,
vast_quotes=vast_quotes,
vast_service=vast_service,
status_callback=status_callback,
environment_status_callback=environment_status_callback,
)
if plan_callback is not None and vast_leases:
plan_callback(
_planned_execution_assignments_payload(
assignments,
components,
configurations_by_id=configurations_by_id,
ssh_hosts_by_id=ssh_hosts_by_id,
vast_quotes=vast_quotes,
vast_leases_by_environment=vast_leases,
),
safe_configurations,
)
return ComponentExecutionPlan(
assignments=assignments,
configuration_set=configuration_set,
configurations_by_id=configurations_by_id,
safe_configurations=safe_configurations,
ssh_hosts_by_id=ssh_hosts_by_id,
vast_service=vast_service,
vast_leases_by_environment=vast_leases,
resolved_llm_profiles=dict(resolved_llm_profiles or {}),
)
def _safe_remote_configuration_payload(
configuration_set: RemoteConfigurationSet,
) -> list[dict[str, Any]]:
"""Return safe configuration metadata enriched with best-effort storage usage."""
payload = configuration_set.to_safe_list()
payload_by_id = {
str(configuration.get("configuration_id") or ""): configuration
for configuration in payload
}
for storage in configuration_set.storage_configurations:
if not isinstance(storage, R2StorageBackingConfiguration):
continue
safe_storage = payload_by_id.get(storage.configuration_id)
if safe_storage is None:
continue
usage, error_code = _cached_r2_storage_usage(storage)
if error_code is not None:
safe_storage["credential_error_code"] = error_code
if usage is None:
continue
safe_storage["storage_usage_bytes"] = usage.size_bytes
safe_storage["storage_object_count"] = usage.object_count
return payload
def _cached_r2_storage_usage(
storage: R2StorageBackingConfiguration,
) -> tuple[R2StorageUsage | None, str | None]:
"""Return cached bucket usage and an optional credential recovery code."""
cache_key = (storage.account_id, storage.bucket, storage.jurisdiction)
now = time.monotonic()
with _R2_STORAGE_USAGE_CACHE_LOCK:
cached = _R2_STORAGE_USAGE_CACHE.get(cache_key)
if cached is not None and now - cached[0] < _R2_STORAGE_USAGE_CACHE_SECONDS:
return cached[1], None
try:
usage = _refresh_r2_storage_usage(storage)
except R2CredentialError as exc:
logger.warning(
"Unable to read R2 storage usage for configuration=%s bucket=%s: %s",
storage.configuration_id,
storage.bucket,
exc,
)
return None, exc.code
except (R2CacheError, RuntimeError, ValueError) as exc:
logger.warning(
"Unable to read R2 storage usage for configuration=%s bucket=%s: %s",
storage.configuration_id,
storage.bucket,
exc,
)
return None, None
return usage, None
def _refresh_r2_storage_usage(
storage: R2StorageBackingConfiguration,
) -> R2StorageUsage:
"""Query current R2 bucket usage and replace the short-lived cache entry."""
configuration = R2CredentialStore().cache_configuration(storage)
usage = R2CacheClient(configuration).storage_usage()
cache_key = (storage.account_id, storage.bucket, storage.jurisdiction)
with _R2_STORAGE_USAGE_CACHE_LOCK:
_R2_STORAGE_USAGE_CACHE[cache_key] = (time.monotonic(), usage)
return usage
def _r2_storage_from_usage_payload(
payload: Mapping[str, Any],
) -> R2StorageBackingConfiguration:
"""Build a validated R2 reference from a same-origin usage refresh request."""
configuration_id = str(
payload.get("configuration_id") or "r2-storage-refresh"
).strip()
return R2StorageBackingConfiguration(
configuration_id=configuration_id,
display_name=str(payload.get("display_name") or "R2 storage").strip(),
account_id=str(payload.get("account_id") or "").strip(),
bucket=str(payload.get("bucket") or "").strip(),
credential_id=str(payload.get("credential_id") or "").strip(),
jurisdiction=str(payload.get("jurisdiction") or "default")
.strip()
.casefold(),
key_prefix=str(
payload.get("key_prefix") or "comfy-modal-cache/v1/blobs/sha256"
).strip(),
write_back_mode=str(payload.get("write_back_mode") or "async")
.strip()
.casefold(),
)
def _planned_execution_assignments_payload(
assignments: Mapping[str, ExecutionAssignment],
components: Sequence[RemoteComponentPlan],
*,
configurations_by_id: Mapping[str, Any] | None = None,
ssh_hosts_by_id: Mapping[str, SshHostConfig] | None = None,
vast_quotes: Mapping[tuple[str, str], VastProfileQuote] | None = None,
vast_leases_by_environment: Mapping[str, Any] | None = None,
) -> dict[str, dict[str, Any]]:
"""Serialize scheduler choices before provider capacity is acquired."""
component_nodes = {
component.representative_node_id: list(component.node_ids)
for component in components
}
return {
component_id: {
"provider": assignment.provider.value,
"environment_id": assignment.environment_id,
"configuration_id": assignment.configuration_id,
"node_ids": component_nodes.get(component_id, [component_id]),
"predicted_cost_usd": assignment.predicted_cost_usd,
"predicted_completion_seconds": assignment.predicted_completion_seconds,
"worker_index": assignment.capacity_slot_index,
"reasons": list(assignment.reasons),
"hardware": _assignment_hardware_payload(
component_id=component_id,
assignment=assignment,
configurations_by_id=configurations_by_id or {},
ssh_hosts_by_id=ssh_hosts_by_id or {},
vast_quotes=vast_quotes or {},
vast_leases_by_environment=vast_leases_by_environment or {},
),
}
for component_id, assignment in sorted(assignments.items())
}
def _configuration_field(configuration: Any, field_name: str) -> Any:
"""Read one field from a configuration model or safe mapping."""
if isinstance(configuration, Mapping):
return configuration.get(field_name)
return getattr(configuration, field_name, None)
def _configuration_host(configuration: Any) -> SshHostConfig | None:
"""Return a workflow SSH host from a model or safe configuration mapping."""
host = _configuration_field(configuration, "host")
if isinstance(host, SshHostConfig):
return host
if isinstance(host, Mapping):
try:
return SshHostConfig.from_dict(host)
except (TypeError, ValueError):
return None
return None
def _assignment_hardware_payload(
*,
component_id: str,
assignment: ExecutionAssignment,
configurations_by_id: Mapping[str, Any],
ssh_hosts_by_id: Mapping[str, SshHostConfig],
vast_quotes: Mapping[tuple[str, str], VastProfileQuote],
vast_leases_by_environment: Mapping[str, Any],
) -> dict[str, Any] | None:
"""Return the best hardware identity known for one planned assignment."""
configuration_id = str(assignment.configuration_id or "")
configuration = configurations_by_id.get(configuration_id)
if assignment.provider is ExecutionProvider.MODAL:
gpu_type = str(
_configuration_field(configuration, "gpu_type")
or assignment.environment_id.rsplit(":", 1)[-1]
)
return _modal_hardware_payload(gpu_type)
if assignment.provider is ExecutionProvider.SSH_DOCKER:
host = ssh_hosts_by_id.get(configuration_id) or _configuration_host(
configuration
)
return _capabilities_hardware_payload(
host.capabilities if host is not None else None
)
lease = vast_leases_by_environment.get(assignment.environment_id)
if lease is not None:
return _vast_hardware_payload(lease)
quote = vast_quotes.get((component_id, configuration_id))
if quote is None:
return None
resource = getattr(quote, "existing_lease", None) or getattr(
quote,
"offer",
None,
)
return _vast_hardware_payload(resource) if resource is not None else None
def _probe_configured_ssh_hosts(
configuration_set: RemoteConfigurationSet,
) -> dict[str, SshHostConfig]:
"""Probe all workflow-declared SSH hosts concurrently."""
ssh_configurations = [
configuration
for configuration in configuration_set.configurations
if isinstance(configuration, SshRemoteConfiguration)
]
if not ssh_configurations:
return {}
with ThreadPoolExecutor(max_workers=min(8, len(ssh_configurations))) as executor:
probed_hosts = {
configuration.configuration_id: future.result()
for configuration, future in (
(
configuration,
executor.submit(
_probe_workflow_ssh_configuration,
configuration,
),
)
for configuration in ssh_configurations
)
}
return probed_hosts
def _configured_vast_service(
configuration_set: RemoteConfigurationSet,
settings: ModalSyncSettings,
) -> tuple[VastService | None, str | None]:
"""Construct one Vast service only when the connected graph requests it."""
vast_configured = any(
isinstance(configuration, VastRemoteConfiguration)
for configuration in configuration_set.capacity_configurations
)
if not vast_configured:
return None, None
try:
return (
VastService.from_environment(
settings,
repo_root=Path(__file__).resolve().parent,
),
None,
)
except (OSError, RuntimeError, ValueError) as exc:
has_fallback_provider = any(
not isinstance(configuration, VastRemoteConfiguration)
for configuration in configuration_set.capacity_configurations
)
if not has_fallback_provider:
raise ModalPromptValidationError(str(exc)) from exc
logger.warning(
"Skipping unavailable workflow Vast.ai capacity because another "
"configured provider can be planned: %s",
exc,
)
return None, str(exc)
def _configured_component_requirements(
*,
components: list[RemoteComponentPlan],
prompt: Mapping[str, Any],
workflow: Mapping[str, Any] | None,
settings: ModalSyncSettings,
resolved_llm_profiles: Mapping[str, Any] | None = None,
) -> dict[str, ComponentResourceRequirements]:
"""Build provider-neutral requirements for every remote component."""
preferences = WorkflowExecutionPreferences.from_workflow(workflow)
active_llm_profiles = (
dict(resolved_llm_profiles)
if resolved_llm_profiles is not None
else _resolve_prompt_llm_profiles(prompt, settings)
)
requirements: dict[str, ComponentResourceRequirements] = {}
for component in components:
memory_estimate = _component_memory_estimate(
component,
prompt,
preferences,
settings,
active_llm_profiles,
)
component_id = component.representative_node_id
requirements[component_id] = ComponentResourceRequirements(
minimum_vram_bytes=memory_estimate.minimum_vram_bytes,
minimum_ram_bytes=memory_estimate.minimum_ram_bytes,
gpu_required=True,
architecture="x86_64",
estimated_execution_seconds=60.0,
preferred_environment_ids=preferences.preferred_environment_ids,
required_provider=_component_required_provider(
component,
prompt,
active_llm_profiles,
),
)
return requirements
def _configured_candidate_environments(
*,
configuration_set: RemoteConfigurationSet,
requirements_by_component: Mapping[str, ComponentResourceRequirements],
settings: ModalSyncSettings,
ssh_hosts_by_id: Mapping[str, SshHostConfig],
vast_service: VastService | None,
vast_unavailable_reason: str | None,
occupied_environment_ids: frozenset[str] = frozenset(),
) -> tuple[
dict[str, list[EnvironmentSchedulingState]],
dict[tuple[str, str], VastProfileQuote],
]:
"""Resolve candidate pool states for every component without acquiring capacity."""
candidates: dict[str, list[EnvironmentSchedulingState]] = {
component_id: [] for component_id in requirements_by_component
}
vast_quotes: dict[tuple[str, str], VastProfileQuote] = {}
for component_id, requirements in requirements_by_component.items():
for configuration in configuration_set.capacity_configurations:
state, quote = _configured_candidate_environment(
configuration=configuration,
requirements=requirements,
settings=settings,
ssh_hosts_by_id=ssh_hosts_by_id,
vast_service=vast_service,
vast_unavailable_reason=vast_unavailable_reason,
occupied_environment_ids=occupied_environment_ids,
)
if state is not None:
candidates[component_id].append(state)
if quote is not None:
vast_quotes[(component_id, configuration.configuration_id)] = quote
return candidates, vast_quotes
def _reclaim_idle_configured_ssh_capacity(
*,
configuration_set: RemoteConfigurationSet,
requirements_by_component: Mapping[str, ComponentResourceRequirements],
ssh_hosts_by_id: dict[str, SshHostConfig],
occupied_environment_ids: frozenset[str],
) -> bool:
"""Recycle idle configured workers when only their cached memory blocks work."""
scheduler = CostAwareEnvironmentScheduler()
reclaimed = False
for configuration in configuration_set.capacity_configurations:
if not isinstance(configuration, SshRemoteConfiguration):
continue
host = ssh_hosts_by_id[configuration.configuration_id]
if host.environment_id in occupied_environment_ids:
continue
state = replace(
host.scheduling_state(),
configuration_id=configuration.configuration_id,
display_name=configuration.display_name,
)
optimistic_state = _maximum_capacity_state(state)
reclaimable = any(
_optional_scheduler_choice(scheduler, [state], requirements)[0] is None
and _optional_scheduler_choice(
scheduler,
[optimistic_state],
requirements,
)[0]
is not None
for requirements in requirements_by_component.values()
)
if not reclaimable:
continue
removed_workers = _remove_idle_ssh_workers_for_reclaim(host)
if not removed_workers:
continue
refreshed_host = _probe_workflow_ssh_configuration(configuration)
ssh_hosts_by_id[configuration.configuration_id] = refreshed_host
reclaimed = True
logger.info(
"Recycled idle workflow-configured SSH worker(s) environment=%s "
"containers=%s and refreshed available memory.",
host.environment_id,
removed_workers,
)
return reclaimed
def _prefetch_configured_vast_offers(
*,
configuration_set: RemoteConfigurationSet,
requirements_by_component: Mapping[str, ComponentResourceRequirements],
vast_service: VastService | None,
) -> None:
"""Warm distinct Vast searches in parallel before deterministic planning."""
if vast_service is None:
return
profiles = [
configuration.profile
for configuration in configuration_set.configurations
if isinstance(configuration, VastRemoteConfiguration)
]
prefetch = getattr(vast_service, "prefetch_offers_sync", None)
if not callable(prefetch):
return
try:
prefetch(
profiles,
tuple(
VastSearchRequirements(
minimum_vram_bytes=requirements.minimum_vram_bytes,
minimum_ram_bytes=requirements.minimum_ram_bytes,
)
for requirements in requirements_by_component.values()
),
)
except (OSError, RuntimeError, ValueError) as exc:
logger.warning(
"Unable to prefetch configured Vast.ai marketplace offers: %s",
exc,
)
def _configured_candidate_environment(
*,
configuration: RemoteConfiguration,
requirements: ComponentResourceRequirements,
settings: ModalSyncSettings,
ssh_hosts_by_id: Mapping[str, SshHostConfig],
vast_service: VastService | None,
vast_unavailable_reason: str | None,
occupied_environment_ids: frozenset[str] = frozenset(),
) -> tuple[EnvironmentSchedulingState | None, VastProfileQuote | None]:
"""Resolve one configuration as a candidate for one component."""
if isinstance(configuration, ModalRemoteConfiguration):
return _modal_configuration_environment_state(configuration, settings), None
if isinstance(configuration, SshRemoteConfiguration):
host = ssh_hosts_by_id[configuration.configuration_id]
state = replace(
host.scheduling_state(),
configuration_id=configuration.configuration_id,
display_name=configuration.display_name,
)
if state.environment_id in occupied_environment_ids:
state = _maximum_capacity_state(state)
logger.info(
"Planning queued work against reclaimable SSH capacity "
"environment=%s because an earlier workflow owns that host.",
state.environment_id,
)
return state, None
if not isinstance(configuration, VastRemoteConfiguration):
raise TypeError(
f"Unsupported remote configuration type {type(configuration).__name__}."
)
if vast_service is None:
return EnvironmentSchedulingState(
environment_id=f"vast:{configuration.configuration_id}",
provider=ExecutionProvider.VAST,
enabled=True,
health=EnvironmentHealth.UNAVAILABLE,
cost_usd_per_second=None,
capabilities=None,
configuration_id=configuration.configuration_id,
display_name=configuration.display_name,
unavailable_reason=(
vast_unavailable_reason
or "Vast.ai controller is unavailable."
),
maximum_workers=configuration.capacity_limit,
), None
try:
quote = vast_service.quote_best_profile_sync(
[configuration.profile],
minimum_vram_bytes=requirements.minimum_vram_bytes,
minimum_ram_bytes=requirements.minimum_ram_bytes,
predicted_execution_seconds=requirements.estimated_execution_seconds,
)
except (OSError, RuntimeError, ValueError) as exc:
logger.warning(
"Vast configuration has no candidate configuration=%s: %s",
configuration.display_name,
exc,
)
return EnvironmentSchedulingState(
environment_id=f"vast:{configuration.configuration_id}",
provider=ExecutionProvider.VAST,
enabled=True,
health=EnvironmentHealth.UNAVAILABLE,
cost_usd_per_second=None,
capabilities=None,
configuration_id=configuration.configuration_id,
display_name=configuration.display_name,
unavailable_reason=str(exc),
maximum_workers=configuration.capacity_limit,
), None
state = vast_service.scheduling_state(quote)
return replace(
state,
environment_id=f"vast:{configuration.configuration_id}",
configuration_id=configuration.configuration_id,
display_name=configuration.display_name,
maximum_workers=configuration.capacity_limit,
), quote
def _apply_historical_execution_estimates(
*,
components: list[RemoteComponentPlan],
environments_by_component: Mapping[str, list[EnvironmentSchedulingState]],
requirements_by_component: dict[str, ComponentResourceRequirements],
prompt: Mapping[str, Any],
settings: ModalSyncSettings,
) -> None:
"""Add environment-specific historical timing estimates in place."""
history = _execution_history(settings)
if history is None:
return
for component in components:
component_id = component.representative_node_id
environment_ids = [
environment.environment_id
for environment in environments_by_component.get(component_id, [])
]
estimates = history.estimates(
_component_execution_signature(component, prompt),
environment_ids,
)
requirements_by_component[component_id] = replace(
requirements_by_component[component_id],
estimated_execution_seconds_by_environment={
environment_id: estimate.execution_seconds
for environment_id, estimate in estimates.items()
},
)
def _prepare_selected_vast_capacity(
*,
assignments: dict[str, ExecutionAssignment],
configuration_set: RemoteConfigurationSet,
requirements_by_component: Mapping[str, ComponentResourceRequirements],
vast_quotes: Mapping[tuple[str, str], VastProfileQuote],
vast_service: VastService | None,
status_callback: SetupStatusCallback | None = None,
environment_status_callback: EnvironmentSetupStatusCallback | None = None,
) -> dict[str, Any]:
"""Acquire only Vast slots selected by the completed provider-neutral plan."""
selected_slots: dict[tuple[str, int], list[str]] = defaultdict(list)
for component_id, assignment in assignments.items():
if assignment.provider is not ExecutionProvider.VAST:
continue
if assignment.configuration_id is None:
raise ModalPromptValidationError(
"Vast assignment is missing its configuration identity."
)
selected_slots[
(assignment.configuration_id, assignment.capacity_slot_index)
].append(component_id)
if not selected_slots:
return {}
if vast_service is None:
raise ModalPromptValidationError(
"Vast capacity was selected without an initialized Vast service."
)
configurations_by_id = {
configuration.configuration_id: configuration
for configuration in configuration_set.configurations
if isinstance(configuration, VastRemoteConfiguration)
}
leases_by_environment: dict[str, Any] = {}
ordered_slots = sorted(selected_slots.items())
total_slots = len(ordered_slots)
for slot_number, ((configuration_id, slot_index), component_ids) in enumerate(
ordered_slots,
start=1,
):
configuration = configurations_by_id[configuration_id]
planned_environment_id = assignments[component_ids[0]].environment_id
quote = _quote_selected_vast_slot(
configuration=configuration,
component_ids=component_ids,
requirements_by_component=requirements_by_component,
vast_quotes=vast_quotes,
vast_service=vast_service,
)
try:
_emit_vast_capacity_status(
environment_id=planned_environment_id,
message=f"Acquiring Vast.ai capacity {slot_number} of {total_slots}",
current=slot_number - 1,
total=total_slots,
status_callback=status_callback,
environment_status_callback=environment_status_callback,
)
if status_callback is None and environment_status_callback is None:
lease = vast_service.acquire_sync(quote, slot=slot_index)
else:
lease = vast_service.acquire_sync(
quote,
slot=slot_index,
status_callback=(
lambda message, environment_id=planned_environment_id,
current=slot_number - 1: _emit_vast_capacity_status(
environment_id=environment_id,
message=message,
current=current,
total=total_slots,
status_callback=status_callback,
environment_status_callback=environment_status_callback,
)
),
)
except SyncCancelledError:
raise
except (OSError, RuntimeError, ValueError) as exc:
raise ModalPromptValidationError(
f"Unable to acquire Vast.ai capacity for configuration "
f"{configuration.display_name!r} slot {slot_index}: {exc}"
) from exc
leases_by_environment[lease.environment_id] = lease
cost_share = quote.predicted_incremental_cost_usd / len(component_ids)
for component_id in component_ids:
assignment = assignments[component_id]
assignments[component_id] = replace(
assignment,
environment_id=lease.environment_id,
predicted_cost_usd=cost_share,
reasons=assignment.reasons
+ (
f"Vast profile {configuration.display_name}",
f"Vast capacity slot {slot_index}",
),
)
_emit_vast_capacity_status(
environment_id=lease.environment_id,
message=f"Vast.ai capacity {slot_number} of {total_slots} is ready",
environment_message="Vast.ai worker ready; preparing remote assets next",
current=slot_number,
total=total_slots,
status_callback=status_callback,
environment_status_callback=environment_status_callback,
)
return leases_by_environment
def _emit_vast_capacity_status(
*,
environment_id: str,
message: str,
environment_message: str | None = None,
current: int,
total: int,
status_callback: SetupStatusCallback | None,
environment_status_callback: EnvironmentSetupStatusCallback | None,
) -> None:
"""Publish one Vast setup update at prompt and environment scopes."""
if status_callback is not None:
status_callback(message, current, total)
if environment_status_callback is not None:
environment_status_callback(
environment_id,
environment_message or message,
None,
None,
)
def _quote_selected_vast_slot(
*,
configuration: VastRemoteConfiguration,
component_ids: list[str],
requirements_by_component: Mapping[str, ComponentResourceRequirements],
vast_quotes: Mapping[tuple[str, str], VastProfileQuote],
vast_service: VastService,
) -> VastProfileQuote:
"""Quote aggregate slot requirements immediately before acquisition."""
minimum_vram_bytes = max(
requirements_by_component[component_id].minimum_vram_bytes
for component_id in component_ids
)
minimum_ram_bytes = max(
requirements_by_component[component_id].minimum_ram_bytes
for component_id in component_ids
)
predicted_execution_seconds = sum(
requirements_by_component[component_id].estimated_execution_seconds
for component_id in component_ids
)
if len(component_ids) == 1:
existing_quote = vast_quotes.get(
(component_ids[0], configuration.configuration_id)
)
if existing_quote is not None:
return existing_quote
return vast_service.quote_best_profile_sync(
[configuration.profile],
minimum_vram_bytes=minimum_vram_bytes,
minimum_ram_bytes=minimum_ram_bytes,
predicted_execution_seconds=predicted_execution_seconds,
)
def _ssh_sync_engine(
*,
host: SshHostConfig,
settings: ModalSyncSettings,
r2_cache: R2CacheClient | None = None,
) -> ModalAssetSyncEngine:
"""Build a content-addressed sync engine for one SSH Docker host."""
ssh_settings = replace(
settings,
execution_mode="ssh_docker",
volume_name=host.resolved_storage_volume_name,
local_storage_root=(
settings.local_storage_root / "ssh" / host.environment_id
).resolve(),
remote_storage_root="/storage",
)
resolved_r2_cache = r2_cache or R2CacheClient.from_environment()
controller = SshDockerController(host)
runtime_manager = SshRuntimeManager(
controller=controller,
repo_root=Path(__file__).resolve().parent,
settings=settings,
)
materializer_spec = runtime_manager.runtime_spec()
def prepare_materializer_image() -> None:
"""Make the SSH runtime image available before an R2 helper starts."""
runtime_manager.ensure_image(materializer_spec)
volume = SshDockerVolumeBackend(
controller,
host.resolved_storage_volume_name,
materializer_image=materializer_spec.image_tag,
materializer_image_preparer=(
prepare_materializer_image if resolved_r2_cache is not None else None
),
)
return ModalAssetSyncEngine(
volume=volume,
settings=ssh_settings,
r2_cache=resolved_r2_cache,
)
def _workflow_r2_cache(
configuration_set: RemoteConfigurationSet | None,
) -> R2CacheClient | None:
"""Resolve the connected R2 backing through its opaque OS-keyring reference."""
if configuration_set is None:
return None
r2_configurations = [
configuration
for configuration in configuration_set.storage_configurations
if isinstance(configuration, R2StorageBackingConfiguration)
]
if not r2_configurations:
return None
storage = r2_configurations[0]
try:
configuration = R2CredentialStore().cache_configuration(storage)
except (RuntimeError, ValueError) as exc:
raise ModalPromptValidationError(str(exc)) from exc
return R2CacheClient(configuration)
def _stamp_execution_assignment(
payload: dict[str, Any],
assignment: ExecutionAssignment,
worker_index: int = 0,
execution_history_signature: str | None = None,
execution_location: str | None = None,
provider_metadata: Mapping[str, Any] | None = None,
) -> None:
"""Attach provider placement to a payload and every nested proxy phase."""
payload["execution_provider"] = assignment.provider.value
payload["execution_environment_id"] = assignment.environment_id
payload["execution_worker_index"] = worker_index
if execution_location:
payload["execution_location"] = execution_location
if execution_history_signature:
payload["execution_history_signature"] = execution_history_signature
if provider_metadata:
payload.update(provider_metadata)
split_payloads = payload.get("split_proxy_payloads")
nested_payloads: list[dict[str, Any]] = []
if isinstance(split_payloads, dict):
nested_payloads.extend(
nested_payload
for nested_payload in split_payloads.values()
if isinstance(nested_payload, dict)
)
elif isinstance(split_payloads, list):
nested_payloads.extend(
nested_payload
for nested_payload in split_payloads
if isinstance(nested_payload, dict)
)
for nested_payload in nested_payloads:
_stamp_execution_assignment(
nested_payload,
assignment,
worker_index,
execution_history_signature,
execution_location,
provider_metadata,
)
def _ssh_hostname(ssh_target: str) -> str:
"""Return the host portion of one validated OpenSSH destination."""
host = ssh_target.rsplit("@", 1)[-1].strip()
if host.startswith("[") and "]" in host:
return host[1 : host.index("]")]
return host
def _execution_location_for_assignment(
assignment: ExecutionAssignment,
ssh_hosts_by_id: Mapping[str, SshHostConfig],
vast_leases_by_environment: Mapping[str, Any] | None = None,
) -> str | None:
"""Return the runtime location label known before remote dispatch."""
if assignment.provider is ExecutionProvider.MODAL:
return None
if assignment.provider is ExecutionProvider.VAST:
lease = (vast_leases_by_environment or {}).get(assignment.environment_id)
if lease is not None:
return str(lease.ssh_host or lease.gpu_name)
return assignment.environment_id
host = ssh_hosts_by_id.get(assignment.environment_id)
return _ssh_hostname(host.ssh_target) if host is not None else assignment.environment_id
def _vast_provider_metadata(lease: Any) -> dict[str, Any]:
"""Return credential-free lease details for execution and local status output."""
return {
"vast_instance_id": lease.instance_id,
"vast_profile_id": lease.profile_id,
"vast_profile_name": lease.profile_name,
"vast_gpu_name": lease.gpu_name,
"vast_gpu_count": lease.gpu_count,
"vast_gpu_ram_mb": lease.gpu_ram_mb,
"vast_hourly_cost_usd": lease.hourly_cost_usd,
"vast_idle_retention_seconds": lease.idle_retention_seconds,
}
def _configured_provider_metadata(
*,
execution_plan: ComponentExecutionPlan,
assignment: ExecutionAssignment,
vast_leases_by_environment: Mapping[str, Any],
) -> dict[str, Any] | None:
"""Return safe provider metadata needed to execute one prepared assignment."""
if assignment.provider is ExecutionProvider.VAST:
return _vast_provider_metadata(
vast_leases_by_environment[assignment.environment_id]
)
configuration_id = assignment.configuration_id
if configuration_id is None:
return None
configuration = execution_plan.configurations_by_id.get(configuration_id)
if isinstance(configuration, ModalRemoteConfiguration):
return {
"remote_configuration_id": configuration.configuration_id,
"remote_configuration_name": configuration.display_name,
"modal_gpu": configuration.gpu_type,
"modal_max_containers": configuration.instance_count,
}
if isinstance(configuration, SshRemoteConfiguration):
host = execution_plan.ssh_hosts_by_id[configuration.configuration_id]
portable_host = replace(
host,
health=EnvironmentHealth.UNKNOWN,
last_error=None,
)
return {
"remote_configuration_id": configuration.configuration_id,
"remote_configuration_name": configuration.display_name,
"ssh_host_config": portable_host.to_dict(),
}
return None
def _ensure_remote_sync_backend(
settings: ModalSyncSettings,
sync_engine: ModalAssetSyncEngine,
) -> None:
"""Fail before queueing when remote execution cannot write to Modal-visible storage."""
if settings.execution_mode == "local":
return
if isinstance(sync_engine.volume, ModalVolumeBackend):
return
raise ModalPromptValidationError(
"Remote Modal execution requires asset sync to use the Modal volume backend, "
f"but the active sync backend is {type(sync_engine.volume).__name__}. "
"Restart ComfyUI with COMFY_MODAL_EXECUTION_MODE=remote and the Modal SDK available "
"so synced assets and custom_nodes bundles are visible inside Modal workers."
)
def _emit_modal_status(
prompt_server: Any,
phase: str,
*,
client_id: str | None,
prompt_id: str | None,
node_ids: list[str],
configurator_node_id: str | None = None,
modal_gpu: str | None = None,
component_node_ids_by_representative: dict[str, list[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,
execution_environment_id: str | None = None,
remote_execution_assignments: Mapping[str, Mapping[str, Any]] | None = None,
remote_execution_configurations: Sequence[Mapping[str, Any]] | None = None,
) -> None:
"""Send a Modal execution status event to the active websocket client."""
if client_id is None:
logger.debug(
"Skipping Modal status event %s because no client id is available.", phase
)
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 configurator_node_id is not None:
payload["configurator_node_id"] = configurator_node_id
if component_node_ids_by_representative:
payload["components"] = [
{
"representative_node_id": representative_node_id,
"node_ids": list(component_node_ids),
}
for representative_node_id, component_node_ids in sorted(
component_node_ids_by_representative.items()
)
]
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 execution_environment_id is not None:
payload["execution_environment_id"] = execution_environment_id
if remote_execution_assignments is not None:
payload["remote_execution_assignments"] = copy.deepcopy(
dict(remote_execution_assignments)
)
if remote_execution_configurations is not None:
payload["remote_execution_configurations"] = copy.deepcopy(
list(remote_execution_configurations)
)
record_modal_ui_event("modal_status", payload, client_id)
prompt_server.send_sync("modal_status", payload, client_id)
def record_modal_ui_event(
event: str, payload: Mapping[str, Any], client_id: str | None
) -> None:
"""Store one client-scoped Modal UI event so the browser can replay it after refocus."""
if client_id is None:
return
event_record = {
"event": event,
"payload": copy.deepcopy(dict(payload)),
"updated_at": time.time(),
}
with _MODAL_UI_EVENTS_LOCK:
client_events = _MODAL_UI_EVENTS_BY_CLIENT.setdefault(
client_id,
deque(maxlen=_MODAL_UI_EVENT_LIMIT_PER_CLIENT),
)
client_events.append(event_record)
_prune_modal_ui_events_locked(client_events)
def modal_ui_events_for_client(client_id: str | None) -> list[dict[str, Any]]:
"""Return recent Modal UI events for one websocket client."""
if not client_id:
return []
with _MODAL_UI_EVENTS_LOCK:
client_events = _MODAL_UI_EVENTS_BY_CLIENT.get(client_id)
if client_events is None:
return []
_prune_modal_ui_events_locked(client_events)
return [copy.deepcopy(event_record) for event_record in client_events]
def _prune_modal_ui_events_locked(client_events: deque[dict[str, Any]]) -> None:
"""Discard stale Modal UI events while the caller holds the event lock."""
cutoff = time.time() - _MODAL_UI_EVENT_RETENTION_SECONDS
while client_events and float(client_events[0].get("updated_at", 0.0)) < cutoff:
client_events.popleft()
def _get_nodes_module() -> Any:
"""Import the ComfyUI nodes module lazily."""
import nodes
return nodes
def _get_server_module() -> Any:
"""Import the ComfyUI server module lazily."""
import server
return server
def _get_execution_module() -> Any:
"""Import the ComfyUI execution module lazily."""
import execution
return execution
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 _prompt_node_class_type(prompt: Mapping[str, Any], node_id: str) -> str:
"""Return the class type for a prompt node, or a diagnostic placeholder."""
prompt_node = prompt.get(str(node_id))
if not isinstance(prompt_node, Mapping):
return "<missing>"
return str(prompt_node.get("class_type", "<missing>"))
def _prompt_graph_links(prompt: Mapping[str, Any]) -> list[PromptGraphLink]:
"""Return direct prompt dependency edges from linked inputs."""
links: list[PromptGraphLink] = []
for target_node_id in sorted(str(node_id) for node_id in prompt):
prompt_node = prompt.get(target_node_id)
if not isinstance(prompt_node, Mapping):
continue
inputs = prompt_node.get("inputs") or {}
if not isinstance(inputs, Mapping):
continue
for input_name, input_value in sorted(
inputs.items(), key=lambda item: str(item[0])
):
if not _is_link(input_value):
continue
source_node_id = str(input_value[0])
links.append(
PromptGraphLink(
source_node_id=source_node_id,
source_output_index=int(input_value[1]),
target_node_id=target_node_id,
target_input_name=str(input_name),
source_class_type=_prompt_node_class_type(prompt, source_node_id),
target_class_type=_prompt_node_class_type(prompt, target_node_id),
)
)
return links
def _prompt_link_dict(link: PromptGraphLink) -> dict[str, Any]:
"""Return a JSON-safe representation of one prompt dependency edge."""
return {
"source_node_id": link.source_node_id,
"source_output_index": link.source_output_index,
"source_class_type": link.source_class_type,
"target_node_id": link.target_node_id,
"target_input_name": link.target_input_name,
"target_class_type": link.target_class_type,
}
def _find_prompt_dependency_cycles(prompt: Mapping[str, Any]) -> list[list[str]]:
"""Return representative dependency cycles in the prompt graph."""
links = _prompt_graph_links(prompt)
adjacency: dict[str, list[str]] = defaultdict(list)
for link in links:
if link.source_node_id not in prompt or link.target_node_id not in prompt:
continue
adjacency[link.source_node_id].append(link.target_node_id)
visited_node_ids: set[str] = set()
active_node_ids: set[str] = set()
path: list[str] = []
cycles: list[list[str]] = []
seen_cycle_keys: set[tuple[str, ...]] = set()
def normalize_cycle(cycle: list[str]) -> tuple[str, ...]:
"""Return a rotation-stable key for a detected cycle."""
if not cycle:
return ()
rotations = [
tuple(cycle[index:] + cycle[:index]) for index in range(len(cycle))
]
return min(rotations)
def visit(node_id: str) -> None:
"""Depth-first search one prompt node for dependency back-edges."""
if node_id in active_node_ids:
cycle_start_index = path.index(node_id)
cycle = path[cycle_start_index:] + [node_id]
cycle_key = normalize_cycle(cycle[:-1])
if cycle_key not in seen_cycle_keys:
seen_cycle_keys.add(cycle_key)
cycles.append(cycle)
return
if node_id in visited_node_ids:
return
visited_node_ids.add(node_id)
active_node_ids.add(node_id)
path.append(node_id)
for downstream_node_id in sorted(adjacency.get(node_id, [])):
visit(downstream_node_id)
path.pop()
active_node_ids.remove(node_id)
for node_id in sorted(str(node_id) for node_id in prompt):
visit(node_id)
return cycles
def _modal_proxy_payload_summaries(prompt: Mapping[str, Any]) -> list[dict[str, Any]]:
"""Return compact summaries for Modal proxy payloads embedded in a rewritten prompt."""
summaries: list[dict[str, Any]] = []
for node_id in sorted(str(node_id) for node_id in prompt):
prompt_node = prompt.get(node_id)
if not isinstance(prompt_node, Mapping):
continue
inputs = prompt_node.get("inputs") or {}
if not isinstance(inputs, Mapping):
continue
payload = inputs.get("original_node_data")
if not isinstance(payload, Mapping):
continue
boundary_inputs = [
{
"proxy_input_name": str(boundary_input.get("proxy_input_name")),
"io_type": str(boundary_input.get("io_type")),
"targets": copy.deepcopy(boundary_input.get("targets", [])),
}
for boundary_input in payload.get("boundary_inputs", [])
if isinstance(boundary_input, Mapping)
]
boundary_outputs = [
{
"proxy_output_name": str(boundary_output.get("proxy_output_name")),
"node_id": str(boundary_output.get("node_id")),
"output_index": int(boundary_output.get("output_index", 0)),
"io_type": str(boundary_output.get("io_type")),
"is_list": bool(boundary_output.get("is_list", False)),
"session_output": bool(boundary_output.get("session_output", False)),
}
for boundary_output in payload.get("boundary_outputs", [])
if isinstance(boundary_output, Mapping)
]
summaries.append(
{
"proxy_node_id": node_id,
"proxy_class_type": _prompt_node_class_type(prompt, node_id),
"payload_kind": str(payload.get("payload_kind")),
"component_id": str(payload.get("component_id")),
"component_node_ids": [
str(value) for value in payload.get("component_node_ids", [])
],
"execute_node_ids": [
str(value) for value in payload.get("execute_node_ids", [])
],
"boundary_inputs": boundary_inputs,
"boundary_outputs": boundary_outputs,
}
)
return summaries
def _modal_rewritten_prompt_diagnostics(
prompt: Mapping[str, Any],
summary: RewriteSummary | None = None,
) -> dict[str, Any]:
"""Return compact diagnostics for a rewritten Modal prompt graph."""
node_class_types = {
str(node_id): _prompt_node_class_type(prompt, str(node_id))
for node_id in sorted(str(node_id) for node_id in prompt)
}
diagnostics: dict[str, Any] = {
"node_count": len(prompt),
"node_class_types": node_class_types,
"links": [_prompt_link_dict(link) for link in _prompt_graph_links(prompt)],
"cycles": _find_prompt_dependency_cycles(prompt),
"modal_proxy_payloads": _modal_proxy_payload_summaries(prompt),
}
if summary is not None:
diagnostics["remote_node_ids"] = list(summary.remote_node_ids)
diagnostics["remote_component_ids"] = list(summary.remote_component_ids)
diagnostics["component_node_ids_by_representative"] = copy.deepcopy(
summary.component_node_ids_by_representative
)
diagnostics["component_dependency_ids_by_representative"] = copy.deepcopy(
summary.component_dependency_ids_by_representative
)
diagnostics["component_execution_stages"] = copy.deepcopy(
summary.component_execution_stages
)
diagnostics["parallel_local_branch_node_ids"] = list(
summary.parallel_local_branch_node_ids
)
diagnostics["rewritten_node_id_map"] = copy.deepcopy(
summary.rewritten_node_id_map
)
return diagnostics
def _log_modal_rewritten_prompt_diagnostics(
*,
prompt_id: str | None,
prompt: Mapping[str, Any],
summary: RewriteSummary | None = None,
reason: str,
level: int = logging.INFO,
) -> None:
"""Log compact diagnostics for a rewritten Modal prompt graph."""
diagnostics = _modal_rewritten_prompt_diagnostics(prompt, summary)
cycles = diagnostics.get("cycles") or []
if cycles:
logger.warning(
"Modal rewritten prompt contains dependency cycle(s) prompt_id=%s reason=%s cycles=%s",
prompt_id,
reason,
cycles,
)
logger.log(
level,
"Modal rewritten prompt diagnostics prompt_id=%s reason=%s diagnostics=%s",
prompt_id,
reason,
json.dumps(diagnostics, sort_keys=True),
)
def _prompt_value_signature_fragment(
prompt: dict[str, Any],
value: Any,
*,
memo: dict[str, str],
) -> Any:
"""Return a stable structural signature fragment for one prompt input value."""
if _is_link(value):
source_node_id = str(value[0])
return {
"kind": "link",
"source_node_id": source_node_id,
"output_index": int(value[1]),
"source_digest": _prompt_node_signature_digest(
prompt,
source_node_id,
memo=memo,
),
}
if value is None or isinstance(value, bool | int | str):
return value
if isinstance(value, float):
if value != value:
return {"kind": "float", "value": "nan"}
if value == float("inf"):
return {"kind": "float", "value": "inf"}
if value == float("-inf"):
return {"kind": "float", "value": "-inf"}
return value
if isinstance(value, list):
return {
"kind": "list",
"items": [
_prompt_value_signature_fragment(prompt, item, memo=memo)
for item in value
],
}
if isinstance(value, tuple):
return {
"kind": "tuple",
"items": [
_prompt_value_signature_fragment(prompt, item, memo=memo)
for item in value
],
}
if isinstance(value, dict):
return {
"kind": "dict",
"items": [
{
"key": str(key),
"value": _prompt_value_signature_fragment(
prompt, value[key], memo=memo
),
}
for key in sorted(value)
],
}
return {
"kind": "repr",
"type": type(value).__name__,
"value": repr(value),
}
def _prompt_node_signature_digest(
prompt: dict[str, Any],
node_id: str,
*,
memo: dict[str, str],
) -> str:
"""Return a stable digest for one prompt node and its upstream prompt inputs."""
if node_id in memo:
return memo[node_id]
prompt_node = prompt.get(str(node_id))
if prompt_node is None:
digest = hashlib.sha256(
json.dumps(
{"kind": "missing-node", "node_id": str(node_id)},
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
).hexdigest()
memo[str(node_id)] = digest
return digest
payload = {
"kind": "prompt-node",
"node_id": str(node_id),
"class_type": str(prompt_node.get("class_type", "")),
"inputs": [
{
"name": str(input_name),
"value": _prompt_value_signature_fragment(
prompt,
input_value,
memo=memo,
),
}
for input_name, input_value in sorted(
(prompt_node.get("inputs") or {}).items()
)
],
}
digest = hashlib.sha256(
json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest()
memo[str(node_id)] = digest
return digest
def _iter_loader_snapshot_prompt_payloads(
payload: Mapping[str, Any]
) -> Iterator[Mapping[str, Any]]:
"""Yield prompt-bearing 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, Mapping):
yield phase_payload
return
if isinstance(split_proxy_payloads, list):
for phase_payload in split_proxy_payloads:
if isinstance(phase_payload, Mapping):
yield phase_payload
return
if isinstance(payload.get("subgraph_prompt"), Mapping):
yield payload
def _is_root_literal_loader_node(prompt_node: Mapping[str, Any]) -> bool:
"""Return whether one prompt node is a supported root loader with 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 _uses_llm_worker_affinity(payload: Mapping[str, Any]) -> bool:
"""Return whether an execution payload belongs to the isolated LLM worker pool."""
affinity_group = (
str(payload.get("remote_worker_affinity_group") or "").strip().lower()
)
return affinity_group == "llm"
def _payload_loader_snapshot_profile_key(payload: Mapping[str, Any]) -> str:
"""Return the stable loader snapshot profile key derivable from one payload."""
prompt_id = payload.get("prompt_id")
normalized_prompt_id = str(prompt_id) if prompt_id is not None else None
plan_signatures: set[str] = set()
for prompt_payload in _iter_loader_snapshot_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
plan_signatures.add(_loader_prewarm_plan_signature(class_type, inputs))
logger.debug(
"Derived rewrite-time loader prewarm plan for component=%s node=%s class_type=%s prompt_id=%s.",
payload.get("component_id"),
node_id,
class_type,
normalized_prompt_id,
)
if not plan_signatures:
return ""
profile_digest = hashlib.sha256(
json.dumps({"plan_signatures": sorted(plan_signatures)}, sort_keys=True).encode(
"utf-8"
)
).hexdigest()
return f"loader-profile:{profile_digest}"
def _stamp_snapshot_profile_key(
payload: dict[str, Any], snapshot_profile_key: str
) -> None:
"""Attach one loader snapshot profile to eligible Comfy payload descendants."""
if not snapshot_profile_key:
return
if _uses_llm_worker_affinity(payload):
payload.pop("snapshot_profile_key", None)
logger.info(
"Omitting Comfy loader snapshot profile from LLM worker component=%s.",
payload.get("component_id"),
)
else:
payload["snapshot_profile_key"] = snapshot_profile_key
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):
_stamp_snapshot_profile_key(phase_payload, snapshot_profile_key)
return
if isinstance(split_proxy_payloads, list):
for phase_payload in split_proxy_payloads:
if isinstance(phase_payload, dict):
_stamp_snapshot_profile_key(phase_payload, snapshot_profile_key)
def _attach_snapshot_profile_key(
payload: dict[str, Any], settings: ModalSyncSettings
) -> dict[str, Any]:
"""Stamp a deterministic loader snapshot profile onto one payload when enabled."""
if not settings.enable_gpu_memory_snapshot or not settings.enable_loader_prewarm:
return payload
snapshot_profile_key = _payload_loader_snapshot_profile_key(payload)
if snapshot_profile_key:
_stamp_snapshot_profile_key(payload, snapshot_profile_key)
logger.info(
"Attached rewrite-time loader snapshot profile %s to component=%s payload_kind=%s.",
snapshot_profile_key,
payload.get("component_id"),
payload.get("payload_kind"),
)
return payload
def _resolved_llm_profile_entry(
profile: Any,
settings: ModalSyncSettings,
) -> dict[str, Any]:
"""Serialize planner metadata plus its persisted security-scan result."""
from .llm_profiles import generated_profile_manifest_path
security_scan_complete = True
if getattr(profile, "source", "curated") == "generated":
manifest_path = generated_profile_manifest_path(
settings.local_storage_root,
profile.profile_id,
)
try:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
except (FileNotFoundError, OSError, json.JSONDecodeError) as exc:
raise ModalPromptValidationError(
f"Planner-resolved LLM manifest {manifest_path} is unavailable."
) from exc
if not isinstance(manifest, Mapping):
raise ModalPromptValidationError(
f"Planner-resolved LLM manifest {manifest_path} is invalid."
)
security_scan_complete = bool(
manifest.get("security_scan_complete", False)
)
return {
"profile": profile.to_mapping(),
"security_scan_complete": security_scan_complete,
}
def _attach_resolved_llm_profiles(
payload: dict[str, Any],
resolved_profiles: Mapping[str, Any],
settings: ModalSyncSettings,
) -> None:
"""Attach only the planner-resolved profiles used by each payload subtree."""
from .llm_profiles import llm_model_references_from_payload
if not resolved_profiles:
return
references = llm_model_references_from_payload(payload)
entries = {
reference: _resolved_llm_profile_entry(resolved_profiles[reference], settings)
for reference in references
if reference in resolved_profiles
}
if entries:
payload["resolved_llm_profiles"] = entries
split_payloads = payload.get("split_proxy_payloads")
if isinstance(split_payloads, Mapping):
children = split_payloads.values()
elif isinstance(split_payloads, list):
children = split_payloads
else:
children = ()
for child in children:
if isinstance(child, dict):
_attach_resolved_llm_profiles(child, resolved_profiles, settings)
def _boundary_source_signature(
prompt: dict[str, Any],
source: LinkedOutputRef,
) -> str:
"""Return a stable prompt-structural fingerprint for one boundary source output."""
payload = {
"kind": "boundary-source",
"source_node_id": source.node_id,
"output_index": int(source.output_index),
"source_digest": _prompt_node_signature_digest(
prompt,
source.node_id,
memo={},
),
}
digest = hashlib.sha256(
json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest()
return f"SRC_{digest}"
def _serialize_boundary_input_specs(
boundary_inputs: list[BoundaryInputSpec],
*,
signature_prompt: dict[str, Any],
) -> list[dict[str, Any]]:
"""Return serialized boundary input payloads with stable provenance for non-transportable inputs."""
serialized_boundary_inputs: list[dict[str, Any]] = []
for boundary_input in boundary_inputs:
serialized_boundary_input = {
"proxy_input_name": boundary_input.proxy_input_name,
"io_type": boundary_input.io_type,
"targets": [
{"node_id": target.node_id, "input_name": target.input_name}
for target in boundary_input.targets
],
}
if not _is_transportable_output_type(boundary_input.io_type):
serialized_boundary_input["source_signature"] = _boundary_source_signature(
signature_prompt,
boundary_input.source,
)
serialized_boundary_inputs.append(serialized_boundary_input)
return serialized_boundary_inputs
def _iter_payload_input_strings(value: Any) -> Iterator[str]:
"""Yield string literals nested inside one serialized prompt input value."""
if isinstance(value, str):
yield value
return
if isinstance(value, list):
if len(value) == 2 and isinstance(value[0], str):
return
for item in value:
yield from _iter_payload_input_strings(item)
return
if isinstance(value, dict):
for nested_value in value.values():
yield from _iter_payload_input_strings(nested_value)
def _looks_like_workflow_node(fragment: dict[str, Any]) -> bool:
"""Return whether a JSON fragment resembles a saved ComfyUI workflow node."""
return "id" in fragment and "properties" in fragment
def _iter_workflow_nodes(
workflow_fragment: Any,
visited_object_ids: set[int] | None = None,
) -> Iterator[dict[str, Any]]:
"""Yield workflow node dictionaries from a nested saved workflow fragment."""
if visited_object_ids is None:
visited_object_ids = set()
if isinstance(workflow_fragment, dict):
object_id = id(workflow_fragment)
if object_id in visited_object_ids:
return
visited_object_ids.add(object_id)
if _looks_like_workflow_node(workflow_fragment):
yield workflow_fragment
for value in workflow_fragment.values():
yield from _iter_workflow_nodes(value, visited_object_ids)
return
if isinstance(workflow_fragment, list):
object_id = id(workflow_fragment)
if object_id in visited_object_ids:
return
visited_object_ids.add(object_id)
for item in workflow_fragment:
yield from _iter_workflow_nodes(item, visited_object_ids)
def _workflow_subgraph_definitions(
workflow_fragment: Any,
) -> dict[str, dict[str, Any]]:
"""Return reusable ComfyUI subgraph definitions keyed by their node type id."""
if not isinstance(workflow_fragment, dict):
return {}
definitions = workflow_fragment.get("definitions")
if not isinstance(definitions, dict):
return {}
subgraphs = definitions.get("subgraphs")
if not isinstance(subgraphs, list):
return {}
return {
str(subgraph["id"]): subgraph
for subgraph in subgraphs
if isinstance(subgraph, dict) and subgraph.get("id") is not None
}
def _iter_workflow_nodes_with_ancestors(
workflow_fragment: Any,
ancestor_node_ids: tuple[str, ...] = (),
visited_fragments: set[tuple[int, tuple[str, ...], tuple[str, ...]]] | None = None,
subgraph_definitions: Mapping[str, dict[str, Any]] | None = None,
active_subgraph_definition_ids: tuple[str, ...] = (),
) -> Iterator[tuple[dict[str, Any], tuple[str, ...]]]:
"""Yield workflow nodes with instance paths across embedded and defined subgraphs."""
if visited_fragments is None:
visited_fragments = set()
if subgraph_definitions is None:
subgraph_definitions = _workflow_subgraph_definitions(workflow_fragment)
if isinstance(workflow_fragment, dict):
fragment_identity = (
id(workflow_fragment),
ancestor_node_ids,
active_subgraph_definition_ids,
)
if fragment_identity in visited_fragments:
return
visited_fragments.add(fragment_identity)
next_ancestor_node_ids = ancestor_node_ids
if _looks_like_workflow_node(workflow_fragment):
node_id = str(workflow_fragment.get("id"))
yield workflow_fragment, ancestor_node_ids
next_ancestor_node_ids = ancestor_node_ids + (node_id,)
subgraph_definition_id = str(workflow_fragment.get("type") or "")
subgraph_definition = subgraph_definitions.get(subgraph_definition_id)
if (
subgraph_definition is not None
and subgraph_definition_id not in active_subgraph_definition_ids
):
yield from _iter_workflow_nodes_with_ancestors(
subgraph_definition.get("nodes", []),
next_ancestor_node_ids,
visited_fragments,
subgraph_definitions,
active_subgraph_definition_ids + (subgraph_definition_id,),
)
for key, value in workflow_fragment.items():
if key == "definitions":
continue
yield from _iter_workflow_nodes_with_ancestors(
value,
next_ancestor_node_ids,
visited_fragments,
subgraph_definitions,
active_subgraph_definition_ids,
)
return
if isinstance(workflow_fragment, list):
fragment_identity = (
id(workflow_fragment),
ancestor_node_ids,
active_subgraph_definition_ids,
)
if fragment_identity in visited_fragments:
return
visited_fragments.add(fragment_identity)
for item in workflow_fragment:
yield from _iter_workflow_nodes_with_ancestors(
item,
ancestor_node_ids,
visited_fragments,
subgraph_definitions,
active_subgraph_definition_ids,
)
def _resolve_prompt_node_ids_for_workflow_node(
workflow_node_id: str,
ancestor_node_ids: tuple[str, ...],
prompt_node_ids: set[str],
) -> set[str]:
"""Resolve one saved workflow node id to matching queued prompt node ids."""
path_segments = ancestor_node_ids + (workflow_node_id,)
for index in range(len(path_segments)):
candidate = ":".join(path_segments[index:])
resolved_prompt_node_ids: set[str] = set()
if candidate in prompt_node_ids:
resolved_prompt_node_ids.add(candidate)
descendant_prefix = f"{candidate}:"
resolved_prompt_node_ids.update(
prompt_node_id
for prompt_node_id in prompt_node_ids
if prompt_node_id.startswith(descendant_prefix)
)
if resolved_prompt_node_ids:
return resolved_prompt_node_ids
resolved_prompt_node_ids: set[str] = set()
for ancestor_node_id in reversed(ancestor_node_ids):
if ancestor_node_id in prompt_node_ids:
resolved_prompt_node_ids.add(ancestor_node_id)
descendant_prefix = f"{ancestor_node_id}:"
resolved_prompt_node_ids.update(
prompt_node_id
for prompt_node_id in prompt_node_ids
if prompt_node_id.startswith(descendant_prefix)
)
break
return resolved_prompt_node_ids
def _workflow_node_path(
workflow_node_id: str, ancestor_node_ids: tuple[str, ...]
) -> str:
"""Return one workflow node's composed path, including subgraph ancestors."""
if not ancestor_node_ids:
return workflow_node_id
return ":".join((*ancestor_node_ids, workflow_node_id))
def _extract_marked_workflow_node_paths(
workflow: dict[str, Any] | None,
settings: ModalSyncSettings | None = None,
) -> set[str]:
"""Return composed workflow paths for nodes explicitly marked remote in metadata."""
if workflow is None:
return set()
marker = (settings or get_settings()).marker_property
marked_workflow_node_paths: set[str] = set()
for node, ancestor_node_ids in _iter_workflow_nodes_with_ancestors(workflow):
properties = node.get("properties") or {}
if not properties.get(marker):
continue
marked_workflow_node_paths.add(
_workflow_node_path(str(node.get("id")), ancestor_node_ids)
)
return marked_workflow_node_paths
def _build_workflow_prompt_resolution_maps(
workflow: dict[str, Any] | None,
prompt_node_ids: set[str],
) -> tuple[dict[str, set[str]], dict[str, set[str]]]:
"""Return bidirectional mappings between workflow paths and queued prompt ids."""
workflow_path_to_prompt_node_ids: dict[str, set[str]] = {}
prompt_node_id_to_workflow_paths: dict[str, set[str]] = defaultdict(set)
if workflow is None:
for prompt_node_id in prompt_node_ids:
workflow_path_to_prompt_node_ids[prompt_node_id] = {prompt_node_id}
prompt_node_id_to_workflow_paths[prompt_node_id].add(prompt_node_id)
return workflow_path_to_prompt_node_ids, prompt_node_id_to_workflow_paths
for node, ancestor_node_ids in _iter_workflow_nodes_with_ancestors(workflow):
workflow_node_id = str(node.get("id"))
workflow_node_path = _workflow_node_path(workflow_node_id, ancestor_node_ids)
resolved_prompt_node_ids = _resolve_prompt_node_ids_for_workflow_node(
workflow_node_id,
ancestor_node_ids,
prompt_node_ids,
)
if not ancestor_node_ids and workflow_node_id in prompt_node_ids:
resolved_prompt_node_ids.add(workflow_node_id)
if workflow_node_path in prompt_node_ids:
resolved_prompt_node_ids.add(workflow_node_path)
if not resolved_prompt_node_ids:
continue
workflow_path_to_prompt_node_ids[workflow_node_path] = resolved_prompt_node_ids
for prompt_node_id in resolved_prompt_node_ids:
prompt_node_id_to_workflow_paths[prompt_node_id].add(workflow_node_path)
for prompt_node_id in prompt_node_ids:
workflow_path_to_prompt_node_ids.setdefault(prompt_node_id, {prompt_node_id})
prompt_node_id_to_workflow_paths[prompt_node_id].add(prompt_node_id)
return workflow_path_to_prompt_node_ids, prompt_node_id_to_workflow_paths
def _resolve_requested_prompt_node_ids(
requested_workflow_node_paths: set[str],
prompt_node_ids: set[str],
workflow_path_to_prompt_node_ids: dict[str, set[str]],
) -> set[str]:
"""Resolve requested workflow-node paths to the prompt ids that queue-time rewrite sees."""
requested_prompt_node_ids: set[str] = set()
for requested_workflow_node_path in requested_workflow_node_paths:
if requested_workflow_node_path in workflow_path_to_prompt_node_ids:
requested_prompt_node_ids.update(
workflow_path_to_prompt_node_ids[requested_workflow_node_path]
)
continue
if requested_workflow_node_path in prompt_node_ids:
requested_prompt_node_ids.add(requested_workflow_node_path)
return requested_prompt_node_ids
def _best_workflow_path_for_prompt_node(
prompt_node_id: str,
prompt_node_id_to_workflow_paths: dict[str, set[str]],
) -> str:
"""Choose the most specific workflow path for one queued prompt node id."""
workflow_node_paths = prompt_node_id_to_workflow_paths.get(
prompt_node_id, {prompt_node_id}
)
return max(
workflow_node_paths,
key=lambda workflow_node_path: (
workflow_node_path.count(":"),
len(workflow_node_path),
workflow_node_path,
),
)
def _resolve_workflow_node_paths_for_prompt_nodes(
prompt_node_ids: set[str],
prompt_node_id_to_workflow_paths: dict[str, set[str]],
) -> set[str]:
"""Map queued prompt ids back to the workflow node paths the UI can mark remote."""
return {
_best_workflow_path_for_prompt_node(
prompt_node_id, prompt_node_id_to_workflow_paths
)
for prompt_node_id in prompt_node_ids
}
def extract_remote_node_ids(
workflow: dict[str, Any] | None,
settings: ModalSyncSettings | None = None,
prompt_node_ids: set[str] | None = None,
) -> set[str]:
"""Return the node ids marked for remote execution in the workflow metadata."""
if workflow is None:
return set()
marker = (settings or get_settings()).marker_property
remote_node_ids: set[str] = set()
for node, ancestor_node_ids in _iter_workflow_nodes_with_ancestors(workflow):
if _workflow_node_remote_enabled(node, marker):
node_id = str(node.get("id"))
if prompt_node_ids is None:
remote_node_ids.add(node_id)
continue
resolved_prompt_node_ids = _resolve_prompt_node_ids_for_workflow_node(
node_id,
ancestor_node_ids,
prompt_node_ids,
)
if resolved_prompt_node_ids:
remote_node_ids.update(resolved_prompt_node_ids)
logger.info(
"Mapped workflow Modal marker from node %s with ancestors %s to prompt nodes %s.",
node_id,
list(ancestor_node_ids),
sorted(resolved_prompt_node_ids),
)
return remote_node_ids
def requested_remote_node_ids(
*,
prompt: Mapping[str, Any],
workflow: Mapping[str, Any] | None,
settings: ModalSyncSettings,
) -> set[str]:
"""Return explicit remote markers or every eligible node for auto placement."""
preferences = WorkflowExecutionPreferences.from_workflow(workflow)
if not preferences.auto_place:
return extract_remote_node_ids(
dict(workflow) if workflow is not None else None,
settings,
set(prompt),
)
selected_node_ids = {
str(node_id)
for node_id, prompt_node in prompt.items()
if isinstance(prompt_node, Mapping)
and str(prompt_node.get("class_type") or "")
not in _LOCAL_ONLY_REMOTE_CLASS_TYPES
and not str(prompt_node.get("class_type") or "").startswith(
"ModalUniversalExecutor"
)
}
logger.info(
"Automatic remote placement selected %d eligible prompt nodes.",
len(selected_node_ids),
)
return selected_node_ids
def _workflow_node_remote_enabled(node: Mapping[str, Any], marker: str) -> bool:
"""Return the visible remote toggle, falling back to the saved marker property."""
named_widget_values = node.get("widgets_values_named")
if isinstance(named_widget_values, Mapping):
visible_toggle_value = named_widget_values.get(_REMOTE_TOGGLE_WIDGET_NAME)
if not isinstance(visible_toggle_value, bool):
visible_toggle_value = named_widget_values.get(
_LEGACY_REMOTE_TOGGLE_WIDGET_NAME
)
if isinstance(visible_toggle_value, bool):
return visible_toggle_value
properties = node.get("properties")
return isinstance(properties, Mapping) and bool(properties.get(marker))
def _normalize_output_metadata(
node_class: type[Any],
) -> tuple[tuple[str, ...], tuple[str, ...], tuple[bool, ...]]:
"""Return normalized output metadata for a node class."""
if hasattr(node_class, "GET_SCHEMA"):
node_class.GET_SCHEMA()
output_types = tuple(getattr(node_class, "RETURN_TYPES", ("*",))) or ("*",)
default_names = tuple(f"output_{index}" for index, _ in enumerate(output_types))
output_names = tuple(getattr(node_class, "RETURN_NAMES", default_names))
output_is_list = tuple(
getattr(node_class, "OUTPUT_IS_LIST", (False,) * len(output_types))
)
if len(output_names) < len(output_types):
output_names = output_names + default_names[len(output_names) :]
if len(output_is_list) < len(output_types):
output_is_list = output_is_list + (False,) * (
len(output_types) - len(output_is_list)
)
return (
output_types,
output_names[: len(output_types)],
output_is_list[: len(output_types)],
)
def _is_transportable_output_type(io_type: str) -> bool:
"""Return whether a ComfyUI output type can cross the current transport."""
normalized_parts = [part.strip() for part in io_type.split(",") if part.strip()]
return bool(normalized_parts) and all(
part in _TRANSPORTABLE_OUTPUT_TYPES for part in normalized_parts
)
def _is_inexpensive_remote_boundary_type(io_type: str) -> bool:
"""Return whether a remote edge is cheap enough to remain a component boundary."""
normalized_parts = [part.strip() for part in io_type.split(",") if part.strip()]
return bool(normalized_parts) and all(
part in _INEXPENSIVE_REMOTE_BOUNDARY_TYPES for part in normalized_parts
)
def _build_consumer_map(
prompt: dict[str, Any]
) -> dict[LinkedOutputRef, list[InputTarget]]:
"""Build a reverse map from node outputs to downstream prompt inputs."""
consumers: dict[LinkedOutputRef, list[InputTarget]] = defaultdict(list)
for node_id, prompt_node in prompt.items():
for input_name, input_value in (prompt_node.get("inputs") or {}).items():
if not _is_link(input_value):
continue
source = LinkedOutputRef(
node_id=str(input_value[0]), output_index=int(input_value[1])
)
consumers[source].append(
InputTarget(node_id=str(node_id), input_name=str(input_name))
)
return consumers
def _sandwiched_local_node_ids(
prompt: dict[str, Any],
remote_node_ids: set[str],
) -> set[str]:
"""Return local nodes on paths that leave and then re-enter remote execution."""
downstream_node_ids: dict[str, set[str]] = defaultdict(set)
upstream_node_ids: dict[str, set[str]] = defaultdict(set)
for target_node_id, prompt_node in prompt.items():
for input_value in (prompt_node.get("inputs") or {}).values():
if not _is_link(input_value):
continue
source_node_id = str(input_value[0])
safe_target_node_id = str(target_node_id)
downstream_node_ids[source_node_id].add(safe_target_node_id)
upstream_node_ids[safe_target_node_id].add(source_node_id)
def local_closure(adjacency: Mapping[str, set[str]]) -> set[str]:
"""Return local nodes reachable from the remote seed set in one direction."""
reachable_local_node_ids: set[str] = set()
pending_node_ids = list(sorted(remote_node_ids))
visited_node_ids = set(remote_node_ids)
while pending_node_ids:
current_node_id = pending_node_ids.pop()
for adjacent_node_id in adjacency.get(current_node_id, set()):
if adjacent_node_id in visited_node_ids:
continue
visited_node_ids.add(adjacent_node_id)
if adjacent_node_id in remote_node_ids:
continue
if adjacent_node_id not in prompt:
continue
reachable_local_node_ids.add(adjacent_node_id)
pending_node_ids.append(adjacent_node_id)
return reachable_local_node_ids
downstream_local_node_ids = local_closure(downstream_node_ids)
upstream_local_node_ids = local_closure(upstream_node_ids)
return downstream_local_node_ids & upstream_local_node_ids
def _node_output_refs(
prompt: dict[str, Any], node_id: str, nodes_module: Any
) -> list[LinkedOutputRef]:
"""Return declared output refs for one prompt node."""
prompt_node = prompt.get(node_id)
if prompt_node is None:
return []
node_class = nodes_module.NODE_CLASS_MAPPINGS.get(
str(prompt_node.get("class_type"))
)
if node_class is None:
return []
output_types, _, _ = _normalize_output_metadata(node_class)
return [
LinkedOutputRef(node_id=node_id, output_index=output_index)
for output_index, _io_type in enumerate(output_types)
]
def _downstream_node_ids_from_targets(
*,
prompt: dict[str, Any],
consumers: dict[LinkedOutputRef, list[InputTarget]],
seed_targets: list[InputTarget],
nodes_module: Any,
) -> set[str]:
"""Return every node reachable downstream from the supplied input targets."""
visited_node_ids: set[str] = set()
pending_node_ids = deque(str(target.node_id) for target in seed_targets)
while pending_node_ids:
node_id = pending_node_ids.popleft()
if node_id in visited_node_ids or node_id not in prompt:
continue
visited_node_ids.add(node_id)
for output_ref in _node_output_refs(prompt, node_id, nodes_module):
for downstream_target in consumers.get(output_ref, []):
downstream_node_id = str(downstream_target.node_id)
if downstream_node_id not in visited_node_ids:
pending_node_ids.append(downstream_node_id)
return visited_node_ids
def _is_non_returning_tap_terminal_node(
*,
prompt_node: dict[str, Any] | None,
nodes_module: Any,
) -> bool:
"""Return whether a local tap node should stay local instead of being absorbed."""
if prompt_node is None:
return False
class_type = str(prompt_node.get("class_type"))
if class_type == "PreviewImage":
return True
node_class = nodes_module.NODE_CLASS_MAPPINGS.get(class_type)
return bool(getattr(node_class, "OUTPUT_NODE", False))
def _non_returning_local_tap_node_ids(
*,
prompt: dict[str, Any],
component_node_ids: set[str],
remote_node_ids: set[str],
consumers: dict[LinkedOutputRef, list[InputTarget]],
nodes_module: Any,
) -> tuple[set[str], set[str]]:
"""Return non-returning local tap branches without making local nodes remote."""
local_tap_node_ids: set[str] = set()
local_tap_terminal_node_ids: set[str] = set()
for component_node_id in sorted(component_node_ids):
for source in _node_output_refs(prompt, component_node_id, nodes_module):
output_consumers = consumers.get(source, [])
local_consumers = [
target
for target in output_consumers
if target.node_id not in component_node_ids
and target.node_id not in remote_node_ids
]
remote_consumers = [
target
for target in output_consumers
if target.node_id in remote_node_ids
]
if not local_consumers or not remote_consumers:
continue
for local_consumer in local_consumers:
branch_node_ids = _downstream_node_ids_from_targets(
prompt=prompt,
consumers=consumers,
seed_targets=[local_consumer],
nodes_module=nodes_module,
)
if branch_node_ids & remote_node_ids:
continue
local_tap_terminal_node_ids.update(branch_node_ids)
return local_tap_node_ids, local_tap_terminal_node_ids
def _non_returning_local_output_consumers(
*,
prompt: dict[str, Any],
source: LinkedOutputRef,
remote_node_ids: set[str],
consumers: dict[LinkedOutputRef, list[InputTarget]],
nodes_module: Any,
) -> tuple[list[InputTarget], list[InputTarget]]:
"""Partition local consumers by whether their branch later returns remote."""
non_returning_consumers: list[InputTarget] = []
returning_consumers: list[InputTarget] = []
for target in consumers.get(source, []):
if target.node_id in remote_node_ids:
continue
branch_node_ids = _downstream_node_ids_from_targets(
prompt=prompt,
consumers=consumers,
seed_targets=[target],
nodes_module=nodes_module,
)
if branch_node_ids and not (branch_node_ids & remote_node_ids):
non_returning_consumers.append(target)
else:
returning_consumers.append(target)
return non_returning_consumers, returning_consumers
def _output_supports_parallel_local_materialization(
*,
prompt: dict[str, Any],
source: LinkedOutputRef,
remote_node_ids: set[str],
consumers: dict[LinkedOutputRef, list[InputTarget]],
nodes_module: Any,
) -> bool:
"""Return whether every local consumer is independent of later remote work."""
(
non_returning_consumers,
returning_consumers,
) = _non_returning_local_output_consumers(
prompt=prompt,
source=source,
remote_node_ids=remote_node_ids,
consumers=consumers,
nodes_module=nodes_module,
)
return bool(non_returning_consumers) and not returning_consumers
def _expand_component_for_non_transportable_local_outputs(
*,
prompt: dict[str, Any],
component_node_ids: set[str],
remote_node_ids: set[str],
consumers: dict[LinkedOutputRef, list[InputTarget]],
nodes_module: Any,
) -> tuple[set[str], set[str]]:
"""Absorb local consumers needed to avoid non-transportable component outputs."""
expanded_node_ids = set(component_node_ids)
added_node_ids: set[str] = set()
changed = True
while changed:
changed = False
(
upstream_expanded_node_ids,
_expansion_reasons,
) = _expand_remote_node_ids_for_non_transportable_inputs(
prompt=prompt,
remote_node_ids=expanded_node_ids,
nodes_module=nodes_module,
)
upstream_added_node_ids = upstream_expanded_node_ids - expanded_node_ids
if upstream_added_node_ids:
expanded_node_ids = upstream_expanded_node_ids
added_node_ids.update(upstream_added_node_ids)
changed = True
for node_id in sorted(expanded_node_ids):
prompt_node = prompt.get(node_id)
if prompt_node is None:
continue
node_class = nodes_module.NODE_CLASS_MAPPINGS.get(
str(prompt_node.get("class_type"))
)
if node_class is None:
continue
output_types, _, _ = _normalize_output_metadata(node_class)
for output_index, io_type in enumerate(output_types):
if _is_transportable_output_type(str(io_type)):
continue
source = LinkedOutputRef(node_id=node_id, output_index=output_index)
for target in consumers.get(source, []):
target_node_id = str(target.node_id)
if (
target_node_id in expanded_node_ids
or target_node_id in remote_node_ids
):
continue
if target_node_id not in prompt:
continue
if _is_non_returning_tap_terminal_node(
prompt_node=prompt.get(target_node_id),
nodes_module=nodes_module,
):
continue
expanded_node_ids.add(target_node_id)
added_node_ids.add(target_node_id)
changed = True
logger.info(
"Absorbing local node %s because it consumes non-transportable output %s:%d from a preview-tap remote component.",
target_node_id,
source.node_id,
source.output_index,
)
return expanded_node_ids, added_node_ids
def _remote_output_io_type(
*,
prompt: dict[str, Any],
node_id: str,
output_index: int,
nodes_module: Any,
) -> str | None:
"""Return the declared output type for one prompt node output when available."""
prompt_node = prompt.get(node_id)
if prompt_node is None:
return None
class_type = str(prompt_node["class_type"])
node_class = nodes_module.NODE_CLASS_MAPPINGS.get(class_type)
if node_class is None:
return None
output_types, _, _ = _normalize_output_metadata(node_class)
if output_index < 0 or output_index >= len(output_types):
return None
return str(output_types[output_index])
def _remote_output_is_list(
*,
prompt: dict[str, Any],
node_id: str,
output_index: int,
nodes_module: Any,
) -> bool:
"""Return whether one prompt node output is declared as a list output."""
prompt_node = prompt.get(node_id)
if prompt_node is None:
return False
class_type = str(prompt_node["class_type"])
node_class = nodes_module.NODE_CLASS_MAPPINGS.get(class_type)
if node_class is None:
return False
output_types, _output_names, output_is_list = _normalize_output_metadata(node_class)
if output_index < 0 or output_index >= len(output_types):
return False
return bool(output_is_list[output_index])
def _remote_group_dependency_edges(
prompt: Mapping[str, Any],
remote_node_ids: set[str],
group_for_node: Callable[[str], str],
) -> dict[str, set[str]]:
"""Return current coarse dependency edges for a remote-node grouping."""
dependency_edges: dict[str, set[str]] = defaultdict(set)
for downstream_node_id in remote_node_ids:
downstream_node = prompt.get(downstream_node_id)
if not isinstance(downstream_node, Mapping):
continue
downstream_group = group_for_node(downstream_node_id)
for input_value in (downstream_node.get("inputs") or {}).values():
if not _is_link(input_value):
continue
upstream_node_id = str(input_value[0])
if upstream_node_id not in remote_node_ids:
continue
upstream_group = group_for_node(upstream_node_id)
if upstream_group != downstream_group:
dependency_edges[upstream_group].add(downstream_group)
return dependency_edges
def _has_alternate_group_path_through_protected_group(
dependency_edges: Mapping[str, set[str]],
start_group: str,
target_group: str,
protected_groups: set[str],
) -> bool:
"""Return whether a non-direct path crosses a provider-constrained group."""
pending = [
(downstream_group, downstream_group in protected_groups)
for downstream_group in dependency_edges.get(start_group, set())
if downstream_group != target_group
]
visited: set[tuple[str, bool]] = set()
while pending:
current_group, crossed_protected_group = pending.pop()
if current_group == target_group and crossed_protected_group:
return True
state = (current_group, crossed_protected_group)
if state in visited:
continue
visited.add(state)
pending.extend(
(
downstream_group,
crossed_protected_group or downstream_group in protected_groups,
)
for downstream_group in dependency_edges.get(current_group, set())
)
return False
def _remote_component_partition_groups(
prompt: dict[str, Any],
remote_node_ids: set[str],
consumers: dict[LinkedOutputRef, list[InputTarget]],
nodes_module: Any,
*,
preserve_nontransportable_affinity: bool = True,
) -> dict[str, set[str]]:
"""Return component groups after merging remote nodes across costly boundaries."""
parent: dict[str, str] = {node_id: node_id for node_id in remote_node_ids}
downstream_remote_node_ids_by_node_id: dict[str, set[str]] = defaultdict(set)
provider_constrained_node_ids = {
node_id
for node_id in remote_node_ids
if isinstance((prompt_node := prompt.get(node_id)), Mapping)
and _prompt_node_required_provider(prompt_node) is not None
}
def find(node_id: str) -> str:
"""Return the canonical union-find representative for one remote node."""
root = parent[node_id]
while root != parent[root]:
root = parent[root]
while node_id != root:
next_node_id = parent[node_id]
parent[node_id] = root
node_id = next_node_id
return root
def union(
left_node_id: str,
right_node_id: str,
*,
preserve_coarse_dag: bool | None = None,
) -> bool:
"""Merge two remote nodes unless doing so creates a coarse dependency cycle."""
left_root = find(left_node_id)
right_root = find(right_node_id)
if left_root == right_root:
return True
should_preserve_coarse_dag = (
bool(provider_constrained_node_ids)
if preserve_coarse_dag is None
else preserve_coarse_dag
)
if should_preserve_coarse_dag:
dependency_edges = _remote_group_dependency_edges(
prompt,
remote_node_ids,
find,
)
protected_groups = {
find(node_id) for node_id in provider_constrained_node_ids
}
creates_cycle = _has_alternate_group_path_through_protected_group(
dependency_edges,
left_root,
right_root,
protected_groups,
) or _has_alternate_group_path_through_protected_group(
dependency_edges,
right_root,
left_root,
protected_groups,
)
if creates_cycle:
logger.info(
"Splitting remote nodes %s and %s to preserve an acyclic "
"component graph.",
left_node_id,
right_node_id,
)
return False
canonical_root = min(left_root, right_root)
merged_root = max(left_root, right_root)
parent[merged_root] = canonical_root
return True
for node_id in sorted(remote_node_ids):
prompt_node = prompt.get(node_id)
if prompt_node is None:
continue
for input_value in (prompt_node.get("inputs") or {}).values():
if not _is_link(input_value):
continue
upstream_node_id = str(input_value[0])
if upstream_node_id not in remote_node_ids:
continue
downstream_remote_node_ids_by_node_id[upstream_node_id].add(node_id)
upstream_prompt_node = prompt.get(upstream_node_id)
source = LinkedOutputRef(
node_id=upstream_node_id, output_index=int(input_value[1])
)
if (
upstream_prompt_node is not None
and str(upstream_prompt_node.get("class_type"))
== MODAL_MAP_INPUT_NODE_ID
):
union(node_id, upstream_node_id, preserve_coarse_dag=False)
continue
io_type = _remote_output_io_type(
prompt=prompt,
node_id=upstream_node_id,
output_index=source.output_index,
nodes_module=nodes_module,
)
if (
preserve_nontransportable_affinity
and io_type is not None
and not _is_transportable_output_type(io_type)
):
logger.info(
"Co-locating remote nodes %s -> %s because %s output %d "
"cannot cross a component boundary.",
upstream_node_id,
node_id,
io_type,
source.output_index,
)
union(node_id, upstream_node_id, preserve_coarse_dag=False)
continue
downstream_provider = _prompt_node_required_provider(prompt_node)
upstream_provider = (
_prompt_node_required_provider(upstream_prompt_node)
if isinstance(upstream_prompt_node, Mapping)
else None
)
if downstream_provider != upstream_provider and (
downstream_provider is not None or upstream_provider is not None
):
logger.info(
"Splitting remote nodes %s -> %s across provider boundary %s -> %s.",
upstream_node_id,
node_id,
upstream_provider.value if upstream_provider is not None else "any",
downstream_provider.value
if downstream_provider is not None
else "any",
)
continue
if _output_supports_parallel_local_materialization(
prompt=prompt,
source=source,
remote_node_ids=remote_node_ids,
consumers=consumers,
nodes_module=nodes_module,
):
logger.info(
"Splitting remote nodes %s -> %s because output %d also feeds only non-returning local work.",
upstream_node_id,
node_id,
source.output_index,
)
continue
if io_type is not None and _is_inexpensive_remote_boundary_type(io_type):
continue
if io_type is not None and _is_transportable_output_type(io_type):
logger.info(
"Co-locating remote nodes %s -> %s because %s output %d is "
"transportable but expensive.",
upstream_node_id,
node_id,
io_type,
source.output_index,
)
union(node_id, upstream_node_id)
for remote_node_id in sorted(remote_node_ids):
prompt_node = prompt.get(remote_node_id)
if (
prompt_node is None
or str(prompt_node.get("class_type")) != MODAL_MAP_INPUT_NODE_ID
):
continue
pending_node_ids = [remote_node_id]
visited_node_ids: set[str] = set()
while pending_node_ids:
current_node_id = pending_node_ids.pop()
if current_node_id in visited_node_ids:
continue
visited_node_ids.add(current_node_id)
union(
remote_node_id,
current_node_id,
preserve_coarse_dag=False,
)
for downstream_node_id in sorted(
downstream_remote_node_ids_by_node_id.get(current_node_id, set())
):
pending_node_ids.append(downstream_node_id)
groups: dict[str, set[str]] = defaultdict(set)
for node_id in sorted(remote_node_ids):
groups[find(node_id)].add(node_id)
return groups
def _component_topological_order(
prompt: dict[str, Any],
component_groups: dict[str, set[str]],
) -> list[list[str]]:
"""Return component node ids ordered from upstream to downstream."""
_, dependency_edges, indegree_by_component_id = _component_dependency_graph(
prompt, component_groups
)
merged_component_groups = _merge_cyclic_component_groups(
component_groups=component_groups,
dependency_edges=dependency_edges,
)
if merged_component_groups != component_groups:
return _component_topological_order(prompt, merged_component_groups)
ready_component_ids = deque(
sorted(
[
component_id
for component_id, indegree in indegree_by_component_id.items()
if indegree == 0
]
)
)
ordered_components: list[list[str]] = []
emitted_component_ids: set[str] = set()
while ready_component_ids:
component_id = ready_component_ids.popleft()
if component_id in emitted_component_ids:
continue
emitted_component_ids.add(component_id)
ordered_components.append(sorted(component_groups[component_id]))
for downstream_component_id in sorted(dependency_edges[component_id]):
indegree_by_component_id[downstream_component_id] -= 1
if indegree_by_component_id[downstream_component_id] == 0:
ready_component_ids.append(downstream_component_id)
if len(emitted_component_ids) == len(component_groups):
return ordered_components
logger.warning(
"Transport-aware component ordering encountered a cycle or unresolved dependency; falling back to stable component order."
)
for component_id in sorted(component_groups):
if component_id in emitted_component_ids:
continue
ordered_components.append(sorted(component_groups[component_id]))
return ordered_components
def _component_dependency_graph(
prompt: dict[str, Any],
component_groups: dict[str, set[str]],
) -> tuple[dict[str, str], dict[str, set[str]], dict[str, int]]:
"""Return component membership, downstream edges, and indegrees for the coarse component DAG."""
component_id_by_node_id: dict[str, str] = {}
for representative_node_id, component_node_ids in component_groups.items():
for node_id in component_node_ids:
component_id_by_node_id[node_id] = representative_node_id
dependency_edges: dict[str, set[str]] = {
representative_node_id: set() for representative_node_id in component_groups
}
indegree_by_component_id: dict[str, int] = {
representative_node_id: 0 for representative_node_id in component_groups
}
for node_id, representative_node_id in component_id_by_node_id.items():
prompt_node = prompt.get(node_id)
if prompt_node is None:
continue
for input_value in (prompt_node.get("inputs") or {}).values():
if not _is_link(input_value):
continue
upstream_node_id = str(input_value[0])
upstream_component_id = component_id_by_node_id.get(upstream_node_id)
if (
upstream_component_id is None
or upstream_component_id == representative_node_id
):
continue
if representative_node_id in dependency_edges[upstream_component_id]:
continue
dependency_edges[upstream_component_id].add(representative_node_id)
indegree_by_component_id[representative_node_id] += 1
return component_id_by_node_id, dependency_edges, indegree_by_component_id
def _component_execution_stages(
prompt: dict[str, Any],
component_groups: dict[str, set[str]],
) -> list[list[str]]:
"""Return one best-effort stage decomposition for concurrent remote component execution."""
_, dependency_edges, indegree_by_component_id = _component_dependency_graph(
prompt, component_groups
)
merged_component_groups = _merge_cyclic_component_groups(
component_groups=component_groups,
dependency_edges=dependency_edges,
)
if merged_component_groups != component_groups:
return _component_execution_stages(prompt, merged_component_groups)
remaining_indegrees = dict(indegree_by_component_id)
ready_component_ids = sorted(
[
component_id
for component_id, indegree in remaining_indegrees.items()
if indegree == 0
]
)
execution_stages: list[list[str]] = []
emitted_component_ids: set[str] = set()
while ready_component_ids:
current_stage = list(ready_component_ids)
execution_stages.append(current_stage)
next_ready_component_ids: set[str] = set()
for component_id in current_stage:
emitted_component_ids.add(component_id)
for downstream_component_id in sorted(
dependency_edges.get(component_id, set())
):
remaining_indegrees[downstream_component_id] -= 1
if remaining_indegrees[downstream_component_id] == 0:
next_ready_component_ids.add(downstream_component_id)
ready_component_ids = sorted(
component_id
for component_id in next_ready_component_ids
if component_id not in emitted_component_ids
)
if len(emitted_component_ids) == len(component_groups):
return execution_stages
fallback_stage = [
component_id
for component_id in sorted(component_groups)
if component_id not in emitted_component_ids
]
if fallback_stage:
logger.warning(
"Transport-aware execution-stage planning encountered a cycle or unresolved dependency; appending fallback stage %s.",
fallback_stage,
)
execution_stages.append(fallback_stage)
return execution_stages
def _estimated_stage_parallelism(
execution_stages: list[list[str]],
mapped_component_ids: set[str],
*,
mapped_component_weight: int,
max_parallelism_cap: int | None = None,
) -> int:
"""Return the weighted best-effort parallelism estimate over staged remote execution."""
stage_parallelism = 0
for stage in execution_stages:
current_stage_parallelism = sum(
mapped_component_weight if component_id in mapped_component_ids else 1
for component_id in stage
)
stage_parallelism = max(stage_parallelism, current_stage_parallelism)
if max_parallelism_cap is not None:
return min(stage_parallelism, max_parallelism_cap)
return stage_parallelism
def _merge_cyclic_component_groups(
*,
component_groups: dict[str, set[str]],
dependency_edges: dict[str, set[str]],
) -> dict[str, set[str]]:
"""Collapse cyclic coarse component groups into SCC-merged groups."""
component_ids = sorted(component_groups)
reverse_edges: dict[str, set[str]] = {
component_id: set() for component_id in component_ids
}
for upstream_component_id, downstream_component_ids in dependency_edges.items():
for downstream_component_id in downstream_component_ids:
reverse_edges.setdefault(downstream_component_id, set()).add(
upstream_component_id
)
visited_component_ids: set[str] = set()
finish_order: list[str] = []
def visit_forward(component_id: str) -> None:
"""Record reverse-topological finish order over the coarse graph."""
if component_id in visited_component_ids:
return
visited_component_ids.add(component_id)
for downstream_component_id in sorted(
dependency_edges.get(component_id, set())
):
visit_forward(downstream_component_id)
finish_order.append(component_id)
for component_id in component_ids:
visit_forward(component_id)
assigned_component_ids: set[str] = set()
merged_groups: dict[str, set[str]] = {}
merged_sccs = 0
def visit_reverse(component_id: str, scc_component_ids: set[str]) -> None:
"""Collect one SCC by walking reverse edges from the finish-order seed."""
if component_id in assigned_component_ids:
return
assigned_component_ids.add(component_id)
scc_component_ids.add(component_id)
for upstream_component_id in sorted(reverse_edges.get(component_id, set())):
visit_reverse(upstream_component_id, scc_component_ids)
for component_id in reversed(finish_order):
if component_id in assigned_component_ids:
continue
scc_component_ids: set[str] = set()
visit_reverse(component_id, scc_component_ids)
merged_node_ids: set[str] = set()
for scc_component_id in scc_component_ids:
merged_node_ids.update(component_groups[scc_component_id])
representative_node_id = min(merged_node_ids)
merged_groups[representative_node_id] = merged_node_ids
if len(scc_component_ids) > 1:
merged_sccs += 1
if merged_sccs:
logger.warning(
"Transport-aware coarse component graph contained %d cyclic SCC(s); merging them back into larger remote components.",
merged_sccs,
)
return merged_groups
def _build_remote_components(
prompt: dict[str, Any],
remote_node_ids: set[str],
consumers: dict[LinkedOutputRef, list[InputTarget]],
nodes_module: Any,
) -> list[list[str]]:
"""Partition remote-marked nodes into transport-aware DAG components."""
component_groups = _remote_component_partition_groups(
prompt,
remote_node_ids,
consumers,
nodes_module,
)
components = _component_topological_order(prompt, component_groups)
logger.info(
"Partitioned %d remote nodes into %d transport-aware remote components: %s",
len(remote_node_ids),
len(components),
components,
)
return components
def _expand_remote_node_ids_for_non_transportable_inputs(
prompt: dict[str, Any],
remote_node_ids: set[str],
nodes_module: Any,
) -> tuple[set[str], list[RemoteExpansionReason]]:
"""Grow the remote set upstream until non-transportable inputs stay inside the remote island."""
expanded_remote_node_ids = set(remote_node_ids)
added_node_ids: set[str] = set()
reasons: list[RemoteExpansionReason] = []
changed = True
while changed:
changed = False
for node_id in sorted(expanded_remote_node_ids):
prompt_node = prompt.get(node_id)
if prompt_node is None:
continue
for input_value in (prompt_node.get("inputs") or {}).values():
if not _is_link(input_value):
continue
upstream_node_id = str(input_value[0])
if upstream_node_id in expanded_remote_node_ids:
continue
upstream_prompt_node = prompt.get(upstream_node_id)
if upstream_prompt_node is None:
continue
upstream_class_type = str(upstream_prompt_node["class_type"])
upstream_class = nodes_module.NODE_CLASS_MAPPINGS.get(
upstream_class_type
)
if upstream_class is None:
continue
output_types, _, _ = _normalize_output_metadata(upstream_class)
output_index = int(input_value[1])
if output_index >= len(output_types):
continue
io_type = str(output_types[output_index])
if _is_transportable_output_type(io_type):
continue
expanded_remote_node_ids.add(upstream_node_id)
reason = RemoteExpansionReason(
node_id=upstream_node_id,
class_type=upstream_class_type,
required_by_node_id=node_id,
required_by_class_type=str(prompt_node["class_type"]),
output_index=output_index,
io_type=io_type,
)
reasons.append(reason)
added_node_ids.add(upstream_node_id)
changed = True
logger.info(
"Auto-expanded remote execution upstream: added node %s (%s) because node %s depends on non-transportable type '%s'.",
reason.node_id,
reason.class_type,
reason.required_by_node_id,
reason.io_type,
)
break
if changed:
break
if added_node_ids:
logger.info(
"Expanded remote node set from %d to %d nodes by absorbing upstream non-transportable dependencies: %s",
len(remote_node_ids),
len(expanded_remote_node_ids),
sorted(added_node_ids),
)
return expanded_remote_node_ids, reasons
def _terminal_remote_video_source(
*,
node_id: str,
prompt: dict[str, Any],
remote_node_ids: set[str],
consumers: dict[LinkedOutputRef, list[InputTarget]],
nodes_module: Any,
) -> LinkedOutputRef | None:
"""Return the remote VIDEO feeding one safe terminal SaveVideo artifact sink."""
prompt_node = prompt.get(node_id)
if prompt_node is None or str(prompt_node.get("class_type")) != "SaveVideo":
return None
inputs = prompt_node.get("inputs") or {}
video_input = inputs.get("video")
if not _is_link(video_input):
return None
video_source = LinkedOutputRef(str(video_input[0]), int(video_input[1]))
linked_source_node_ids = {
str(input_value[0]) for input_value in inputs.values() if _is_link(input_value)
}
if not linked_source_node_ids.issubset(remote_node_ids):
return None
if any(
consumers.get(output_ref)
for output_ref in _node_output_refs(prompt, node_id, nodes_module)
):
return None
io_type = _remote_output_io_type(
prompt=prompt,
node_id=video_source.node_id,
output_index=video_source.output_index,
nodes_module=nodes_module,
)
return video_source if io_type == "VIDEO" else None
def _expand_remote_node_ids_for_terminal_video_sinks(
prompt: dict[str, Any],
remote_node_ids: set[str],
nodes_module: Any,
) -> set[str]:
"""Run terminal SaveVideo consumers remotely to avoid exporting raw VIDEO tensors."""
expanded_remote_node_ids = set(remote_node_ids)
consumers = _build_consumer_map(prompt)
added_node_ids: set[str] = set()
for node_id in sorted(prompt):
if node_id in expanded_remote_node_ids:
continue
video_source = _terminal_remote_video_source(
node_id=node_id,
prompt=prompt,
remote_node_ids=expanded_remote_node_ids,
consumers=consumers,
nodes_module=nodes_module,
)
if video_source is None:
continue
expanded_remote_node_ids.add(node_id)
added_node_ids.add(node_id)
logger.info(
"Auto-expanded terminal SaveVideo node %s into remote execution because its "
"VIDEO input comes from remote node %s; the encoded output artifact will be "
"materialized locally without exporting raw frames.",
node_id,
video_source.node_id,
)
if added_node_ids:
logger.info(
"Expanded remote node set with terminal video artifact sinks: %s.",
sorted(added_node_ids),
)
return expanded_remote_node_ids
def analyze_remote_node_selection(
prompt: dict[str, Any],
workflow: dict[str, Any] | None,
seed_workflow_node_paths: list[str],
settings: ModalSyncSettings | None = None,
nodes_module: Any | None = None,
) -> RemoteNodeAnalysis:
"""Return the nodes the UI should mark remote for one context-menu expansion request."""
resolved_settings = settings or get_settings()
resolved_nodes_module = nodes_module or _get_nodes_module()
prompt_node_ids = {str(node_id) for node_id in prompt.keys()}
requested_workflow_node_paths = {
str(seed_workflow_node_path)
for seed_workflow_node_path in seed_workflow_node_paths
if str(seed_workflow_node_path)
}
current_remote_node_ids = extract_remote_node_ids(
workflow,
resolved_settings,
prompt_node_ids,
)
current_remote_workflow_node_paths = _extract_marked_workflow_node_paths(
workflow,
resolved_settings,
)
(
workflow_path_to_prompt_node_ids,
prompt_node_id_to_workflow_paths,
) = _build_workflow_prompt_resolution_maps(workflow, prompt_node_ids)
requested_node_ids = _resolve_requested_prompt_node_ids(
requested_workflow_node_paths,
prompt_node_ids,
workflow_path_to_prompt_node_ids,
)
initial_remote_node_ids = current_remote_node_ids | requested_node_ids
(
resolved_remote_node_ids,
reasons,
) = _expand_remote_node_ids_for_non_transportable_inputs(
prompt=prompt,
remote_node_ids=initial_remote_node_ids,
nodes_module=resolved_nodes_module,
)
resolved_remote_node_ids = _expand_remote_node_ids_for_terminal_video_sinks(
prompt=prompt,
remote_node_ids=resolved_remote_node_ids,
nodes_module=resolved_nodes_module,
)
sandwiched_local_node_ids = _sandwiched_local_node_ids(
prompt,
resolved_remote_node_ids,
)
resolved_workflow_node_paths = (
_resolve_workflow_node_paths_for_prompt_nodes(
resolved_remote_node_ids,
prompt_node_id_to_workflow_paths,
)
| current_remote_workflow_node_paths
)
added_node_ids = resolved_remote_node_ids - current_remote_node_ids
added_workflow_node_paths = (
resolved_workflow_node_paths - current_remote_workflow_node_paths
)
return RemoteNodeAnalysis(
requested_node_ids=sorted(requested_node_ids),
requested_workflow_node_paths=sorted(requested_workflow_node_paths),
current_remote_node_ids=sorted(current_remote_node_ids),
current_remote_workflow_node_paths=sorted(current_remote_workflow_node_paths),
resolved_remote_node_ids=sorted(resolved_remote_node_ids),
resolved_workflow_node_paths=sorted(resolved_workflow_node_paths),
added_node_ids=sorted(added_node_ids),
added_workflow_node_paths=sorted(added_workflow_node_paths),
sandwiched_local_node_ids=sorted(sandwiched_local_node_ids),
reasons=reasons,
)
def _build_component_plan(
component_node_ids: list[str],
prompt: dict[str, Any],
consumers: dict[LinkedOutputRef, list[InputTarget]],
remote_node_ids: set[str],
nodes_module: Any,
) -> RemoteComponentPlan:
"""Build rewrite metadata for a connected remote component."""
original_component_node_id_set = set(component_node_ids)
representative_node_id = component_node_ids[0]
local_tap_node_ids, local_tap_terminal_node_ids = _non_returning_local_tap_node_ids(
prompt=prompt,
component_node_ids=original_component_node_id_set,
remote_node_ids=remote_node_ids,
consumers=consumers,
nodes_module=nodes_module,
)
if local_tap_node_ids:
logger.info(
"Absorbing non-returning local tap nodes into remote component %s: %s",
representative_node_id,
sorted(local_tap_node_ids),
)
if local_tap_terminal_node_ids:
logger.info(
"Keeping non-returning local tap terminal nodes outside remote component %s: %s",
representative_node_id,
sorted(local_tap_terminal_node_ids),
)
component_node_id_set = original_component_node_id_set | local_tap_node_ids
if local_tap_node_ids:
(
expanded_component_node_id_set,
tap_dependency_node_ids,
) = _expand_component_for_non_transportable_local_outputs(
prompt=prompt,
component_node_ids=component_node_id_set,
remote_node_ids=remote_node_ids,
consumers=consumers,
nodes_module=nodes_module,
)
if tap_dependency_node_ids:
logger.info(
"Absorbing non-transportable boundary dependencies for remote component %s preview taps: %s",
representative_node_id,
sorted(tap_dependency_node_ids),
)
local_tap_node_ids.update(tap_dependency_node_ids)
component_node_id_set = expanded_component_node_id_set
component_node_ids = sorted(component_node_id_set)
boundary_inputs_by_source: dict[LinkedOutputRef, BoundaryInputSpec] = {}
boundary_outputs_by_source: dict[LinkedOutputRef, BoundaryOutputSpec] = {}
output_execution_targets: set[str] = set()
contains_output_node = False
for node_id in component_node_ids:
prompt_node = prompt[node_id]
class_type = str(prompt_node["class_type"])
node_class = nodes_module.NODE_CLASS_MAPPINGS[class_type]
output_types, output_names, output_is_list = _normalize_output_metadata(
node_class
)
if getattr(node_class, "OUTPUT_NODE", False):
contains_output_node = True
output_execution_targets.add(node_id)
for input_name, input_value in (prompt_node.get("inputs") or {}).items():
if not _is_link(input_value):
continue
upstream_node_id = str(input_value[0])
if upstream_node_id in component_node_id_set:
continue
source = LinkedOutputRef(
node_id=upstream_node_id, output_index=int(input_value[1])
)
spec = boundary_inputs_by_source.get(source)
if spec is None:
source_io_type = _remote_output_io_type(
prompt=prompt,
node_id=source.node_id,
output_index=source.output_index,
nodes_module=nodes_module,
)
spec = BoundaryInputSpec(
proxy_input_name=f"remote_input_{len(boundary_inputs_by_source)}",
source=source,
io_type=source_io_type,
)
boundary_inputs_by_source[source] = spec
spec.targets.append(
InputTarget(node_id=node_id, input_name=str(input_name))
)
has_downstream_consumer = False
for output_index, io_type in enumerate(output_types):
source = LinkedOutputRef(node_id=node_id, output_index=output_index)
output_consumers = consumers.get(source, [])
if output_consumers:
has_downstream_consumer = True
local_consumers = [
consumer
for consumer in output_consumers
if consumer.node_id not in component_node_id_set
]
if not local_consumers:
continue
output_execution_targets.add(node_id)
if source in boundary_outputs_by_source:
continue
output_name = output_names[output_index]
boundary_outputs_by_source[source] = BoundaryOutputSpec(
proxy_output_name=f"{node_id}_{output_name}",
source=source,
io_type=str(io_type),
is_list=bool(output_is_list[output_index]),
preview_target_node_ids=_preview_target_node_ids(
prompt=prompt,
local_consumers=local_consumers,
),
)
if not has_downstream_consumer:
output_execution_targets.add(node_id)
mapped_boundary_spec: BoundaryInputSpec | None = None
mapped_boundary_input_io_type: str | None = None
for boundary_input in boundary_inputs_by_source.values():
source_prompt_node = prompt.get(boundary_input.source.node_id)
source_class_type = (
str(source_prompt_node.get("class_type"))
if source_prompt_node is not None
else None
)
mapped_targets = [
target
for target in boundary_input.targets
if str(prompt[target.node_id]["class_type"]) == MODAL_MAP_INPUT_NODE_ID
]
source_is_modal_map_input = source_class_type == MODAL_MAP_INPUT_NODE_ID
if not mapped_targets and not source_is_modal_map_input:
continue
if mapped_targets and len(mapped_targets) != len(boundary_input.targets):
raise ModalPromptValidationError(
"Mapped remote execution requires the mapped boundary input to feed only ModalMapInput nodes."
)
if mapped_boundary_spec is not None:
raise ModalPromptValidationError(
"Remote components currently support only one mapped ModalMapInput boundary."
)
mapped_boundary_spec = boundary_input
mapped_boundary_input_io_type = _mapped_boundary_origin_io_type(
prompt,
boundary_input,
nodes_module,
)
mapped_node_ids: list[str] = []
mapped_execute_node_ids: list[str] = []
static_execute_node_ids: list[str] = []
static_node_ids: list[str] = []
static_to_mapped_boundaries: list[StaticToMappedBoundarySpec] = []
if mapped_boundary_spec is not None:
mapped_reachable_node_ids = _component_downstream_closure(
seed_node_ids={target.node_id for target in mapped_boundary_spec.targets},
component_node_id_set=component_node_id_set,
consumers=consumers,
)
mapped_node_ids = sorted(mapped_reachable_node_ids)
mapped_node_id_set = set(mapped_node_ids)
static_node_ids = sorted(component_node_id_set - mapped_node_id_set)
static_to_mapped_boundaries_by_source: dict[
LinkedOutputRef, StaticToMappedBoundarySpec
] = {}
for mapped_node_id in mapped_node_ids:
prompt_node = prompt[mapped_node_id]
for input_name, input_value in (prompt_node.get("inputs") or {}).items():
if not _is_link(input_value):
continue
upstream_node_id = str(input_value[0])
if (
upstream_node_id not in component_node_id_set
or upstream_node_id in mapped_node_id_set
):
continue
source = LinkedOutputRef(
node_id=upstream_node_id,
output_index=int(input_value[1]),
)
boundary_spec = static_to_mapped_boundaries_by_source.get(source)
if boundary_spec is None:
boundary_spec = StaticToMappedBoundarySpec(
proxy_name=f"static_input_{len(static_to_mapped_boundaries_by_source)}",
source=source,
io_type=str(
_remote_output_io_type(
prompt=prompt,
node_id=source.node_id,
output_index=source.output_index,
nodes_module=nodes_module,
)
or "*"
),
is_list=_remote_output_is_list(
prompt=prompt,
node_id=source.node_id,
output_index=source.output_index,
nodes_module=nodes_module,
),
)
static_to_mapped_boundaries_by_source[source] = boundary_spec
boundary_spec.targets.append(
InputTarget(node_id=mapped_node_id, input_name=str(input_name))
)
static_to_mapped_boundaries = sorted(
static_to_mapped_boundaries_by_source.values(),
key=lambda spec: (spec.source.node_id, spec.source.output_index),
)
mapped_execute_node_ids = sorted(output_execution_targets & mapped_node_id_set)
static_execute_node_ids = sorted(
(output_execution_targets - mapped_node_id_set)
| {
boundary_spec.source.node_id
for boundary_spec in static_to_mapped_boundaries
}
)
component = RemoteComponentPlan(
node_ids=component_node_ids,
representative_node_id=representative_node_id,
boundary_inputs=sorted(
boundary_inputs_by_source.values(),
key=lambda spec: (spec.source.node_id, spec.source.output_index),
),
boundary_outputs=sorted(
boundary_outputs_by_source.values(),
key=lambda spec: (spec.source.node_id, spec.source.output_index),
),
execute_node_ids=sorted(output_execution_targets),
contains_output_node=contains_output_node,
mapped_boundary_input_name=(
mapped_boundary_spec.proxy_input_name
if mapped_boundary_spec is not None
else None
),
mapped_boundary_input_io_type=mapped_boundary_input_io_type,
mapped_boundary_source_node_id=(
mapped_boundary_spec.source.node_id
if mapped_boundary_spec is not None
and prompt.get(mapped_boundary_spec.source.node_id, {}).get("class_type")
== MODAL_MAP_INPUT_NODE_ID
else None
),
static_node_ids=static_node_ids,
mapped_node_ids=mapped_node_ids,
mapped_execute_node_ids=mapped_execute_node_ids,
static_execute_node_ids=static_execute_node_ids,
static_to_mapped_boundaries=static_to_mapped_boundaries,
local_tap_node_ids=sorted(local_tap_node_ids),
local_tap_terminal_node_ids=sorted(local_tap_terminal_node_ids),
)
logger.info(
"Planned remote component %s: nodes=%s boundary_inputs=%d boundary_outputs=%d execute_nodes=%s output_node=%s mapped_input=%s static_nodes=%s mapped_nodes=%s mapped_execute_nodes=%s static_execute_nodes=%s local_tap_nodes=%s local_tap_terminal_nodes=%s static_to_mapped_boundaries=%s",
component.representative_node_id,
component.node_ids,
len(component.boundary_inputs),
len(component.boundary_outputs),
component.execute_node_ids,
component.contains_output_node,
component.mapped_boundary_input_name,
component.static_node_ids,
component.mapped_node_ids,
component.mapped_execute_node_ids,
component.static_execute_node_ids,
component.local_tap_node_ids,
component.local_tap_terminal_node_ids,
[
{
"proxy_name": boundary_spec.proxy_name,
"source": (
boundary_spec.source.node_id,
boundary_spec.source.output_index,
),
"targets": [
(target.node_id, target.input_name)
for target in boundary_spec.targets
],
}
for boundary_spec in component.static_to_mapped_boundaries
],
)
return component
def _filter_boundary_inputs_for_node_ids(
boundary_inputs: list[BoundaryInputSpec],
allowed_node_ids: set[str],
) -> list[BoundaryInputSpec]:
"""Return boundary inputs whose targets belong to one node-id subset."""
filtered_boundary_inputs: list[BoundaryInputSpec] = []
for boundary_input in boundary_inputs:
filtered_targets = [
target
for target in boundary_input.targets
if target.node_id in allowed_node_ids
]
if not filtered_targets:
continue
filtered_boundary_inputs.append(
BoundaryInputSpec(
proxy_input_name=boundary_input.proxy_input_name,
source=boundary_input.source,
io_type=boundary_input.io_type,
targets=filtered_targets,
)
)
return filtered_boundary_inputs
def _filter_boundary_outputs_for_node_ids(
boundary_outputs: list[BoundaryOutputSpec],
allowed_node_ids: set[str],
) -> list[BoundaryOutputSpec]:
"""Return boundary outputs exported by one node-id subset."""
return [
boundary_output
for boundary_output in boundary_outputs
if boundary_output.source.node_id in allowed_node_ids
]
def _subset_component_prompt(
component_prompt: dict[str, Any],
node_ids: list[str],
) -> dict[str, Any]:
"""Return one deep-copied prompt subset for a phase-local node set."""
return {
node_id: copy.deepcopy(component_prompt[node_id])
for node_id in node_ids
if node_id in component_prompt
}
def _preview_target_node_ids(
*,
prompt: dict[str, Any],
local_consumers: list[InputTarget],
) -> list[str]:
"""Return direct local PreviewImage consumers for one remote boundary output."""
preview_target_node_ids: set[str] = set()
for local_consumer in local_consumers:
consumer_prompt_node = prompt.get(local_consumer.node_id)
if consumer_prompt_node is None:
continue
if str(consumer_prompt_node.get("class_type")) != "PreviewImage":
continue
preview_target_node_ids.add(str(local_consumer.node_id))
return sorted(preview_target_node_ids)
def _component_downstream_closure(
*,
seed_node_ids: set[str],
component_node_id_set: set[str],
consumers: dict[LinkedOutputRef, list[InputTarget]],
) -> set[str]:
"""Return component-local nodes reachable downstream from one seed set."""
reachable_node_ids: set[str] = set()
pending_node_ids = list(sorted(seed_node_ids))
while pending_node_ids:
current_node_id = pending_node_ids.pop()
if (
current_node_id in reachable_node_ids
or current_node_id not in component_node_id_set
):
continue
reachable_node_ids.add(current_node_id)
for consumer_source, consumer_targets in consumers.items():
if consumer_source.node_id != current_node_id:
continue
for consumer_target in consumer_targets:
if consumer_target.node_id in component_node_id_set:
pending_node_ids.append(consumer_target.node_id)
return reachable_node_ids
def _component_upstream_closure(
*,
prompt: dict[str, Any],
seed_node_ids: set[str],
candidate_node_ids: set[str],
) -> set[str]:
"""Return one candidate-local upstream closure for the supplied seed nodes."""
reachable_node_ids: set[str] = set()
pending_node_ids = list(sorted(seed_node_ids))
while pending_node_ids:
current_node_id = pending_node_ids.pop()
if (
current_node_id in reachable_node_ids
or current_node_id not in candidate_node_ids
):
continue
reachable_node_ids.add(current_node_id)
prompt_node = prompt.get(current_node_id)
if prompt_node is None:
continue
for input_value in (prompt_node.get("inputs") or {}).values():
if not _is_link(input_value):
continue
upstream_node_id = str(input_value[0])
if upstream_node_id in candidate_node_ids:
pending_node_ids.append(upstream_node_id)
return reachable_node_ids
def _component_ancestors_of_local_source(
*,
prompt: dict[str, Any],
source_node_id: str,
component_node_id_set: set[str],
) -> set[str]:
"""Return component nodes that feed one local boundary-input source."""
ancestor_node_ids: set[str] = set()
visited_node_ids: set[str] = set()
pending_node_ids = [source_node_id]
while pending_node_ids:
current_node_id = str(pending_node_ids.pop())
if current_node_id in visited_node_ids:
continue
visited_node_ids.add(current_node_id)
if current_node_id in component_node_id_set:
ancestor_node_ids.add(current_node_id)
continue
prompt_node = prompt.get(current_node_id)
if prompt_node is None:
continue
for input_value in (prompt_node.get("inputs") or {}).values():
if _is_link(input_value):
pending_node_ids.append(str(input_value[0]))
return ancestor_node_ids
def _component_has_local_reentry_dependency(
*,
prompt: dict[str, Any],
component: RemoteComponentPlan,
) -> bool:
"""Return whether a component boundary input depends on that same component's output."""
component_node_id_set = set(component.node_ids)
for boundary_input in component.boundary_inputs:
if boundary_input.source.node_id in component_node_id_set:
continue
if _component_ancestors_of_local_source(
prompt=prompt,
source_node_id=boundary_input.source.node_id,
component_node_id_set=component_node_id_set,
):
return True
return False
def _component_has_parallel_local_remote_fanout(
component: RemoteComponentPlan,
) -> bool:
"""Return whether one remote output feeds local work and a remote continuation."""
return any(
boundary_output.local_materializer_node_id is not None
and bool(boundary_output.session_consumer_node_ids)
for boundary_output in component.boundary_outputs
)
def _order_execute_node_ids_for_transportable_splits(
*,
prompt: dict[str, Any],
component_prompt: dict[str, Any],
component_node_ids: set[str],
execute_node_ids: list[str],
) -> list[str]:
"""Return split execute targets ordered by component and local feedback dependencies."""
base_order = list(execute_node_ids)
if len(base_order) <= 1:
return base_order
closure_by_execute_node_id = {
execute_node_id: _component_upstream_closure(
prompt=component_prompt,
seed_node_ids={execute_node_id},
candidate_node_ids=component_node_ids,
)
for execute_node_id in base_order
}
producer_execute_node_ids_by_component_node_id: dict[str, list[str]] = defaultdict(
list
)
for execute_node_id in base_order:
for component_node_id in closure_by_execute_node_id[execute_node_id]:
producer_execute_node_ids_by_component_node_id[component_node_id].append(
execute_node_id
)
base_index_by_execute_node_id = {
execute_node_id: index for index, execute_node_id in enumerate(base_order)
}
dependency_edges: dict[str, set[str]] = {
execute_node_id: set() for execute_node_id in base_order
}
indegree_by_execute_node_id: dict[str, int] = {
execute_node_id: 0 for execute_node_id in base_order
}
def producer_for_component_node(component_node_id: str) -> str | None:
"""Return the earliest execute target that produces a component node."""
producer_execute_node_ids = producer_execute_node_ids_by_component_node_id.get(
component_node_id,
[],
)
if not producer_execute_node_ids:
return None
exact_producers = [
execute_node_id
for execute_node_id in producer_execute_node_ids
if execute_node_id == component_node_id
]
if exact_producers:
return exact_producers[0]
return min(
producer_execute_node_ids,
key=lambda execute_node_id: base_index_by_execute_node_id[execute_node_id],
)
for target_execute_node_id in base_order:
target_closure = closure_by_execute_node_id[target_execute_node_id]
for phase_node_id in sorted(target_closure):
prompt_node = component_prompt.get(phase_node_id)
if prompt_node is None:
continue
for input_value in (prompt_node.get("inputs") or {}).values():
if not _is_link(input_value):
continue
source_node_id = str(input_value[0])
if source_node_id in component_node_ids:
continue
source_ancestor_node_ids = _component_ancestors_of_local_source(
prompt=prompt,
source_node_id=source_node_id,
component_node_id_set=component_node_ids,
)
for source_ancestor_node_id in sorted(source_ancestor_node_ids):
if source_ancestor_node_id in target_closure:
continue
producer_execute_node_id = producer_for_component_node(
source_ancestor_node_id
)
if (
producer_execute_node_id is None
or producer_execute_node_id == target_execute_node_id
or target_execute_node_id
in dependency_edges[producer_execute_node_id]
):
continue
dependency_edges[producer_execute_node_id].add(
target_execute_node_id
)
indegree_by_execute_node_id[target_execute_node_id] += 1
ready_execute_node_ids = [
execute_node_id
for execute_node_id in base_order
if indegree_by_execute_node_id[execute_node_id] == 0
]
ordered_execute_node_ids: list[str] = []
while ready_execute_node_ids:
ready_execute_node_ids.sort(
key=lambda node_id: base_index_by_execute_node_id[node_id]
)
execute_node_id = ready_execute_node_ids.pop(0)
ordered_execute_node_ids.append(execute_node_id)
for downstream_execute_node_id in sorted(
dependency_edges[execute_node_id],
key=lambda node_id: base_index_by_execute_node_id[node_id],
):
indegree_by_execute_node_id[downstream_execute_node_id] -= 1
if indegree_by_execute_node_id[downstream_execute_node_id] == 0:
ready_execute_node_ids.append(downstream_execute_node_id)
if len(ordered_execute_node_ids) != len(base_order):
logger.warning(
"Split phase local-feedback dependency ordering encountered a cycle; keeping base execute order %s.",
base_order,
)
return base_order
if ordered_execute_node_ids != base_order:
logger.info(
"Reordered split execute targets from %s to %s using local-feedback dependencies %s.",
base_order,
ordered_execute_node_ids,
{
execute_node_id: sorted(downstream_execute_node_ids)
for execute_node_id, downstream_execute_node_ids in dependency_edges.items()
if downstream_execute_node_ids
},
)
return ordered_execute_node_ids
def _subgraph_topological_node_order(
prompt: dict[str, Any],
node_ids: set[str],
) -> list[str]:
"""Return one best-effort topological order for a prompt node subset."""
indegree_by_node_id: dict[str, int] = {node_id: 0 for node_id in node_ids}
downstream_node_ids_by_node_id: dict[str, set[str]] = {
node_id: set() for node_id in node_ids
}
for node_id in sorted(node_ids):
prompt_node = prompt.get(node_id)
if prompt_node is None:
continue
for input_value in (prompt_node.get("inputs") or {}).values():
if not _is_link(input_value):
continue
upstream_node_id = str(input_value[0])
if upstream_node_id not in node_ids:
continue
if node_id in downstream_node_ids_by_node_id[upstream_node_id]:
continue
downstream_node_ids_by_node_id[upstream_node_id].add(node_id)
indegree_by_node_id[node_id] += 1
ready_node_ids = deque(
sorted(
node_id
for node_id, indegree in indegree_by_node_id.items()
if indegree == 0
)
)
ordered_node_ids: list[str] = []
while ready_node_ids:
node_id = ready_node_ids.popleft()
ordered_node_ids.append(node_id)
for downstream_node_id in sorted(downstream_node_ids_by_node_id[node_id]):
indegree_by_node_id[downstream_node_id] -= 1
if indegree_by_node_id[downstream_node_id] == 0:
ready_node_ids.append(downstream_node_id)
if len(ordered_node_ids) == len(node_ids):
return ordered_node_ids
logger.warning(
"Component phase planning encountered a node-level cycle; falling back to stable node order for %s.",
sorted(node_ids),
)
return sorted(node_ids)
def _remote_dependency_closure(
prompt: Mapping[str, Any],
source_node_id: str,
) -> set[str]:
"""Return the complete upstream dependency closure for one remote producer."""
closure: set[str] = set()
pending = [source_node_id]
while pending:
node_id = pending.pop()
if node_id in closure:
continue
prompt_node = prompt.get(node_id)
if not isinstance(prompt_node, Mapping):
continue
closure.add(node_id)
pending.extend(
str(input_value[0])
for input_value in (prompt_node.get("inputs") or {}).values()
if _is_link(input_value)
)
return closure
def _can_replicate_remote_dependency_closure(
*,
prompt: Mapping[str, Any],
source_node_id: str,
dependency_node_ids: set[str],
remote_node_ids: set[str],
) -> bool:
"""Return whether a producer closure can be safely rebuilt on another worker."""
if not dependency_node_ids or not dependency_node_ids.issubset(remote_node_ids):
return False
source_node = prompt.get(source_node_id)
if not isinstance(source_node, Mapping):
return False
source_inputs = source_node.get("inputs")
if not isinstance(source_inputs, Mapping):
return False
if not any(_is_link(value) for value in source_inputs.values()):
return True
return all(
"sampler"
not in str((prompt.get(node_id) or {}).get("class_type") or "").casefold()
for node_id in dependency_node_ids
)
def _replica_node_id_mapping(
*,
prompt: Mapping[str, Any],
dependency_node_ids: set[str],
source_node_id: str,
target_group_id: str,
) -> dict[str, str]:
"""Return collision-free deterministic ids for one replicated dependency closure."""
replica_digest = hashlib.sha256(
f"{source_node_id}\0{target_group_id}".encode("utf-8")
).hexdigest()[:12]
reserved_node_ids = set(prompt)
mapping: dict[str, str] = {}
for node_id in sorted(dependency_node_ids):
base_id = f"{_REMOTE_REPLICA_NODE_PREFIX}{replica_digest}__{node_id}"
replica_node_id = base_id
suffix = 1
while replica_node_id in reserved_node_ids:
replica_node_id = f"{base_id}_{suffix}"
suffix += 1
mapping[node_id] = replica_node_id
reserved_node_ids.add(replica_node_id)
return mapping
def _install_remote_dependency_replica(
*,
prompt: dict[str, Any],
remote_node_ids: set[str],
dependency_node_ids: set[str],
source_node_id: str,
target_group_id: str,
) -> str:
"""Clone one safe producer closure and return its replicated source node id."""
replica_ids = _replica_node_id_mapping(
prompt=prompt,
dependency_node_ids=dependency_node_ids,
source_node_id=source_node_id,
target_group_id=target_group_id,
)
for node_id in sorted(dependency_node_ids):
replica_node = copy.deepcopy(prompt[node_id])
for input_name, input_value in list(
(replica_node.get("inputs") or {}).items()
):
if _is_link(input_value) and str(input_value[0]) in replica_ids:
replica_node["inputs"][input_name] = [
replica_ids[str(input_value[0])],
int(input_value[1]),
]
prompt[replica_ids[node_id]] = replica_node
remote_node_ids.update(replica_ids.values())
return replica_ids[source_node_id]
def _cross_group_replicable_boundaries(
*,
prompt: dict[str, Any],
remote_node_ids: set[str],
component_groups: Mapping[str, set[str]],
nodes_module: Any,
) -> list[tuple[str, int, str, str, str]]:
"""Return replicable non-transportable edges crossing tentative components."""
group_by_node_id = {
node_id: group_id
for group_id, node_ids in component_groups.items()
for node_id in node_ids
}
boundaries: list[tuple[str, int, str, str, str]] = []
for target_node_id in sorted(remote_node_ids):
target_node = prompt.get(target_node_id)
if not isinstance(target_node, Mapping):
continue
for input_name, input_value in (target_node.get("inputs") or {}).items():
if not _is_link(input_value):
continue
source_node_id = str(input_value[0])
if (
source_node_id not in remote_node_ids
or group_by_node_id.get(source_node_id)
== group_by_node_id.get(target_node_id)
):
continue
output_index = int(input_value[1])
io_type = _remote_output_io_type(
prompt=prompt,
node_id=source_node_id,
output_index=output_index,
nodes_module=nodes_module,
)
if io_type not in _REPLICABLE_REMOTE_OBJECT_TYPES:
continue
boundaries.append(
(
source_node_id,
output_index,
target_node_id,
str(input_name),
str(group_by_node_id[target_node_id]),
)
)
return boundaries
def _replicate_safe_nontransportable_provider_boundaries(
*,
prompt: dict[str, Any],
remote_node_ids: set[str],
nodes_module: Any,
) -> None:
"""Replicate safe object producers instead of forcing provider cycles together."""
for _iteration in range(len(prompt) + 1):
consumers = _build_consumer_map(prompt)
tentative_groups = _remote_component_partition_groups(
prompt,
remote_node_ids,
consumers,
nodes_module,
preserve_nontransportable_affinity=False,
)
boundaries = _cross_group_replicable_boundaries(
prompt=prompt,
remote_node_ids=remote_node_ids,
component_groups=tentative_groups,
nodes_module=nodes_module,
)
replica_source_by_group: dict[tuple[str, str], str] = {}
changed = False
for (
source_node_id,
output_index,
target_node_id,
input_name,
group_id,
) in boundaries:
dependency_node_ids = _remote_dependency_closure(prompt, source_node_id)
if not _can_replicate_remote_dependency_closure(
prompt=prompt,
source_node_id=source_node_id,
dependency_node_ids=dependency_node_ids,
remote_node_ids=remote_node_ids,
):
continue
replica_key = (source_node_id, group_id)
replica_source_node_id = replica_source_by_group.get(replica_key)
if replica_source_node_id is None:
replica_source_node_id = _install_remote_dependency_replica(
prompt=prompt,
remote_node_ids=remote_node_ids,
dependency_node_ids=dependency_node_ids,
source_node_id=source_node_id,
target_group_id=group_id,
)
replica_source_by_group[replica_key] = replica_source_node_id
logger.info(
"Replicated non-transportable producer %s and dependencies %s "
"for remote component group %s.",
source_node_id,
sorted(dependency_node_ids),
group_id,
)
prompt[target_node_id]["inputs"][input_name] = [
replica_source_node_id,
output_index,
]
changed = True
if not changed:
return
raise ModalPromptValidationError(
"Unable to stabilize provider-aware non-transportable dependency replicas."
)
def _workflow_visible_remote_node_ids(node_ids: Iterable[str]) -> list[str]:
"""Exclude internal dependency replicas from workflow-facing rewrite metadata."""
return [
str(node_id)
for node_id in node_ids
if not str(node_id).startswith(_REMOTE_REPLICA_NODE_PREFIX)
]
def _build_component_plans(
prompt: dict[str, Any],
remote_node_ids: set[str],
nodes_module: Any,
) -> list[RemoteComponentPlan]:
"""Build plans for every connected remote component."""
_replicate_safe_nontransportable_provider_boundaries(
prompt=prompt,
remote_node_ids=remote_node_ids,
nodes_module=nodes_module,
)
consumers = _build_consumer_map(prompt)
components = _build_remote_components(
prompt, remote_node_ids, consumers, nodes_module
)
return [
_build_component_plan(
component, prompt, consumers, remote_node_ids, nodes_module
)
for component in components
]
def _mark_remote_to_remote_session_boundaries(
prompt: dict[str, Any],
components: list[RemoteComponentPlan],
nodes_module: Any,
assignments_by_component_id: Mapping[str, ExecutionAssignment],
) -> set[str]:
"""Keep only same-environment remote edges in provider-local session storage."""
consumers = _build_consumer_map(prompt)
component_id_by_node_id = {
node_id: component.representative_node_id
for component in components
for node_id in component.node_ids
}
session_sources: set[LinkedOutputRef] = set()
for component in components:
for boundary_output in component.boundary_outputs:
output_consumers = consumers.get(boundary_output.source, [])
if not output_consumers:
continue
remote_consumers = [
consumer
for consumer in output_consumers
if consumer.node_id in component_id_by_node_id
]
if not remote_consumers:
continue
producer_assignment = assignments_by_component_id[
component.representative_node_id
]
consumer_component_ids = {
component_id_by_node_id[consumer.node_id]
for consumer in remote_consumers
}
cross_environment_consumers = sorted(
consumer_component_id
for consumer_component_id in consumer_component_ids
if assignments_by_component_id[consumer_component_id].environment_id
!= producer_assignment.environment_id
)
if cross_environment_consumers:
logger.info(
"Materializing remote boundary output through ComfyUI across "
"execution environments source=%s:%d producer=%s environment=%s "
"consumer_components=%s.",
boundary_output.source.node_id,
boundary_output.source.output_index,
component.representative_node_id,
producer_assignment.environment_id,
cross_environment_consumers,
)
continue
(
non_returning_local_consumers,
returning_local_consumers,
) = _non_returning_local_output_consumers(
prompt=prompt,
source=boundary_output.source,
remote_node_ids=set(component_id_by_node_id),
consumers=consumers,
nodes_module=nodes_module,
)
if returning_local_consumers:
logger.info(
"Materializing remote boundary output normally because local consumers %s later return to remote execution source=%s:%d.",
[consumer.node_id for consumer in returning_local_consumers],
boundary_output.source.node_id,
boundary_output.source.output_index,
)
continue
if (
non_returning_local_consumers
and producer_assignment.provider is not ExecutionProvider.MODAL
):
logger.info(
"Materializing remote boundary output through ComfyUI because "
"provider-local bridge storage is not accessible to local consumers "
"source=%s:%d provider=%s producer_component=%s local_consumers=%s.",
boundary_output.source.node_id,
boundary_output.source.output_index,
producer_assignment.provider.value,
component.representative_node_id,
[
consumer.node_id
for consumer in non_returning_local_consumers
],
)
continue
if boundary_output.is_list and _is_transportable_output_type(
boundary_output.io_type
):
logger.info(
"Materializing transportable list boundary output through ComfyUI "
"to preserve scheduler item semantics source=%s:%d io_type=%s "
"producer_component=%s consumer_components=%s.",
boundary_output.source.node_id,
boundary_output.source.output_index,
boundary_output.io_type,
component.representative_node_id,
sorted(consumer_component_ids),
)
continue
boundary_output.session_output = True
boundary_output.session_consumer_node_ids = sorted(
{consumer.node_id for consumer in remote_consumers}
)
if non_returning_local_consumers:
materializer_node_id = (
f"__ModalLocalBridgeMaterializer__"
f"{boundary_output.source.node_id}_"
f"{boundary_output.source.output_index}"
)
reserved_node_ids = {
output.local_materializer_node_id
for planned_component in components
for output in planned_component.boundary_outputs
if output.local_materializer_node_id is not None
}
while (
materializer_node_id in prompt
or materializer_node_id in reserved_node_ids
):
materializer_node_id = f"{materializer_node_id}_proxy"
boundary_output.local_materializer_node_id = materializer_node_id
boundary_output.local_materializer_consumer_node_ids = sorted(
{consumer.node_id for consumer in non_returning_local_consumers}
)
boundary_output.preview_target_node_ids = []
session_sources.add(boundary_output.source)
logger.info(
"Keeping remote boundary output in provider-local storage source=%s:%d io_type=%s provider=%s producer_component=%s consumer_components=%s local_materializer=%s local_consumers=%s.",
boundary_output.source.node_id,
boundary_output.source.output_index,
boundary_output.io_type,
producer_assignment.provider.value,
component.representative_node_id,
sorted(
{
component_id_by_node_id[consumer.node_id]
for consumer in output_consumers
if consumer.node_id in component_id_by_node_id
}
),
boundary_output.local_materializer_node_id,
boundary_output.local_materializer_consumer_node_ids,
)
return _remote_session_component_ids(
components=components,
session_sources=session_sources,
)
def _remote_session_component_ids(
*,
components: list[RemoteComponentPlan],
session_sources: set[LinkedOutputRef],
) -> set[str]:
"""Return components that produce or consume provider-local boundary refs."""
return {
component.representative_node_id
for component in components
if any(output.session_output for output in component.boundary_outputs)
or any(
boundary_input.source in session_sources
for boundary_input in component.boundary_inputs
)
}
def _boundary_output_payload(
boundary_output: BoundaryOutputSpec,
*,
mapped_output: bool | None = None,
) -> dict[str, Any]:
"""Serialize one component boundary output for a remote payload."""
payload = {
"proxy_output_name": boundary_output.proxy_output_name,
"node_id": boundary_output.source.node_id,
"output_index": boundary_output.source.output_index,
"io_type": boundary_output.io_type,
"is_list": boundary_output.is_list,
"preview_target_node_ids": list(boundary_output.preview_target_node_ids),
}
if boundary_output.session_output:
payload["session_output"] = True
if mapped_output is not None:
payload["mapped_output"] = mapped_output
if mapped_output:
payload["scheduler_is_list"] = True
return payload
def _proxy_boundary_output_is_list(boundary_output: Mapping[str, Any]) -> bool:
"""Return whether ComfyUI should expose one proxy boundary as a list output."""
return bool(
boundary_output.get("is_list", False)
or boundary_output.get("scheduler_is_list", False)
)
def _implicitly_mapped_boundary_output_sources(
*,
component: RemoteComponentPlan,
original_prompt: dict[str, Any],
rewritten_prompt: dict[str, Any],
nodes_module: Any,
) -> set[LinkedOutputRef]:
"""Return component outputs transitively driven by scheduler-list proxy inputs."""
mapped_target_node_ids: set[str] = set()
list_boundary_input_names: set[str] = set()
for boundary_input in component.boundary_inputs:
input_is_scheduler_list = False
for target in boundary_input.targets:
target_prompt_node = rewritten_prompt.get(target.node_id)
if target_prompt_node is None:
continue
current_input_value = (target_prompt_node.get("inputs") or {}).get(
target.input_name
)
if not _is_link(current_input_value):
continue
if not _remote_output_is_list(
prompt=rewritten_prompt,
node_id=str(current_input_value[0]),
output_index=int(current_input_value[1]),
nodes_module=nodes_module,
):
continue
input_is_scheduler_list = True
mapped_target_node_ids.add(target.node_id)
if input_is_scheduler_list:
list_boundary_input_names.add(boundary_input.proxy_input_name)
if not mapped_target_node_ids:
return set()
consumers = _build_consumer_map(original_prompt)
mapped_node_ids = _component_downstream_closure(
seed_node_ids=mapped_target_node_ids,
component_node_id_set=set(component.node_ids),
consumers=consumers,
)
mapped_output_sources = {
boundary_output.source
for boundary_output in component.boundary_outputs
if boundary_output.source.node_id in mapped_node_ids
}
logger.info(
"Propagating scheduler-list metadata through remote component %s from "
"boundary inputs=%s to outputs=%s.",
component.representative_node_id,
sorted(list_boundary_input_names),
sorted(
f"{source.node_id}:{source.output_index}"
for source in mapped_output_sources
),
)
return mapped_output_sources
def _mark_payload_scheduler_list_outputs(
payload: dict[str, Any],
scheduler_list_output_sources: set[LinkedOutputRef],
) -> None:
"""Mark matching outputs across a component payload and any split phase payloads."""
if not scheduler_list_output_sources:
return
boundary_outputs = payload.get("boundary_outputs")
if isinstance(boundary_outputs, list):
for boundary_output in boundary_outputs:
if not isinstance(boundary_output, dict):
continue
source = LinkedOutputRef(
node_id=str(boundary_output.get("node_id")),
output_index=int(boundary_output.get("output_index", -1)),
)
if source in scheduler_list_output_sources:
boundary_output["scheduler_is_list"] = True
for phase_name in ("static_phase", "mapped_phase"):
phase_payload = payload.get(phase_name)
if isinstance(phase_payload, dict):
_mark_payload_scheduler_list_outputs(
phase_payload,
scheduler_list_output_sources,
)
split_proxy_payloads = payload.get("split_proxy_payloads")
if isinstance(split_proxy_payloads, dict):
for split_payload in split_proxy_payloads.values():
if isinstance(split_payload, dict):
_mark_payload_scheduler_list_outputs(
split_payload,
scheduler_list_output_sources,
)
elif isinstance(split_proxy_payloads, list):
for split_payload in split_proxy_payloads:
if isinstance(split_payload, dict):
_mark_payload_scheduler_list_outputs(
split_payload,
scheduler_list_output_sources,
)
def _mapped_boundary_origin_io_type(
prompt: dict[str, Any],
boundary_input: BoundaryInputSpec,
nodes_module: Any,
) -> str | None:
"""Return the effective io_type for one mapped boundary, unwrapping local ModalMapInput markers."""
source_prompt_node = prompt.get(boundary_input.source.node_id)
if source_prompt_node is None:
return boundary_input.io_type
if str(source_prompt_node.get("class_type")) != MODAL_MAP_INPUT_NODE_ID:
return _remote_output_io_type(
prompt=prompt,
node_id=boundary_input.source.node_id,
output_index=boundary_input.source.output_index,
nodes_module=nodes_module,
)
mapped_value = (source_prompt_node.get("inputs") or {}).get("value")
if not _is_link(mapped_value):
return boundary_input.io_type
return _remote_output_io_type(
prompt=prompt,
node_id=str(mapped_value[0]),
output_index=int(mapped_value[1]),
nodes_module=nodes_module,
)
def _describe_output_boundary_error(
component: RemoteComponentPlan,
source: LinkedOutputRef,
source_class_type: str,
io_type: str,
local_consumer: InputTarget,
local_consumer_class_type: str,
) -> str:
"""Format a human-readable component-output transport validation error."""
return (
"Remote component rooted at node "
f"{component.representative_node_id} exports node {source.node_id} "
f"({source_class_type}) output index {source.output_index} of type '{io_type}' "
f"to node {local_consumer.node_id} ({local_consumer_class_type}) outside that "
f"component at input '{local_consumer.input_name}', which cannot cross the "
"current component boundary. "
"Current ComfyUI-Modal transport only supports JSON-compatible values, bytes, "
"and media or tensor-like outputs such as VIDEO, AUDIO, IMAGE, MASK, LATENT, "
"SIGMAS, INT, FLOAT, BOOLEAN, and STRING."
)
def _describe_input_boundary_error(
component: RemoteComponentPlan,
target: InputTarget,
target_class_type: str,
source: LinkedOutputRef,
source_class_type: str,
io_type: str,
) -> str:
"""Format a human-readable component-input transport validation error."""
return (
"Remote node "
f"{target.node_id} ({target_class_type}) input '{target.input_name}' "
f"depends on node {source.node_id} ({source_class_type}) outside its component, "
f"using output index {source.output_index} of type '{io_type}', which cannot "
"cross the current component boundary. Current ComfyUI-Modal transport only "
"supports JSON-compatible values, bytes, and media or tensor-like outputs such "
"as VIDEO, AUDIO, IMAGE, MASK, LATENT, SIGMAS, INT, FLOAT, BOOLEAN, and STRING."
)
def validate_remote_component_transport_compatibility(
prompt: dict[str, Any],
components: list[RemoteComponentPlan],
nodes_module: Any,
) -> None:
"""Reject remote components whose true graph boundaries require unsupported transport."""
validation_errors: list[str] = []
consumers = _build_consumer_map(prompt)
logger.info(
"Validating %d remote components for transport compatibility.", len(components)
)
for component in components:
for boundary_input in component.boundary_inputs:
source_prompt_node = prompt.get(boundary_input.source.node_id)
if source_prompt_node is None:
continue
source_class_type = str(source_prompt_node["class_type"])
source_class = nodes_module.NODE_CLASS_MAPPINGS.get(source_class_type)
if source_class is None:
continue
source_output_types, _, _ = _normalize_output_metadata(source_class)
if boundary_input.source.output_index >= len(source_output_types):
continue
io_type = str(source_output_types[boundary_input.source.output_index])
if _is_transportable_output_type(io_type):
continue
for target in boundary_input.targets:
target_class_type = str(prompt[target.node_id]["class_type"])
validation_errors.append(
_describe_input_boundary_error(
component=component,
target=target,
target_class_type=target_class_type,
source=boundary_input.source,
source_class_type=source_class_type,
io_type=io_type,
)
)
for boundary_output in component.boundary_outputs:
if _is_transportable_output_type(boundary_output.io_type):
continue
source_class_type = str(
prompt[boundary_output.source.node_id]["class_type"]
)
for local_consumer in consumers.get(boundary_output.source, []):
if local_consumer.node_id in component.node_ids:
continue
local_consumer_class_type = str(
prompt[local_consumer.node_id]["class_type"]
)
validation_errors.append(
_describe_output_boundary_error(
component=component,
source=boundary_output.source,
source_class_type=source_class_type,
io_type=boundary_output.io_type,
local_consumer=local_consumer,
local_consumer_class_type=local_consumer_class_type,
)
)
if validation_errors:
raise ModalPromptValidationError("\n".join(validation_errors))
logger.info("Remote component transport validation passed.")
def _sync_component_prompt_inputs(
component: RemoteComponentPlan,
rewritten_prompt: dict[str, Any],
sync_engine: ModalAssetSyncEngine,
request_cache: AssetSyncRequestCache,
status_callback: Any | None = None,
) -> tuple[dict[str, Any], list[SyncedAsset]]:
"""Build a synced prompt payload for one remote component."""
component_prompt: dict[str, Any] = {}
synced_assets: list[SyncedAsset] = []
logger.info(
"Syncing prompt inputs for remote component %s with %d nodes.",
component.representative_node_id,
len(component.node_ids),
)
for node_id in component.node_ids:
prompt_node = rewritten_prompt[node_id]
synced_inputs, node_assets = sync_engine.sync_prompt_inputs(
copy.deepcopy(prompt_node.get("inputs", {})),
status_callback=status_callback,
request_cache=request_cache,
)
synced_assets.extend(node_assets)
component_prompt[node_id] = {
"class_type": str(prompt_node["class_type"]),
"inputs": synced_inputs,
"_meta": copy.deepcopy(prompt_node.get("_meta", {})),
}
logger.info(
"Prepared remote node %s (%s) with %d synced assets.",
node_id,
component_prompt[node_id]["class_type"],
len(node_assets),
)
logger.info(
"Finished syncing remote component %s with %d total synced assets.",
component.representative_node_id,
len(synced_assets),
)
return component_prompt, synced_assets
def _deduplicate_synced_assets(synced_assets: list[SyncedAsset]) -> list[SyncedAsset]:
"""Return one request summary record per content-addressed remote path."""
unique_assets_by_remote_path: dict[str, SyncedAsset] = {}
for synced_asset in synced_assets:
unique_assets_by_remote_path.setdefault(synced_asset.remote_path, synced_asset)
return list(unique_assets_by_remote_path.values())
def _build_component_payload(
component: RemoteComponentPlan,
component_prompt: dict[str, Any],
signature_prompt: dict[str, Any],
extra_data: dict[str, Any] | None,
settings: ModalSyncSettings,
requires_volume_reload: bool,
volume_reload_marker: str | None,
custom_nodes_bundle: SyncedAsset | None,
uploaded_volume_paths: list[str],
terminate_container_on_error: bool,
nodes_module: Any,
remote_session: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build the serialized execution payload for one remote component."""
prompt_id = (extra_data or {}).get("prompt_id")
custom_nodes_bundle_path = (
custom_nodes_bundle.remote_path if custom_nodes_bundle is not None else None
)
def remote_worker_affinity_group(component_node_ids: list[str]) -> str:
"""Return the worker-pool group required by one remote component phase."""
class_types = {
str((component_prompt.get(node_id) or {}).get("class_type") or "")
for node_id in component_node_ids
}
return "llm" if "ModalLLM" in class_types else "comfy"
def build_subgraph_payload(
*,
component_id: str,
component_node_ids: list[str],
boundary_inputs: list[BoundaryInputSpec],
boundary_outputs: list[dict[str, Any]],
execute_node_ids: list[str],
remote_session: dict[str, Any] | None = None,
clear_remote_session: bool = False,
mapped_progress_display_node_id: str | None = None,
) -> dict[str, Any]:
"""Build one ordinary subgraph payload for a proxy node."""
payload = {
"payload_kind": "subgraph",
"component_id": component_id,
"prompt_id": prompt_id,
"modal_gpu": settings.modal_gpu,
"remote_worker_affinity_group": remote_worker_affinity_group(
component_node_ids
),
"component_node_ids": list(component_node_ids),
"subgraph_prompt": _subset_component_prompt(
component_prompt, component_node_ids
),
"boundary_inputs": _serialize_boundary_input_specs(
boundary_inputs,
signature_prompt=signature_prompt,
),
"boundary_outputs": copy.deepcopy(boundary_outputs),
"execute_node_ids": list(execute_node_ids),
"extra_data": copy.deepcopy(extra_data or {}),
"requires_volume_reload": requires_volume_reload,
"volume_reload_marker": volume_reload_marker,
"uploaded_volume_paths": list(uploaded_volume_paths),
"terminate_container_on_error": terminate_container_on_error,
"custom_nodes_bundle": custom_nodes_bundle_path,
}
if remote_session is not None:
payload["remote_session"] = copy.deepcopy(remote_session)
if clear_remote_session:
payload["clear_remote_session"] = True
if mapped_progress_display_node_id is not None:
payload["mapped_progress_display_node_id"] = mapped_progress_display_node_id
return payload
def build_phase_payloads_for_transportable_splits() -> list[dict[str, Any]] | None:
"""Return ordered phases for local feedback or parallel local fanout."""
if len(component.execute_node_ids) <= 1:
return None
has_local_reentry_dependency = _component_has_local_reentry_dependency(
prompt=signature_prompt,
component=component,
)
has_parallel_local_remote_fanout = _component_has_parallel_local_remote_fanout(
component
)
if not has_local_reentry_dependency and not has_parallel_local_remote_fanout:
logger.info(
"Keeping remote component %s as one proxy because execute targets %s "
"have neither a local re-entry dependency nor parallel local/remote fanout.",
component.representative_node_id,
component.execute_node_ids,
)
return None
if component.local_tap_node_ids:
logger.info(
"Allowing remote component %s with local tap nodes %s to split because it has a local re-entry dependency.",
component.representative_node_id,
component.local_tap_node_ids,
)
if has_parallel_local_remote_fanout:
logger.info(
"Forcing ordered phases for remote component %s because a remote output "
"feeds both a non-returning local branch and later remote execution.",
component.representative_node_id,
)
component_node_id_set = set(component.node_ids)
topological_node_ids = _subgraph_topological_node_order(
component_prompt, component_node_id_set
)
remaining_node_ids = set(component.node_ids)
remaining_execute_node_ids = [
node_id
for node_id in topological_node_ids
if node_id in set(component.execute_node_ids)
]
remaining_execute_node_ids = _order_execute_node_ids_for_transportable_splits(
prompt=signature_prompt,
component_prompt=component_prompt,
component_node_ids=component_node_id_set,
execute_node_ids=remaining_execute_node_ids,
)
if len(remaining_execute_node_ids) <= 1:
return None
phase_payloads: list[dict[str, Any]] = []
produced_outputs_by_source: dict[LinkedOutputRef, ProducedPhaseOutputSpec] = {}
local_boundary_outputs_by_source = {
boundary_output.source: boundary_output
for boundary_output in component.boundary_outputs
}
bridge_output_counter = 0
while remaining_execute_node_ids:
target_node_id = remaining_execute_node_ids[0]
phase_node_ids = sorted(
_component_upstream_closure(
prompt=component_prompt,
seed_node_ids={target_node_id},
candidate_node_ids=remaining_node_ids,
)
)
phase_node_id_set = set(phase_node_ids)
if not phase_node_ids:
raise ModalPromptValidationError(
f"Unable to derive split phase nodes for remote component {component.representative_node_id}."
)
phase_boundary_inputs = _filter_boundary_inputs_for_node_ids(
component.boundary_inputs,
phase_node_id_set,
)
phase_boundary_inputs_by_name = {
boundary_input.proxy_input_name: boundary_input
for boundary_input in phase_boundary_inputs
}
for phase_node_id in phase_node_ids:
prompt_node = component_prompt.get(phase_node_id)
if prompt_node is None:
continue
for input_name, input_value in (
prompt_node.get("inputs") or {}
).items():
if not _is_link(input_value):
continue
source = LinkedOutputRef(
node_id=str(input_value[0]),
output_index=int(input_value[1]),
)
if source.node_id in phase_node_id_set:
continue
produced_output = produced_outputs_by_source.get(source)
if produced_output is None:
continue
boundary_input = phase_boundary_inputs_by_name.get(
produced_output.proxy_output_name
)
if boundary_input is None:
boundary_input = BoundaryInputSpec(
proxy_input_name=produced_output.proxy_output_name,
source=source,
io_type=produced_output.io_type,
)
phase_boundary_inputs.append(boundary_input)
phase_boundary_inputs_by_name[
boundary_input.proxy_input_name
] = boundary_input
boundary_input.targets.append(
InputTarget(node_id=phase_node_id, input_name=str(input_name))
)
phase_boundary_outputs: list[dict[str, Any]] = []
phase_output_names_by_source: dict[LinkedOutputRef, str] = {}
for boundary_output in _filter_boundary_outputs_for_node_ids(
component.boundary_outputs,
phase_node_id_set,
):
phase_boundary_outputs.append(_boundary_output_payload(boundary_output))
phase_output_names_by_source[
boundary_output.source
] = boundary_output.proxy_output_name
pending_node_ids = remaining_node_ids - phase_node_id_set
for pending_node_id in sorted(pending_node_ids):
prompt_node = component_prompt.get(pending_node_id)
if prompt_node is None:
continue
for input_value in (prompt_node.get("inputs") or {}).values():
if not _is_link(input_value):
continue
source = LinkedOutputRef(
node_id=str(input_value[0]),
output_index=int(input_value[1]),
)
if (
source.node_id not in phase_node_id_set
or source in produced_outputs_by_source
):
continue
local_boundary_output = local_boundary_outputs_by_source.get(source)
io_type = (
local_boundary_output.io_type
if local_boundary_output is not None
else str(
_remote_output_io_type(
prompt=component_prompt,
node_id=source.node_id,
output_index=source.output_index,
nodes_module=nodes_module,
)
or "*"
)
)
is_list = (
local_boundary_output.is_list
if local_boundary_output is not None
else _remote_output_is_list(
prompt=component_prompt,
node_id=source.node_id,
output_index=source.output_index,
nodes_module=nodes_module,
)
)
proxy_output_name = phase_output_names_by_source.get(source)
if proxy_output_name is None:
proxy_output_name = f"phase_bridge_{bridge_output_counter}"
bridge_output_counter += 1
phase_boundary_outputs.append(
{
"proxy_output_name": proxy_output_name,
"node_id": source.node_id,
"output_index": source.output_index,
"io_type": io_type,
"is_list": is_list,
"preview_target_node_ids": [],
"session_output": True,
}
)
phase_output_names_by_source[source] = proxy_output_name
produced_outputs_by_source[source] = ProducedPhaseOutputSpec(
proxy_output_name=proxy_output_name,
source=source,
io_type=io_type,
is_list=is_list,
session_output=True,
)
phase_execute_node_ids = [
node_id
for node_id in component.execute_node_ids
if node_id in phase_node_id_set
]
phase_payloads.append(
build_subgraph_payload(
component_id=str(target_node_id),
component_node_ids=phase_node_ids,
boundary_inputs=phase_boundary_inputs,
boundary_outputs=phase_boundary_outputs,
execute_node_ids=phase_execute_node_ids,
)
)
remaining_node_ids -= phase_node_id_set
remaining_execute_node_ids = [
node_id
for node_id in remaining_execute_node_ids
if node_id not in phase_node_id_set
]
has_session_bridges = any(
bool(boundary_output.get("session_output"))
for phase_payload in phase_payloads
for boundary_output in phase_payload.get("boundary_outputs", [])
)
if not phase_payloads or len(phase_payloads) <= 1:
return None
active_remote_session = remote_session
if active_remote_session is None and has_session_bridges:
active_remote_session = RemoteSessionHandle(
session_id=uuid.uuid4().hex,
prompt_id=(str(prompt_id) if prompt_id is not None else None),
owner_component_id=component.representative_node_id,
).to_payload()
if active_remote_session is not None:
for phase_index, phase_payload in enumerate(phase_payloads):
phase_payload["remote_session"] = copy.deepcopy(active_remote_session)
if phase_index == len(phase_payloads) - 1:
phase_payload["clear_remote_session"] = True
logger.info(
"Split ordinary remote component %s into ordered phases: %s",
component.representative_node_id,
[
{
"component_id": phase_payload["component_id"],
"component_node_ids": phase_payload["component_node_ids"],
"execute_node_ids": phase_payload["execute_node_ids"],
}
for phase_payload in phase_payloads
],
)
return phase_payloads
split_phase_payloads = build_phase_payloads_for_transportable_splits()
if split_phase_payloads is not None:
return _attach_snapshot_profile_key(
{"split_proxy_payloads": split_phase_payloads},
settings,
)
payload = {
"payload_kind": "mapped_subgraph"
if component.mapped_boundary_input_name
else "subgraph",
"component_id": component.representative_node_id,
"prompt_id": prompt_id,
"modal_gpu": settings.modal_gpu,
"remote_worker_affinity_group": remote_worker_affinity_group(
list(component.node_ids)
),
"component_node_ids": list(component.node_ids),
"subgraph_prompt": component_prompt,
"boundary_inputs": _serialize_boundary_input_specs(
component.boundary_inputs,
signature_prompt=signature_prompt,
),
"boundary_outputs": [
_boundary_output_payload(
boundary_output,
mapped_output=(
boundary_output.source.node_id in set(component.mapped_node_ids)
if component.mapped_boundary_input_name
else None
),
)
for boundary_output in component.boundary_outputs
],
"execute_node_ids": list(component.execute_node_ids),
"mapped_execute_node_ids": list(component.mapped_execute_node_ids),
"static_execute_node_ids": list(component.static_execute_node_ids),
"extra_data": copy.deepcopy(extra_data or {}),
"requires_volume_reload": requires_volume_reload,
"volume_reload_marker": volume_reload_marker,
"uploaded_volume_paths": list(uploaded_volume_paths),
"terminate_container_on_error": terminate_container_on_error,
"custom_nodes_bundle": custom_nodes_bundle_path,
"mapped_input": (
{
"proxy_input_name": component.mapped_boundary_input_name,
"io_type": str(component.mapped_boundary_input_io_type or "*"),
}
if component.mapped_boundary_input_name
else None
),
}
if remote_session is not None:
payload["remote_session"] = copy.deepcopy(remote_session)
payload["clear_remote_session"] = True
logger.info(
"Built remote payload for component %s: boundary_inputs=%d boundary_outputs=%d execute_nodes=%s custom_nodes_bundle=%s",
component.representative_node_id,
len(payload["boundary_inputs"]),
len(payload["boundary_outputs"]),
payload["execute_node_ids"],
payload["custom_nodes_bundle"],
)
logger.info(
"Remote payload for component %s requires_volume_reload=%s volume_reload_marker=%s",
component.representative_node_id,
requires_volume_reload,
volume_reload_marker,
)
if component.mapped_boundary_input_name:
static_node_id_set = set(component.static_node_ids)
mapped_node_id_set = set(component.mapped_node_ids)
static_boundary_inputs = _filter_boundary_inputs_for_node_ids(
component.boundary_inputs,
static_node_id_set,
)
mapped_boundary_inputs = _filter_boundary_inputs_for_node_ids(
component.boundary_inputs,
mapped_node_id_set,
)
static_boundary_outputs = _filter_boundary_outputs_for_node_ids(
component.boundary_outputs,
static_node_id_set,
)
mapped_boundary_outputs = _filter_boundary_outputs_for_node_ids(
component.boundary_outputs,
mapped_node_id_set,
)
static_bridge_outputs = [
{
"proxy_output_name": boundary_spec.proxy_name,
"node_id": boundary_spec.source.node_id,
"output_index": boundary_spec.source.output_index,
"io_type": boundary_spec.io_type,
"is_list": boundary_spec.is_list,
"preview_target_node_ids": [],
"session_output": True,
}
for boundary_spec in component.static_to_mapped_boundaries
]
payload["static_to_mapped_boundaries"] = [
{
"proxy_name": boundary_spec.proxy_name,
"node_id": boundary_spec.source.node_id,
"output_index": boundary_spec.source.output_index,
"io_type": boundary_spec.io_type,
"is_list": boundary_spec.is_list,
"targets": [
{"node_id": target.node_id, "input_name": target.input_name}
for target in boundary_spec.targets
],
}
for boundary_spec in component.static_to_mapped_boundaries
]
payload["static_phase"] = {
"component_node_ids": list(component.static_node_ids),
"subgraph_prompt": _subset_component_prompt(
component_prompt, component.static_node_ids
),
"boundary_inputs": _serialize_boundary_input_specs(
static_boundary_inputs,
signature_prompt=signature_prompt,
),
"boundary_outputs": [
_boundary_output_payload(boundary_output)
for boundary_output in static_boundary_outputs
]
+ static_bridge_outputs,
"execute_node_ids": list(component.static_execute_node_ids),
}
payload["mapped_phase"] = {
"component_node_ids": list(component.mapped_node_ids),
"subgraph_prompt": _subset_component_prompt(
component_prompt, component.mapped_node_ids
),
"boundary_inputs": _serialize_boundary_input_specs(
mapped_boundary_inputs
+ [
BoundaryInputSpec(
proxy_input_name=boundary_spec.proxy_name,
source=boundary_spec.source,
io_type=boundary_spec.io_type,
targets=list(boundary_spec.targets),
)
for boundary_spec in component.static_to_mapped_boundaries
],
signature_prompt=signature_prompt,
),
"boundary_outputs": [
_boundary_output_payload(boundary_output, mapped_output=True)
for boundary_output in mapped_boundary_outputs
],
"execute_node_ids": list(component.mapped_execute_node_ids),
}
if not component.static_node_ids:
return _attach_snapshot_profile_key(payload, settings)
if remote_session is None:
remote_session = RemoteSessionHandle(
session_id=uuid.uuid4().hex,
prompt_id=(str(prompt_id) if prompt_id is not None else None),
owner_component_id=component.representative_node_id,
).to_payload()
logger.info(
"Split hybrid component %s into static nodes=%s and mapped nodes=%s using remote_session session_id=%s with %d static bridge outputs.",
component.representative_node_id,
component.static_node_ids,
component.mapped_node_ids,
remote_session["session_id"],
len(static_bridge_outputs),
)
payload = {
"split_proxy_payloads": {
"static": build_subgraph_payload(
component_id=component.static_node_ids[0],
component_node_ids=list(component.static_node_ids),
boundary_inputs=static_boundary_inputs,
boundary_outputs=[
_boundary_output_payload(boundary_output)
for boundary_output in static_boundary_outputs
]
+ static_bridge_outputs,
execute_node_ids=list(component.static_execute_node_ids),
remote_session=remote_session,
),
"mapped": build_subgraph_payload(
component_id=f"{component.representative_node_id}__mapped",
component_node_ids=list(component.mapped_node_ids),
boundary_inputs=mapped_boundary_inputs
+ [
BoundaryInputSpec(
proxy_input_name=boundary_spec.proxy_name,
source=boundary_spec.source,
io_type=boundary_spec.io_type,
targets=list(boundary_spec.targets),
)
for boundary_spec in component.static_to_mapped_boundaries
],
boundary_outputs=[
_boundary_output_payload(boundary_output)
for boundary_output in mapped_boundary_outputs
],
execute_node_ids=list(component.mapped_execute_node_ids),
remote_session=remote_session,
clear_remote_session=True,
mapped_progress_display_node_id=component.static_node_ids[0],
),
}
}
payload["split_proxy_payloads"]["mapped"]["static_to_mapped_boundaries"] = [
{
"proxy_name": boundary_spec.proxy_name,
"node_id": boundary_spec.source.node_id,
"output_index": boundary_spec.source.output_index,
"io_type": boundary_spec.io_type,
"is_list": boundary_spec.is_list,
"targets": [
{"node_id": target.node_id, "input_name": target.input_name}
for target in boundary_spec.targets
],
}
for boundary_spec in component.static_to_mapped_boundaries
]
payload["split_proxy_payloads"]["mapped"]["static_phase"] = {
"component_node_ids": list(component.static_node_ids),
"subgraph_prompt": _subset_component_prompt(
component_prompt, component.static_node_ids
),
"boundary_inputs": _serialize_boundary_input_specs(
static_boundary_inputs,
signature_prompt=signature_prompt,
),
"boundary_outputs": copy.deepcopy(static_bridge_outputs),
"execute_node_ids": list(
dict.fromkeys(
boundary_spec.source.node_id
for boundary_spec in component.static_to_mapped_boundaries
)
),
}
return _attach_snapshot_profile_key(payload, settings)
def _component_uploaded_volume_paths(
*,
component_prompt: dict[str, Any],
synced_assets: list[SyncedAsset],
custom_nodes_bundle: SyncedAsset | None,
) -> list[str]:
"""Return newly uploaded mounted-volume paths that this component can actually reference."""
referenced_paths: set[str] = set()
for prompt_node in component_prompt.values():
if not isinstance(prompt_node, dict):
continue
inputs = prompt_node.get("inputs", {})
if not isinstance(inputs, dict):
continue
for input_value in inputs.values():
for candidate_path in _iter_payload_input_strings(input_value):
if isinstance(candidate_path, str) and candidate_path.startswith("/"):
referenced_paths.add(candidate_path)
uploaded_paths = {
asset.remote_path
for asset in synced_assets
if asset.uploaded and asset.remote_path in referenced_paths
}
if custom_nodes_bundle is not None and custom_nodes_bundle.uploaded:
uploaded_paths.add(custom_nodes_bundle.remote_path)
return sorted(uploaded_paths)
def _rewrite_component_into_proxy(
component: RemoteComponentPlan,
rewritten_prompt: dict[str, Any],
payload: dict[str, Any],
nodes_module: Any,
) -> list[str]:
"""Replace a remote component with a single proxy node in the prompt."""
def contains_output_node(node_ids: list[str]) -> bool:
"""Return whether one node subset contains an output node."""
for node_id in node_ids:
prompt_node = rewritten_prompt.get(node_id)
if prompt_node is None:
continue
node_class = nodes_module.NODE_CLASS_MAPPINGS.get(
str(prompt_node["class_type"])
)
if node_class is not None and getattr(node_class, "OUTPUT_NODE", False):
return True
return False
def proxy_inputs_for_boundary_inputs(
boundary_inputs: list[dict[str, Any]]
) -> dict[str, Any]:
"""Resolve one proxy input mapping from the current prompt graph."""
proxy_inputs: dict[str, Any] = {}
for boundary_input in boundary_inputs:
current_input_value: Any = None
for target in boundary_input.get("targets", []):
target_prompt_node = rewritten_prompt.get(str(target["node_id"]))
if target_prompt_node is None:
continue
target_input_value = (target_prompt_node.get("inputs") or {}).get(
str(target["input_name"])
)
if _is_link(target_input_value):
current_input_value = list(target_input_value)
break
if current_input_value is None:
raise ModalPromptValidationError(
"Unable to resolve proxy boundary input wiring while rewriting split Modal proxies."
)
proxy_inputs[str(boundary_input["proxy_input_name"])] = current_input_value
return proxy_inputs
def register_proxy_node(
*,
prompt_node_id: str,
payload_mapping: dict[str, Any],
proxy_inputs: dict[str, Any],
meta: dict[str, Any],
is_output_node: bool,
) -> None:
"""Insert one dynamic proxy node into the rewritten prompt."""
boundary_outputs = list(payload_mapping.get("boundary_outputs", []))
proxy_node_id = ensure_modal_component_proxy_node_registered(
output_types=tuple(str(output["io_type"]) for output in boundary_outputs),
output_names=tuple(
str(output["proxy_output_name"]) for output in boundary_outputs
),
output_is_list=tuple(
_proxy_boundary_output_is_list(output) for output in boundary_outputs
),
nodes_module=nodes_module,
is_output_node=is_output_node,
include_completion_output=True,
)
proxy_inputs["original_node_data"] = register_cache_friendly_proxy_payload(
prompt_node_id,
payload_mapping,
)
rewritten_prompt[prompt_node_id] = {
"class_type": proxy_node_id,
"inputs": proxy_inputs,
"_meta": copy.deepcopy(meta),
}
split_proxy_payloads = payload.get("split_proxy_payloads")
component_node_id_set = set(component.node_ids)
if isinstance(split_proxy_payloads, list):
phase_payloads = [dict(phase_payload) for phase_payload in split_proxy_payloads]
phase_proxy_node_ids: list[str] = []
produced_output_indices_by_name: dict[str, list[Any]] = {}
replacement_output_indices: dict[LinkedOutputRef, list[Any]] = {}
materializer_output_by_consumer_and_source: dict[
tuple[str, LinkedOutputRef], list[Any]
] = {}
proxy_output_by_consumer_and_source: dict[
tuple[str, LinkedOutputRef], list[Any]
] = {}
materializer_proxy_output_by_node_id: dict[str, list[Any]] = {}
boundary_output_specs_by_source = {
boundary_output.source: boundary_output
for boundary_output in component.boundary_outputs
}
component_proxy_node_ids: set[str] = set()
phase_proxy_inputs_by_node_id: dict[str, dict[str, Any]] = {}
phase_proxy_meta_by_node_id: dict[str, dict[str, Any]] = {}
if any(
spec.local_materializer_node_id is not None
for spec in component.boundary_outputs
):
ensure_modal_local_bridge_materializer_registered(nodes_module)
for phase_payload in phase_payloads:
phase_proxy_node_id = str(phase_payload["component_id"])
while (
phase_proxy_node_id in rewritten_prompt
and phase_proxy_node_id not in component_node_id_set
):
phase_proxy_node_id = f"{phase_proxy_node_id}_proxy"
phase_payload["component_id"] = phase_proxy_node_id
phase_proxy_node_ids.append(phase_proxy_node_id)
component_proxy_node_ids.add(phase_proxy_node_id)
for phase_payload in phase_payloads:
phase_proxy_node_id = str(phase_payload["component_id"])
phase_proxy_inputs = proxy_inputs_for_boundary_inputs(
list(phase_payload["boundary_inputs"])
)
for boundary_input in phase_payload.get("boundary_inputs", []):
proxy_input_name = str(boundary_input["proxy_input_name"])
produced_output_index = produced_output_indices_by_name.get(
proxy_input_name
)
if produced_output_index is None:
continue
phase_proxy_inputs[proxy_input_name] = list(produced_output_index)
first_phase_node_id = str(phase_payload["component_node_ids"][0])
phase_proxy_meta = copy.deepcopy(
rewritten_prompt[first_phase_node_id].get("_meta", {})
)
phase_proxy_inputs_by_node_id[phase_proxy_node_id] = phase_proxy_inputs
phase_proxy_meta_by_node_id[phase_proxy_node_id] = phase_proxy_meta
for output_index, boundary_output in enumerate(
phase_payload.get("boundary_outputs", [])
):
proxy_output_name = str(boundary_output["proxy_output_name"])
proxy_output = [phase_proxy_node_id, output_index]
produced_output_indices_by_name[proxy_output_name] = proxy_output
source = LinkedOutputRef(
node_id=str(boundary_output["node_id"]),
output_index=int(boundary_output["output_index"]),
)
boundary_output_spec = boundary_output_specs_by_source.get(source)
if boundary_output_spec is not None:
for (
consumer_node_id
) in boundary_output_spec.session_consumer_node_ids:
proxy_output_by_consumer_and_source[
(consumer_node_id, source)
] = proxy_output
materializer_node_id = (
boundary_output_spec.local_materializer_node_id
)
if materializer_node_id is not None:
materializer_proxy_output_by_node_id[
materializer_node_id
] = proxy_output
for (
consumer_node_id
) in boundary_output_spec.local_materializer_consumer_node_ids:
materializer_output_by_consumer_and_source[
(consumer_node_id, source)
] = [materializer_node_id, 0]
if bool(boundary_output.get("session_output")):
continue
replacement_output_indices[source] = proxy_output
for node_id in component.node_ids:
rewritten_prompt.pop(node_id, None)
for phase_payload in phase_payloads:
phase_proxy_node_id = str(phase_payload["component_id"])
register_proxy_node(
prompt_node_id=phase_proxy_node_id,
payload_mapping=phase_payload,
proxy_inputs=phase_proxy_inputs_by_node_id[phase_proxy_node_id],
meta=phase_proxy_meta_by_node_id[phase_proxy_node_id],
is_output_node=contains_output_node(
list(phase_payload["component_node_ids"])
),
)
for (
materializer_node_id,
proxy_output,
) in materializer_proxy_output_by_node_id.items():
rewritten_prompt[materializer_node_id] = {
"class_type": MODAL_LOCAL_BRIDGE_MATERIALIZER_NODE_ID,
"inputs": {"bridge_ref": list(proxy_output)},
"_meta": {"title": "Modal Local Bridge Materializer"},
}
if component.mapped_boundary_source_node_id is not None:
mapped_node_id_set = set(component.mapped_node_ids)
for phase_payload in phase_payloads:
phase_node_ids = {
str(node_id)
for node_id in phase_payload.get("component_node_ids", [])
}
if not (phase_node_ids & mapped_node_id_set):
continue
register_modal_map_input_warmup_context(
component.mapped_boundary_source_node_id,
phase_payload,
str(component.mapped_boundary_input_io_type or "*"),
)
break
for node_id, prompt_node in list(rewritten_prompt.items()):
if node_id in component_proxy_node_ids:
continue
for input_name, input_value in list(
(prompt_node.get("inputs") or {}).items()
):
if not _is_link(input_value):
continue
source = LinkedOutputRef(
node_id=str(input_value[0]), output_index=int(input_value[1])
)
replacement_output = materializer_output_by_consumer_and_source.get(
(node_id, source)
)
if replacement_output is None:
replacement_output = proxy_output_by_consumer_and_source.get(
(node_id, source)
)
if replacement_output is None:
replacement_output = replacement_output_indices.get(source)
if replacement_output is not None:
prompt_node["inputs"][input_name] = list(replacement_output)
logger.info(
"Rewrote remote component %s into ordered proxies %s.",
component.representative_node_id,
phase_proxy_node_ids,
)
return phase_proxy_node_ids
if isinstance(split_proxy_payloads, dict):
static_payload = dict(split_proxy_payloads["static"])
mapped_payload = dict(split_proxy_payloads["mapped"])
static_proxy_node_id = str(static_payload["component_id"])
mapped_proxy_node_id = str(mapped_payload["component_id"])
while (
mapped_proxy_node_id in rewritten_prompt
and mapped_proxy_node_id not in component_node_id_set
):
mapped_proxy_node_id = f"{mapped_proxy_node_id}_proxy"
mapped_payload["component_id"] = mapped_proxy_node_id
static_proxy_inputs = proxy_inputs_for_boundary_inputs(
list(static_payload["boundary_inputs"])
)
static_boundary_outputs = list(static_payload["boundary_outputs"])
static_proxy_meta = copy.deepcopy(
rewritten_prompt[static_proxy_node_id].get("_meta", {})
)
mapped_proxy_meta = copy.deepcopy(
rewritten_prompt[component.mapped_node_ids[0]].get("_meta", {})
)
bridge_output_indices = {
str(boundary_output["proxy_output_name"]): output_index
for output_index, boundary_output in enumerate(static_boundary_outputs)
if bool(boundary_output.get("session_output"))
}
mapped_proxy_inputs = proxy_inputs_for_boundary_inputs(
list(mapped_payload["boundary_inputs"])
)
mapped_static_phase = mapped_payload.get("static_phase")
if isinstance(mapped_static_phase, dict):
for input_name, input_value in proxy_inputs_for_boundary_inputs(
list(mapped_static_phase.get("boundary_inputs", []))
).items():
mapped_proxy_inputs.setdefault(input_name, input_value)
for boundary_input in mapped_payload.get("boundary_inputs", []):
proxy_input_name = str(boundary_input["proxy_input_name"])
if proxy_input_name not in bridge_output_indices:
continue
mapped_proxy_inputs[proxy_input_name] = [
static_proxy_node_id,
bridge_output_indices[proxy_input_name],
]
replacement_output_indices = {
LinkedOutputRef(
node_id=str(boundary_output["node_id"]),
output_index=int(boundary_output["output_index"]),
): [static_proxy_node_id, output_index]
for output_index, boundary_output in enumerate(static_boundary_outputs)
if not bool(boundary_output.get("session_output"))
}
replacement_output_indices.update(
{
LinkedOutputRef(
node_id=str(boundary_output["node_id"]),
output_index=int(boundary_output["output_index"]),
): [mapped_proxy_node_id, output_index]
for output_index, boundary_output in enumerate(
mapped_payload.get("boundary_outputs", [])
)
}
)
for node_id in component.node_ids:
rewritten_prompt.pop(node_id, None)
register_proxy_node(
prompt_node_id=static_proxy_node_id,
payload_mapping=static_payload,
proxy_inputs=static_proxy_inputs,
meta=static_proxy_meta,
is_output_node=contains_output_node(component.static_node_ids),
)
register_proxy_node(
prompt_node_id=mapped_proxy_node_id,
payload_mapping=mapped_payload,
proxy_inputs=mapped_proxy_inputs,
meta=mapped_proxy_meta,
is_output_node=contains_output_node(component.mapped_node_ids),
)
if component.mapped_boundary_source_node_id is not None:
register_modal_map_input_warmup_context(
component.mapped_boundary_source_node_id,
mapped_payload,
str(component.mapped_boundary_input_io_type or "*"),
)
for node_id, prompt_node in list(rewritten_prompt.items()):
if node_id in {static_proxy_node_id, mapped_proxy_node_id}:
continue
for input_name, input_value in list(
(prompt_node.get("inputs") or {}).items()
):
if not _is_link(input_value):
continue
source = LinkedOutputRef(
node_id=str(input_value[0]), output_index=int(input_value[1])
)
if source in replacement_output_indices:
prompt_node["inputs"][input_name] = list(
replacement_output_indices[source]
)
logger.info(
"Rewrote hybrid remote component %s into static proxy %s and mapped proxy %s.",
component.representative_node_id,
static_proxy_node_id,
mapped_proxy_node_id,
)
return [static_proxy_node_id, mapped_proxy_node_id]
boundary_outputs = list(payload.get("boundary_outputs", []))
proxy_node_id = ensure_modal_component_proxy_node_registered(
output_types=tuple(str(output["io_type"]) for output in boundary_outputs),
output_names=tuple(
str(output["proxy_output_name"]) for output in boundary_outputs
),
output_is_list=tuple(
_proxy_boundary_output_is_list(output) for output in boundary_outputs
),
nodes_module=nodes_module,
is_output_node=component.contains_output_node,
include_completion_output=True,
)
representative_node_id = component.representative_node_id
proxy_inputs = proxy_inputs_for_boundary_inputs(
list(payload.get("boundary_inputs", []))
)
proxy_inputs["original_node_data"] = register_cache_friendly_proxy_payload(
representative_node_id,
payload,
)
representative_meta = copy.deepcopy(
rewritten_prompt[representative_node_id].get("_meta", {})
)
rewritten_prompt[representative_node_id] = {
"class_type": proxy_node_id,
"inputs": proxy_inputs,
"_meta": representative_meta,
}
if component.mapped_boundary_source_node_id is not None:
register_modal_map_input_warmup_context(
component.mapped_boundary_source_node_id,
payload,
str(component.mapped_boundary_input_io_type or "*"),
)
materializer_output_by_consumer_and_source: dict[
tuple[str, LinkedOutputRef], list[Any]
] = {}
proxy_output_by_consumer_and_source: dict[
tuple[str, LinkedOutputRef], list[Any]
] = {}
default_proxy_output_by_source: dict[LinkedOutputRef, list[Any]] = {}
if any(
spec.local_materializer_node_id is not None
for spec in component.boundary_outputs
):
ensure_modal_local_bridge_materializer_registered(nodes_module)
for output_index, spec in enumerate(component.boundary_outputs):
proxy_output = [representative_node_id, output_index]
default_proxy_output_by_source.setdefault(spec.source, proxy_output)
for consumer_node_id in spec.session_consumer_node_ids:
proxy_output_by_consumer_and_source[
(consumer_node_id, spec.source)
] = proxy_output
materializer_node_id = spec.local_materializer_node_id
if materializer_node_id is None:
continue
rewritten_prompt[materializer_node_id] = {
"class_type": MODAL_LOCAL_BRIDGE_MATERIALIZER_NODE_ID,
"inputs": {"bridge_ref": proxy_output},
"_meta": {"title": "Modal Local Bridge Materializer"},
}
for consumer_node_id in spec.local_materializer_consumer_node_ids:
materializer_output_by_consumer_and_source[
(consumer_node_id, spec.source)
] = [materializer_node_id, 0]
for node_id, prompt_node in list(rewritten_prompt.items()):
if node_id in component_node_id_set and node_id != representative_node_id:
del rewritten_prompt[node_id]
continue
if node_id == representative_node_id:
continue
for input_name, input_value in list((prompt_node.get("inputs") or {}).items()):
if not _is_link(input_value):
continue
source = LinkedOutputRef(
node_id=str(input_value[0]), output_index=int(input_value[1])
)
replacement_output = materializer_output_by_consumer_and_source.get(
(node_id, source)
)
if replacement_output is None:
replacement_output = proxy_output_by_consumer_and_source.get(
(node_id, source)
)
if replacement_output is None:
replacement_output = default_proxy_output_by_source.get(source)
if replacement_output is not None:
prompt_node["inputs"][input_name] = list(replacement_output)
logger.info(
"Rewrote remote component %s with %d nodes to Modal proxy %s.",
representative_node_id,
len(component.node_ids),
proxy_node_id,
)
return [representative_node_id]
def _modal_component_completion_output_index(
*,
proxy_node_id: str,
rewritten_prompt: dict[str, Any],
nodes_module: Any,
) -> int:
"""Return the synthetic completion-token output index for one Modal proxy."""
prompt_node = rewritten_prompt.get(proxy_node_id)
if prompt_node is None:
raise ModalPromptValidationError(
f"Modal proxy node {proxy_node_id!r} is missing before artifact finalization."
)
class_type = str(prompt_node.get("class_type"))
node_class = nodes_module.NODE_CLASS_MAPPINGS.get(class_type)
if node_class is None:
raise ModalPromptValidationError(
f"Modal proxy class {class_type!r} is not registered before artifact finalization."
)
_output_types, output_names, _output_is_list = _normalize_output_metadata(
node_class
)
try:
return output_names.index(MODAL_COMPONENT_COMPLETION_OUTPUT_NAME)
except ValueError as exc:
raise ModalPromptValidationError(
f"Modal proxy {proxy_node_id!r} does not expose its completion token."
) from exc
def _attach_modal_artifact_finalizer(
*,
rewritten_prompt: dict[str, Any],
remote_component_ids: list[str],
nodes_module: Any,
) -> str:
"""Attach one internal output sink to every rewritten Modal component."""
if not remote_component_ids:
raise ModalPromptValidationError(
"Modal artifact finalization requires at least one remote component."
)
if len(remote_component_ids) > MODAL_ARTIFACT_FINALIZER_MAX_COMPONENTS:
raise ModalPromptValidationError(
"Modal artifact finalization supports at most "
f"{MODAL_ARTIFACT_FINALIZER_MAX_COMPONENTS} remote components per prompt."
)
ensure_modal_artifact_finalizer_registered(nodes_module)
finalizer_node_id = f"__{MODAL_ARTIFACT_FINALIZER_NODE_ID}__"
while finalizer_node_id in rewritten_prompt:
finalizer_node_id = f"{finalizer_node_id}_proxy"
finalizer_inputs = {
f"components.component_{component_index}": [
proxy_node_id,
_modal_component_completion_output_index(
proxy_node_id=proxy_node_id,
rewritten_prompt=rewritten_prompt,
nodes_module=nodes_module,
),
]
for component_index, proxy_node_id in enumerate(remote_component_ids)
}
rewritten_prompt[finalizer_node_id] = {
"class_type": MODAL_ARTIFACT_FINALIZER_NODE_ID,
"inputs": finalizer_inputs,
"_meta": {"title": "Modal Artifact Finalizer"},
}
logger.info(
"Attached Modal artifact finalizer %s to remote components %s.",
finalizer_node_id,
remote_component_ids,
)
return finalizer_node_id
def _nearest_downstream_remote_component_ids(
*,
rewritten_prompt: dict[str, Any],
consumers: dict[LinkedOutputRef, list[InputTarget]],
seed_targets: list[InputTarget],
remote_component_id_set: set[str],
nodes_module: Any,
) -> list[str]:
"""Return the first remote proxies reached along every downstream branch."""
nearest_remote_component_ids: set[str] = set()
visited_node_ids: set[str] = set()
pending_node_ids = deque(str(target.node_id) for target in seed_targets)
while pending_node_ids:
node_id = pending_node_ids.popleft()
if node_id in visited_node_ids or node_id not in rewritten_prompt:
continue
visited_node_ids.add(node_id)
if node_id in remote_component_id_set:
nearest_remote_component_ids.add(node_id)
continue
for output_ref in _node_output_refs(
rewritten_prompt,
node_id,
nodes_module,
):
for downstream_target in consumers.get(output_ref, []):
pending_node_ids.append(str(downstream_target.node_id))
return sorted(nearest_remote_component_ids)
def _parallelize_non_returning_local_branches(
*,
rewritten_prompt: dict[str, Any],
remote_component_ids: list[str],
nodes_module: Any,
) -> list[str]:
"""Release local-only taps once the nearest remote continuations are in flight."""
remote_component_id_set = set(remote_component_ids)
if len(remote_component_id_set) < 2:
return []
consumers = _build_consumer_map(rewritten_prompt)
ensure_modal_parallel_local_passthrough_registered(nodes_module)
passthrough_node_ids: list[str] = []
dispatch_group_id = uuid.uuid4().hex
for component_id in remote_component_ids:
embedded_payload = (
rewritten_prompt.get(component_id, {})
.get("inputs", {})
.get("original_node_data")
)
if not isinstance(embedded_payload, Mapping):
continue
payload_prompt_id = embedded_payload.get("prompt_id")
if payload_prompt_id is not None and str(payload_prompt_id).strip():
dispatch_group_id = str(payload_prompt_id)
break
for source_component_id in remote_component_ids:
for source in _node_output_refs(
rewritten_prompt,
source_component_id,
nodes_module,
):
output_consumers = list(consumers.get(source, []))
if not output_consumers:
continue
downstream_remote_component_ids = _nearest_downstream_remote_component_ids(
rewritten_prompt=rewritten_prompt,
consumers=consumers,
seed_targets=output_consumers,
remote_component_id_set=remote_component_id_set,
nodes_module=nodes_module,
)
downstream_remote_component_ids = [
component_id
for component_id in downstream_remote_component_ids
if component_id != source_component_id
]
if not downstream_remote_component_ids:
continue
local_only_consumers: list[InputTarget] = []
for target in output_consumers:
if target.node_id in remote_component_id_set:
continue
branch_node_ids = _downstream_node_ids_from_targets(
prompt=rewritten_prompt,
consumers=consumers,
seed_targets=[target],
nodes_module=nodes_module,
)
if branch_node_ids & remote_component_id_set:
continue
local_only_consumers.append(target)
if not local_only_consumers:
continue
passthrough_node_id = (
f"__{MODAL_PARALLEL_LOCAL_PASSTHROUGH_NODE_ID}__"
f"{len(passthrough_node_ids)}"
)
while passthrough_node_id in rewritten_prompt:
passthrough_node_id = f"{passthrough_node_id}_proxy"
passthrough_inputs: dict[str, Any] = {
"value": [source.node_id, source.output_index],
"dispatch_context": {
"dispatch_group_id": dispatch_group_id,
"component_ids": downstream_remote_component_ids,
},
}
rewritten_prompt[passthrough_node_id] = {
"class_type": MODAL_PARALLEL_LOCAL_PASSTHROUGH_NODE_ID,
"inputs": passthrough_inputs,
"_meta": {"title": "Modal Parallel Local Branch"},
}
for target in local_only_consumers:
rewritten_prompt[target.node_id]["inputs"][target.input_name] = [
passthrough_node_id,
0,
]
for component_id in downstream_remote_component_ids:
prompt_inputs = rewritten_prompt[component_id].get("inputs", {})
embedded_payload = prompt_inputs.get("original_node_data")
if not isinstance(embedded_payload, Mapping):
continue
prompt_inputs[
"original_node_data"
] = update_registered_proxy_payload_fields(
component_id,
embedded_payload,
{
"parallel_local_dispatch_group_id": dispatch_group_id,
"signal_parallel_local_dispatch": True,
},
)
passthrough_node_ids.append(passthrough_node_id)
logger.info(
"Parallelized local-only consumers %s of remote output %s:%d after downstream Modal dispatches %s via passthrough=%s.",
[target.node_id for target in local_only_consumers],
source.node_id,
source.output_index,
downstream_remote_component_ids,
passthrough_node_id,
)
return passthrough_node_ids
def _configure_local_gap_keepalive_payloads(
*,
rewritten_prompt: dict[str, Any],
remote_component_ids: list[str],
sandwiched_local_node_ids: set[str],
) -> None:
"""Mark only remote proxies that precede a local-to-remote execution gap."""
if not sandwiched_local_node_ids:
return
payloads_by_component_id: dict[str, Mapping[str, Any]] = {}
for component_id in remote_component_ids:
payload = (
rewritten_prompt.get(component_id, {})
.get("inputs", {})
.get("original_node_data")
)
if not isinstance(payload, dict):
continue
payloads_by_component_id[component_id] = payload
keepalive_component_ids: set[str] = set()
continuation_component_ids: set[str] = set()
for local_node_id in sandwiched_local_node_ids:
upstream_component_ids: set[str] = set()
downstream_component_ids: set[str] = set()
for component_id in remote_component_ids:
if _component_ancestors_of_local_source(
prompt=rewritten_prompt,
source_node_id=local_node_id,
component_node_id_set={component_id},
):
upstream_component_ids.add(component_id)
if _component_ancestors_of_local_source(
prompt=rewritten_prompt,
source_node_id=component_id,
component_node_id_set={local_node_id},
):
downstream_component_ids.add(component_id)
for upstream_component_id in upstream_component_ids:
upstream_payload = payloads_by_component_id.get(upstream_component_id)
if upstream_payload is None:
continue
upstream_environment_id = str(
upstream_payload.get("execution_environment_id") or ""
)
if upstream_payload.get("execution_provider") != ExecutionProvider.MODAL.value:
continue
for downstream_component_id in downstream_component_ids:
downstream_payload = payloads_by_component_id.get(
downstream_component_id
)
if (
downstream_payload is None
or downstream_payload.get("execution_provider")
!= ExecutionProvider.MODAL.value
or str(downstream_payload.get("execution_environment_id") or "")
!= upstream_environment_id
):
continue
keepalive_component_ids.add(upstream_component_id)
continuation_component_ids.add(downstream_component_id)
for component_id, embedded_payload in payloads_by_component_id.items():
if component_id not in keepalive_component_ids | continuation_component_ids:
continue
payload_fields: dict[str, Any] = {"remote_local_gap_pool": True}
if component_id in keepalive_component_ids:
payload_fields["keepalive_after_remote_component"] = True
if component_id in continuation_component_ids:
payload_fields["stop_local_gap_keepalive_before_remote_component"] = True
rewritten_prompt[component_id]["inputs"][
"original_node_data"
] = update_registered_proxy_payload_fields(
component_id,
embedded_payload,
payload_fields,
)
logger.info(
"Configured local-gap Modal pool for components=%s keepalive_producers=%s continuations=%s.",
sorted(keepalive_component_ids | continuation_component_ids),
sorted(keepalive_component_ids),
sorted(continuation_component_ids),
)
def _remote_proxy_payload(
rewritten_prompt: Mapping[str, Any], component_id: str
) -> Mapping[str, Any] | None:
"""Return the registered execution payload embedded in one remote proxy."""
prompt_node = rewritten_prompt.get(component_id)
if not isinstance(prompt_node, Mapping):
return None
inputs = prompt_node.get("inputs")
if not isinstance(inputs, Mapping):
return None
payload = inputs.get("original_node_data")
if not isinstance(payload, Mapping):
return None
return registered_proxy_execution_payload(component_id, payload)
def _remote_proxy_dependency_edges(
rewritten_prompt: Mapping[str, Any],
component_ids: set[str],
) -> dict[str, set[str]]:
"""Return nearest remote dependencies while traversing intervening local nodes."""
dependency_edges = {component_id: set() for component_id in component_ids}
for downstream_component_id in component_ids:
downstream_node = rewritten_prompt.get(downstream_component_id)
if not isinstance(downstream_node, Mapping):
continue
pending_node_ids = [
str(input_value[0])
for input_value in (downstream_node.get("inputs") or {}).values()
if _is_link(input_value)
]
visited_node_ids: set[str] = set()
while pending_node_ids:
upstream_node_id = pending_node_ids.pop()
if upstream_node_id in visited_node_ids:
continue
visited_node_ids.add(upstream_node_id)
if upstream_node_id in component_ids:
dependency_edges[upstream_node_id].add(downstream_component_id)
continue
upstream_node = rewritten_prompt.get(upstream_node_id)
if not isinstance(upstream_node, Mapping):
continue
pending_node_ids.extend(
str(input_value[0])
for input_value in (upstream_node.get("inputs") or {}).values()
if _is_link(input_value)
)
return dependency_edges
def _component_descendant_distances(
component_id: str,
dependency_edges: Mapping[str, set[str]],
) -> dict[str, int]:
"""Return shortest downstream distances from one remote component."""
distances: dict[str, int] = {}
pending = deque(
(descendant_id, 1)
for descendant_id in dependency_edges.get(component_id, set())
)
while pending:
descendant_id, distance = pending.popleft()
previous_distance = distances.get(descendant_id)
if previous_distance is not None and previous_distance <= distance:
continue
distances[descendant_id] = distance
pending.extend(
(nested_descendant_id, distance + 1)
for nested_descendant_id in dependency_edges.get(descendant_id, set())
)
return distances
def _speculative_prewarm_target_payload(
payload: Mapping[str, Any],
) -> dict[str, Any]:
"""Return the minimal side-effect-free payload needed to prepare one worker."""
return {
field_name: copy.deepcopy(payload[field_name])
for field_name in _SPECULATIVE_PREWARM_PAYLOAD_FIELDS
if field_name in payload
}
def _configure_speculative_affinity_prewarm_payloads(
*,
rewritten_prompt: dict[str, Any],
execution_stages: list[list[str]],
) -> None:
"""Attach one next-affinity worker preparation target to each eligible proxy."""
stage_index_by_component_id = {
component_id: stage_index
for stage_index, stage in enumerate(execution_stages)
for component_id in stage
}
payloads_by_component_id = {
component_id: payload
for component_id in stage_index_by_component_id
if (payload := _remote_proxy_payload(rewritten_prompt, component_id))
is not None
}
dependency_edges = _remote_proxy_dependency_edges(
rewritten_prompt,
set(payloads_by_component_id),
)
configured_targets: dict[str, str] = {}
for component_id, payload in payloads_by_component_id.items():
if payload.get("execution_provider") != ExecutionProvider.MODAL.value:
continue
execution_environment_id = str(
payload.get("execution_environment_id") or ""
)
affinity_group = (
str(payload.get("remote_worker_affinity_group") or "comfy").strip().lower()
)
descendant_distances = _component_descendant_distances(
component_id, dependency_edges
)
candidate_ids = sorted(
(
descendant_id
for descendant_id in descendant_distances
if descendant_id in payloads_by_component_id
and payloads_by_component_id[descendant_id].get(
"execution_provider"
)
== ExecutionProvider.MODAL.value
and str(
payloads_by_component_id[descendant_id].get(
"execution_environment_id"
)
or ""
)
== execution_environment_id
and str(
payloads_by_component_id[descendant_id].get(
"remote_worker_affinity_group"
)
or "comfy"
)
.strip()
.lower()
!= affinity_group
),
key=lambda descendant_id: (
descendant_distances[descendant_id],
stage_index_by_component_id[descendant_id],
descendant_id,
),
)
if not candidate_ids:
continue
target_component_id = candidate_ids[0]
target_payload = _speculative_prewarm_target_payload(
payloads_by_component_id[target_component_id]
)
rewritten_prompt[component_id]["inputs"][
"original_node_data"
] = update_registered_proxy_payload_fields(
component_id,
payload,
{_SPECULATIVE_PREWARM_TARGET_KEY: target_payload},
)
configured_targets[component_id] = target_component_id
if configured_targets:
logger.info(
"Configured one-step speculative Modal worker prewarm targets=%s.",
configured_targets,
)
def _environment_setup_status_callback(
environment_id: str,
status_callback: SetupStatusCallback | None,
environment_status_callback: EnvironmentSetupStatusCallback | None,
) -> SetupStatusCallback | None:
"""Return a setup callback that updates prompt-wide and environment UI."""
if status_callback is None and environment_status_callback is None:
return None
def emit_status(
message: str,
current: int | None,
total: int | None,
) -> None:
"""Forward one setup update with its concrete execution environment."""
if status_callback is not None:
status_callback(message, current, total)
if environment_status_callback is not None:
environment_status_callback(environment_id, message, current, total)
return emit_status
def _prepare_environment_assets(
*,
environment_id: str,
components: Sequence[RemoteComponentPlan],
sync_engine: ModalAssetSyncEngine,
engine_lock: threading.Lock,
rewritten_prompt: dict[str, Any],
sync_custom_nodes: bool,
completion_message: str,
status_callback: SetupStatusCallback | None,
environment_status_callback: EnvironmentSetupStatusCallback | None,
) -> _EnvironmentAssetPreparationResult:
"""Prepare one environment's custom nodes and prompt assets in order."""
started_at = time.perf_counter()
logger.info(
"Starting remote asset preparation environment=%s components=%d.",
environment_id,
len(components),
)
environment_callback = _environment_setup_status_callback(
environment_id,
status_callback,
environment_status_callback,
)
if environment_callback is not None:
environment_callback("Preparing remote assets", None, None)
with engine_lock:
sync_engine.preflight_r2_access(status_callback=environment_callback)
custom_nodes_bundle = (
sync_engine.sync_custom_nodes_directory(
status_callback=environment_callback,
)
if sync_custom_nodes
else None
)
component_prompts, assets_by_component_id = (
_sync_environment_prompt_assets(
components=components,
sync_engine=sync_engine,
rewritten_prompt=rewritten_prompt,
status_callback=environment_callback,
)
)
if environment_callback is not None:
environment_callback(completion_message, None, None)
logger.info(
"Finished remote asset preparation environment=%s components=%d assets=%d elapsed_seconds=%.3f.",
environment_id,
len(components),
sum(len(assets) for assets in assets_by_component_id.values()),
time.perf_counter() - started_at,
)
return _EnvironmentAssetPreparationResult(
environment_id=environment_id,
custom_nodes_bundle=custom_nodes_bundle,
component_prompts=component_prompts,
assets_by_component_id=assets_by_component_id,
)
def _sync_environment_prompt_assets(
*,
components: Sequence[RemoteComponentPlan],
sync_engine: ModalAssetSyncEngine,
rewritten_prompt: dict[str, Any],
status_callback: SetupStatusCallback | None,
) -> tuple[dict[str, dict[str, Any]], dict[str, list[SyncedAsset]]]:
"""Sync every component assigned to one environment with one request cache."""
request_cache = sync_engine.create_request_asset_cache(
rewritten_prompt[node_id].get("inputs", {})
for component in components
for node_id in component.node_ids
)
component_prompts: dict[str, dict[str, Any]] = {}
assets_by_component_id: dict[str, list[SyncedAsset]] = {}
for component in components:
component_prompt, synced_assets = _sync_component_prompt_inputs(
component=component,
rewritten_prompt=rewritten_prompt,
sync_engine=sync_engine,
request_cache=request_cache,
status_callback=status_callback,
)
component_id = component.representative_node_id
component_prompts[component_id] = component_prompt
assets_by_component_id[component_id] = list(synced_assets)
return component_prompts, assets_by_component_id
def _prepare_remote_environment_assets(
*,
components: Sequence[RemoteComponentPlan],
assignments_by_component_id: Mapping[str, ExecutionAssignment],
sync_engines_by_environment: Mapping[str, ModalAssetSyncEngine],
rewritten_prompt: dict[str, Any],
sync_custom_nodes: bool,
status_callback: SetupStatusCallback | None,
environment_status_callback: EnvironmentSetupStatusCallback | None,
) -> dict[str, _EnvironmentAssetPreparationResult]:
"""Prepare distinct remote environments concurrently in a bounded pool."""
components_by_environment: dict[str, list[RemoteComponentPlan]] = {
environment_id: [] for environment_id in sync_engines_by_environment
}
providers_by_environment: dict[str, ExecutionProvider] = {}
for component in components:
assignment = assignments_by_component_id[component.representative_node_id]
components_by_environment[assignment.environment_id].append(component)
providers_by_environment[assignment.environment_id] = assignment.provider
locks_by_engine: dict[int, threading.Lock] = {}
environment_ids = list(components_by_environment)
if not environment_ids:
return {}
max_workers = min(_REMOTE_ASSET_PREPARATION_MAX_WORKERS, len(environment_ids))
logger.info(
"Preparing remote assets across %d environments with %d workers.",
len(environment_ids),
max_workers,
)
with ThreadPoolExecutor(
max_workers=max_workers,
thread_name_prefix="comfy-remote-assets",
) as executor:
futures_by_environment = {
environment_id: executor.submit(
_prepare_environment_assets,
environment_id=environment_id,
components=components_by_environment[environment_id],
sync_engine=sync_engines_by_environment[environment_id],
engine_lock=locks_by_engine.setdefault(
id(sync_engines_by_environment[environment_id]),
threading.Lock(),
),
rewritten_prompt=rewritten_prompt,
sync_custom_nodes=sync_custom_nodes,
completion_message=_asset_preparation_completion_message(
providers_by_environment[environment_id]
),
status_callback=status_callback,
environment_status_callback=environment_status_callback,
)
for environment_id in environment_ids
}
return {
environment_id: futures_by_environment[environment_id].result()
for environment_id in environment_ids
}
def _asset_preparation_completion_message(provider: ExecutionProvider) -> str:
"""Return the truthful state after assets, but not runtime startup, finish."""
if provider is ExecutionProvider.SSH_DOCKER:
return "Remote assets prepared; SSH runtime starts on dispatch"
if provider is ExecutionProvider.MODAL:
return "Remote assets prepared; Modal runtime starts on dispatch"
return "Ready for remote execution"
def rewrite_prompt_for_modal(
prompt: dict[str, Any],
workflow: dict[str, Any] | None,
sync_engine: ModalAssetSyncEngine | None = None,
settings: ModalSyncSettings | None = None,
nodes_module: Any | None = None,
extra_data: dict[str, Any] | None = None,
status_callback: SetupStatusCallback | None = None,
environment_status_callback: EnvironmentSetupStatusCallback | None = None,
plan_callback: ExecutionPlanStatusCallback | None = None,
cancellation_check: Callable[[], bool] | None = None,
occupied_environment_ids: frozenset[str] = frozenset(),
) -> tuple[dict[str, Any], RewriteSummary]:
"""Rewrite connected remote components into Modal proxy nodes."""
if cancellation_check is not None and cancellation_check():
raise SyncCancelledError("Remote workflow preparation was cancelled.")
resolved_settings = settings or get_settings()
remote_node_ids = requested_remote_node_ids(
prompt=prompt,
workflow=workflow,
settings=resolved_settings,
)
summary = RewriteSummary(remote_node_ids=sorted(remote_node_ids))
logger.info(
"Found %d workflow nodes marked for remote execution.", len(remote_node_ids)
)
if not remote_node_ids:
return copy.deepcopy(prompt), summary
resolved_nodes_module = nodes_module or _get_nodes_module()
resolved_sync_engine = sync_engine or ModalAssetSyncEngine.from_environment(
resolved_settings
)
rewritten_prompt = copy.deepcopy(prompt)
expanded_remote_node_ids, _ = _expand_remote_node_ids_for_non_transportable_inputs(
prompt=rewritten_prompt,
remote_node_ids=remote_node_ids,
nodes_module=resolved_nodes_module,
)
expanded_remote_node_ids = _expand_remote_node_ids_for_terminal_video_sinks(
prompt=rewritten_prompt,
remote_node_ids=expanded_remote_node_ids,
nodes_module=resolved_nodes_module,
)
summary.remote_node_ids = sorted(expanded_remote_node_ids)
summary.sandwiched_local_node_ids = sorted(
_sandwiched_local_node_ids(rewritten_prompt, expanded_remote_node_ids)
)
if summary.sandwiched_local_node_ids:
logger.warning(
"Detected local nodes sandwiched between remote graph regions; "
"these nodes may force additional remote phases: %s",
summary.sandwiched_local_node_ids,
)
components = _build_component_plans(
rewritten_prompt,
expanded_remote_node_ids,
resolved_nodes_module,
)
validate_remote_component_transport_compatibility(
prompt=rewritten_prompt,
components=components,
nodes_module=resolved_nodes_module,
)
execution_plan = _plan_component_execution(
components=components,
prompt=rewritten_prompt,
workflow=workflow,
settings=resolved_settings,
status_callback=status_callback,
environment_status_callback=environment_status_callback,
plan_callback=plan_callback,
occupied_environment_ids=occupied_environment_ids,
)
assignments_by_component_id = execution_plan.assignments
if execution_plan.configuration_set is not None:
summary.remote_configurations = list(execution_plan.safe_configurations)
for assignment in assignments_by_component_id.values():
execution_location = _execution_location_for_assignment(
assignment,
execution_plan.ssh_hosts_by_id,
execution_plan.vast_leases_by_environment,
)
if execution_location:
summary.execution_locations_by_environment[
assignment.environment_id
] = execution_location
session_component_ids = _mark_remote_to_remote_session_boundaries(
rewritten_prompt,
components,
resolved_nodes_module,
assignments_by_component_id,
)
prompt_id = (extra_data or {}).get("prompt_id")
remote_sessions_by_component_id = {
component.representative_node_id: RemoteSessionHandle(
session_id=uuid.uuid4().hex,
prompt_id=(str(prompt_id) if prompt_id is not None else None),
owner_component_id=component.representative_node_id,
).to_payload()
for component in components
if component.representative_node_id in session_component_ids
}
if remote_sessions_by_component_id:
logger.info(
"Enabled environment-local remote references for components=%s.",
sorted(remote_sessions_by_component_id),
)
summary.execution_assignments_by_representative = dict(assignments_by_component_id)
vast_service = execution_plan.vast_service
vast_leases_by_environment = dict(
execution_plan.vast_leases_by_environment
)
if vast_service is None and any(
assignment.provider is ExecutionProvider.VAST
for assignment in assignments_by_component_id.values()
):
try:
vast_service = VastService.from_environment(
resolved_settings,
repo_root=Path(__file__).resolve().parent,
)
vast_leases_by_environment = {
assignment.environment_id: vast_service.lease_for_environment_id(
assignment.environment_id
)
for assignment in assignments_by_component_id.values()
if assignment.provider is ExecutionProvider.VAST
}
except (KeyError, OSError, RuntimeError, ValueError) as exc:
raise ModalPromptValidationError(
f"Unable to resolve the acquired Vast.ai lease: {exc}"
) from exc
workflow_r2_cache = (
_workflow_r2_cache(execution_plan.configuration_set)
if any(
assignment.provider is not ExecutionProvider.MODAL
for assignment in assignments_by_component_id.values()
)
else None
)
if workflow_r2_cache is not None and vast_service is not None:
vast_service.r2_cache = workflow_r2_cache
sync_engines_by_environment: dict[str, ModalAssetSyncEngine] = {}
ssh_hosts_by_id = dict(execution_plan.ssh_hosts_by_id)
if not ssh_hosts_by_id:
ssh_hosts_by_id = {
host.environment_id: host
for host in _configured_ssh_hosts(resolved_settings)
}
session_environment_ids = {
assignments_by_component_id[component_id].environment_id
for component_id in session_component_ids
}
worker_counts_by_environment: dict[str, int] = defaultdict(int)
for component in components:
component_id = component.representative_node_id
assignment = assignments_by_component_id[component_id]
if assignment.provider is ExecutionProvider.MODAL:
summary.execution_worker_indices_by_representative[component_id] = 0
continue
if assignment.provider is ExecutionProvider.VAST:
summary.execution_worker_indices_by_representative[component_id] = 0
continue
if execution_plan.configuration_set is not None:
summary.execution_worker_indices_by_representative[component_id] = (
assignment.capacity_slot_index
)
logger.info(
"Assigned workflow-configured remote component=%s environment=%s "
"worker_index=%d.",
component_id,
assignment.environment_id,
assignment.capacity_slot_index,
)
continue
host = ssh_hosts_by_id.get(assignment.environment_id)
maximum_workers = max(1, host.maximum_workers if host is not None else 1)
if component_id in session_component_ids:
worker_index = 0
else:
first_worker = (
1
if assignment.environment_id in session_environment_ids
and maximum_workers > 1
else 0
)
available_workers = max(1, maximum_workers - first_worker)
worker_index = first_worker + (
worker_counts_by_environment[assignment.environment_id]
% available_workers
)
worker_counts_by_environment[assignment.environment_id] += 1
summary.execution_worker_indices_by_representative[component_id] = worker_index
logger.info(
"Assigned remote component=%s environment=%s worker_index=%d.",
component_id,
assignment.environment_id,
worker_index,
)
for assignment in assignments_by_component_id.values():
if assignment.environment_id in sync_engines_by_environment:
continue
if assignment.provider is ExecutionProvider.MODAL:
_ensure_remote_sync_backend(resolved_settings, resolved_sync_engine)
sync_engines_by_environment[
assignment.environment_id
] = resolved_sync_engine
continue
if assignment.provider is ExecutionProvider.VAST:
if vast_service is None:
raise ModalPromptValidationError(
"Vast.ai assignment has no active controller service."
)
lease = vast_leases_by_environment[assignment.environment_id]
vast_sync_engine = vast_service.sync_engine(lease)
vast_sync_engine.cancellation_check = cancellation_check
sync_engines_by_environment[assignment.environment_id] = vast_sync_engine
continue
host = ssh_hosts_by_id.get(assignment.environment_id)
if host is None:
raise ModalPromptValidationError(
f"Assigned SSH execution environment {assignment.environment_id!r} is no longer configured."
)
sync_engines_by_environment[assignment.environment_id] = _ssh_sync_engine(
host=host,
settings=resolved_settings,
r2_cache=workflow_r2_cache,
)
if status_callback is not None:
status_callback("Preparing assets for remote execution", None, None)
try:
environment_preparations = _prepare_remote_environment_assets(
components=components,
assignments_by_component_id=assignments_by_component_id,
sync_engines_by_environment=sync_engines_by_environment,
rewritten_prompt=rewritten_prompt,
sync_custom_nodes=resolved_settings.sync_custom_nodes,
status_callback=status_callback,
environment_status_callback=environment_status_callback,
)
except R2CacheError as exc:
raise ModalPromptValidationError(str(exc)) from exc
if resolved_settings.sync_custom_nodes:
summary.custom_nodes_bundles_by_environment = {
environment_id: preparation.custom_nodes_bundle
for environment_id, preparation in environment_preparations.items()
}
modal_bundle = next(
(
summary.custom_nodes_bundles_by_environment[assignment.environment_id]
for assignment in assignments_by_component_id.values()
if assignment.provider is ExecutionProvider.MODAL
),
None,
)
summary.custom_nodes_bundle = modal_bundle or next(
iter(summary.custom_nodes_bundles_by_environment.values()),
None,
)
else:
logger.info(
"Skipping custom_nodes bundle sync because sync is disabled for execution_mode=%s.",
resolved_settings.execution_mode,
)
synced_component_prompts: dict[str, dict[str, Any]] = {}
synced_assets_by_component_id: dict[str, list[SyncedAsset]] = {}
for component in components:
assignment = assignments_by_component_id[component.representative_node_id]
preparation = environment_preparations[assignment.environment_id]
component_id = component.representative_node_id
synced_component_prompts[component_id] = preparation.component_prompts[
component_id
]
synced_assets_by_component_id[component_id] = (
preparation.assets_by_component_id[component_id]
)
summary.synced_assets.extend(
preparation.assets_by_component_id[component_id]
)
summary.synced_assets = _deduplicate_synced_assets(summary.synced_assets)
modal_environment_ids = {
assignment.environment_id
for assignment in assignments_by_component_id.values()
if assignment.provider is ExecutionProvider.MODAL
}
modal_component_ids = {
component.representative_node_id
for component in components
if assignments_by_component_id[component.representative_node_id].provider
is ExecutionProvider.MODAL
}
requires_volume_reload = any(
asset.uploaded
for component_id, assets in synced_assets_by_component_id.items()
if component_id in modal_component_ids
for asset in assets
) or any(
bundle is not None and bundle.uploaded
for environment_id, bundle in summary.custom_nodes_bundles_by_environment.items()
if environment_id in modal_environment_ids
)
volume_reload_marker = uuid.uuid4().hex if requires_volume_reload else None
logger.info(
"Resolved request-wide Modal volume reload requirement: requires_volume_reload=%s volume_reload_marker=%s synced_assets=%d custom_nodes_uploaded=%s",
requires_volume_reload,
volume_reload_marker,
len(summary.synced_assets),
bool(
summary.custom_nodes_bundle is not None
and summary.custom_nodes_bundle.uploaded
),
)
summary.requires_volume_reload = requires_volume_reload
summary.volume_reload_marker = volume_reload_marker
summary.uploaded_volume_paths = [
asset.remote_path for asset in summary.synced_assets if asset.uploaded
]
mapped_proxy_component_ids: set[str] = set()
sandwiched_local_node_id_set = set(summary.sandwiched_local_node_ids)
for component in components:
logger.info(
"Rewriting remote component %s covering nodes %s.",
component.representative_node_id,
component.node_ids,
)
assignment = assignments_by_component_id[component.representative_node_id]
component_custom_nodes_bundle = summary.custom_nodes_bundles_by_environment.get(
assignment.environment_id
)
uploaded_volume_paths = _component_uploaded_volume_paths(
component_prompt=synced_component_prompts[component.representative_node_id],
synced_assets=synced_assets_by_component_id[
component.representative_node_id
],
custom_nodes_bundle=component_custom_nodes_bundle,
)
payload = _build_component_payload(
component=component,
component_prompt=synced_component_prompts[component.representative_node_id],
signature_prompt=prompt,
extra_data=extra_data,
settings=resolved_settings,
requires_volume_reload=(
assignment.provider is ExecutionProvider.MODAL
and bool(uploaded_volume_paths)
),
volume_reload_marker=volume_reload_marker,
custom_nodes_bundle=component_custom_nodes_bundle,
uploaded_volume_paths=uploaded_volume_paths,
terminate_container_on_error=resolved_settings.terminate_container_on_error,
nodes_module=resolved_nodes_module,
remote_session=remote_sessions_by_component_id.get(
component.representative_node_id
),
)
_attach_resolved_llm_profiles(
payload,
execution_plan.resolved_llm_profiles,
resolved_settings,
)
_stamp_execution_assignment(
payload,
assignment,
summary.execution_worker_indices_by_representative.get(
component.representative_node_id,
0,
),
_component_execution_signature(component, prompt),
_execution_location_for_assignment(
assignment,
ssh_hosts_by_id,
vast_leases_by_environment,
),
_configured_provider_metadata(
execution_plan=execution_plan,
assignment=assignment,
vast_leases_by_environment=vast_leases_by_environment,
),
)
implicitly_mapped_output_sources = _implicitly_mapped_boundary_output_sources(
component=component,
original_prompt=prompt,
rewritten_prompt=rewritten_prompt,
nodes_module=resolved_nodes_module,
)
_mark_payload_scheduler_list_outputs(
payload,
implicitly_mapped_output_sources,
)
proxy_node_ids = _rewrite_component_into_proxy(
component=component,
rewritten_prompt=rewritten_prompt,
payload=payload,
nodes_module=resolved_nodes_module,
)
for proxy_node_id in proxy_node_ids:
summary.execution_assignments_by_representative[proxy_node_id] = assignment
summary.execution_worker_indices_by_representative[proxy_node_id] = (
summary.execution_worker_indices_by_representative.get(
component.representative_node_id,
0,
)
)
split_proxy_payloads = payload.get("split_proxy_payloads")
if isinstance(split_proxy_payloads, dict):
static_proxy_node_id, mapped_proxy_node_id = proxy_node_ids
summary.remote_component_ids.extend(proxy_node_ids)
summary.component_node_ids_by_representative[static_proxy_node_id] = (
_workflow_visible_remote_node_ids(component.static_node_ids)
)
summary.component_node_ids_by_representative[mapped_proxy_node_id] = (
_workflow_visible_remote_node_ids(component.mapped_node_ids)
)
for node_id in _workflow_visible_remote_node_ids(
component.static_node_ids
):
summary.rewritten_node_id_map[node_id] = static_proxy_node_id
for node_id in _workflow_visible_remote_node_ids(
component.mapped_node_ids
):
summary.rewritten_node_id_map[node_id] = mapped_proxy_node_id
mapped_proxy_component_ids.add(mapped_proxy_node_id)
continue
if isinstance(split_proxy_payloads, list):
summary.remote_component_ids.extend(proxy_node_ids)
mapped_node_id_set = set(component.mapped_node_ids)
for phase_payload in split_proxy_payloads:
phase_proxy_node_id = str(phase_payload["component_id"])
phase_component_node_ids = [
str(node_id) for node_id in phase_payload["component_node_ids"]
]
summary.component_node_ids_by_representative[
phase_proxy_node_id
] = _workflow_visible_remote_node_ids(phase_component_node_ids)
for node_id in _workflow_visible_remote_node_ids(
phase_component_node_ids
):
summary.rewritten_node_id_map[node_id] = phase_proxy_node_id
if mapped_node_id_set and mapped_node_id_set.intersection(
phase_component_node_ids
):
mapped_proxy_component_ids.add(phase_proxy_node_id)
continue
summary.remote_component_ids.extend(proxy_node_ids)
summary.component_node_ids_by_representative[proxy_node_ids[0]] = (
_workflow_visible_remote_node_ids(component.node_ids)
)
for node_id in _workflow_visible_remote_node_ids(component.node_ids):
summary.rewritten_node_id_map[node_id] = proxy_node_ids[0]
if component.mapped_boundary_input_name or implicitly_mapped_output_sources:
mapped_proxy_component_ids.add(proxy_node_ids[0])
_configure_local_gap_keepalive_payloads(
rewritten_prompt=rewritten_prompt,
remote_component_ids=summary.remote_component_ids,
sandwiched_local_node_ids=sandwiched_local_node_id_set,
)
summary.parallel_local_branch_node_ids = _parallelize_non_returning_local_branches(
rewritten_prompt=rewritten_prompt,
remote_component_ids=summary.remote_component_ids,
nodes_module=resolved_nodes_module,
)
summary.artifact_finalizer_node_id = _attach_modal_artifact_finalizer(
rewritten_prompt=rewritten_prompt,
remote_component_ids=summary.remote_component_ids,
nodes_module=resolved_nodes_module,
)
proxy_component_groups = {
component_id: {component_id} for component_id in summary.remote_component_ids
}
_, dependency_edges, _ = _component_dependency_graph(
rewritten_prompt, proxy_component_groups
)
execution_stages = _component_execution_stages(
rewritten_prompt, proxy_component_groups
)
_configure_speculative_affinity_prewarm_payloads(
rewritten_prompt=rewritten_prompt,
execution_stages=execution_stages,
)
summary.component_dependency_ids_by_representative = {
representative_node_id: sorted(
upstream_component_id
for upstream_component_id, downstream_component_ids in dependency_edges.items()
if representative_node_id in downstream_component_ids
)
for representative_node_id in sorted(proxy_component_groups)
}
summary.component_execution_stages = [list(stage) for stage in execution_stages]
summary.mapped_component_ids = sorted(mapped_proxy_component_ids)
summary.estimated_max_parallel_requests = _estimated_stage_parallelism(
execution_stages,
mapped_proxy_component_ids,
mapped_component_weight=1,
)
if execution_plan.configuration_set is not None:
configured_capacity = sum(
configuration.capacity_limit
for configuration in execution_plan.configuration_set.capacity_configurations
)
summary.max_parallel_requests_upper_bound = min(
summary.estimated_max_parallel_requests,
configured_capacity,
)
elif resolved_settings.max_containers is not None:
summary.max_parallel_requests_upper_bound = min(
summary.estimated_max_parallel_requests,
resolved_settings.max_containers,
)
else:
summary.max_parallel_requests_upper_bound = (
summary.estimated_max_parallel_requests
)
logger.info(
"Estimated remote parallelism after proxy rewrite: known_max_parallel_requests=%d max_parallel_requests_upper_bound=%s mapped_components=%s execution_stages=%s",
summary.estimated_max_parallel_requests,
summary.max_parallel_requests_upper_bound,
summary.mapped_component_ids,
summary.component_execution_stages,
)
_log_modal_rewritten_prompt_diagnostics(
prompt_id=(
str(extra_data.get("prompt_id"))
if isinstance(extra_data, Mapping) and extra_data.get("prompt_id")
else None
),
prompt=rewritten_prompt,
summary=summary,
reason="post_rewrite",
)
return rewritten_prompt, summary
async def rewrite_prompt_for_modal_async(
prompt: dict[str, Any],
workflow: dict[str, Any] | None,
sync_engine: ModalAssetSyncEngine | None = None,
settings: ModalSyncSettings | None = None,
nodes_module: Any | None = None,
extra_data: dict[str, Any] | None = None,
status_callback: SetupStatusCallback | None = None,
environment_status_callback: EnvironmentSetupStatusCallback | None = None,
plan_callback: ExecutionPlanStatusCallback | None = None,
cancellation_check: Callable[[], bool] | None = None,
occupied_environment_ids: frozenset[str] = frozenset(),
) -> tuple[dict[str, Any], RewriteSummary]:
"""Prepare one Modal prompt without blocking ComfyUI's event loop."""
return await asyncio.to_thread(
rewrite_prompt_for_modal,
prompt=prompt,
workflow=workflow,
sync_engine=sync_engine,
settings=settings,
nodes_module=nodes_module,
extra_data=extra_data,
status_callback=status_callback,
environment_status_callback=environment_status_callback,
plan_callback=plan_callback,
cancellation_check=cancellation_check,
occupied_environment_ids=occupied_environment_ids,
)
def _execution_assignments_payload(
summary: RewriteSummary,
settings: ModalSyncSettings,
) -> dict[str, dict[str, Any]]:
"""Return planner assignments with member nodes and safe runtime labels."""
ssh_hosts_by_id = {
host.environment_id: host for host in _configured_ssh_hosts(settings)
}
vast_leases_by_environment: dict[str, Any] = {}
vast_registry = _vast_registry(settings)
if vast_registry is not None:
try:
vast_leases_by_environment = {
lease.environment_id: lease for lease in vast_registry.load().leases
}
except ValueError as exc:
logger.warning("Unable to load Vast lease labels for UI status: %s", exc)
configurations_by_id = {
str(configuration.get("configuration_id") or ""): configuration
for configuration in summary.remote_configurations
}
return {
component_id: {
"provider": assignment.provider.value,
"environment_id": assignment.environment_id,
"configuration_id": assignment.configuration_id,
"execution_location": _execution_location_for_assignment(
assignment, ssh_hosts_by_id, vast_leases_by_environment
)
or summary.execution_locations_by_environment.get(
assignment.environment_id
),
"node_ids": list(
summary.component_node_ids_by_representative.get(component_id, [])
),
"predicted_cost_usd": assignment.predicted_cost_usd,
"predicted_completion_seconds": assignment.predicted_completion_seconds,
"worker_index": summary.execution_worker_indices_by_representative.get(
component_id,
0,
),
"reasons": list(assignment.reasons),
"hardware": _assignment_hardware_payload(
component_id=component_id,
assignment=assignment,
configurations_by_id=configurations_by_id,
ssh_hosts_by_id=ssh_hosts_by_id,
vast_quotes={},
vast_leases_by_environment=vast_leases_by_environment,
),
}
for component_id, assignment in sorted(
summary.execution_assignments_by_representative.items()
)
}
def _prompt_uses_remote_execution_configurator(prompt: Mapping[str, Any]) -> bool:
"""Return whether the serialized prompt opts into workflow-scoped capacity."""
return _remote_execution_configurator_node_id(prompt) is not None
def _remote_execution_configurator_node_id(
prompt: Mapping[str, Any],
) -> str | None:
"""Return the serialized identity of the workflow's sole configurator node."""
configurator_node_ids = [
str(node_id)
for node_id, prompt_node in prompt.items()
if isinstance(prompt_node, Mapping)
and str(prompt_node.get("class_type") or "")
== REMOTE_EXECUTION_CONFIGURATOR_NODE_ID
]
return configurator_node_ids[0] if len(configurator_node_ids) == 1 else None
def _selected_modal_gpus(
summary: RewriteSummary,
legacy_modal_gpu: str,
) -> list[str]:
"""Return distinct Modal GPUs selected by the completed execution plan."""
modal_assignments = [
assignment
for assignment in summary.execution_assignments_by_representative.values()
if assignment.provider is ExecutionProvider.MODAL
]
if not modal_assignments:
return []
configured_gpus = {
str(configuration.get("configuration_id") or ""): str(
configuration.get("gpu_type") or ""
)
for configuration in summary.remote_configurations
if configuration.get("provider") == ExecutionProvider.MODAL.value
and configuration.get("configuration_id")
and configuration.get("gpu_type")
}
selected_gpus = {
configured_gpus.get(str(assignment.configuration_id or ""), "")
for assignment in modal_assignments
}
selected_gpus.discard("")
if not selected_gpus:
selected_gpus.add(legacy_modal_gpu)
return sorted(selected_gpus)
async def _queue_prompt_json(
prompt_server: Any,
json_data: dict[str, Any],
modal_response_payload: dict[str, Any] | None = None,
) -> web.Response:
"""Queue a possibly rewritten prompt using ComfyUI's native semantics."""
execution = _get_execution_module()
json_data = prompt_server.trigger_on_prompt(json_data)
if "number" in json_data:
number = float(json_data["number"])
else:
number = prompt_server.number
if json_data.get("front"):
number = -number
prompt_server.number += 1
if "prompt" not in json_data:
return web.json_response(
{
"error": {
"type": "no_prompt",
"message": "No prompt provided",
"details": "No prompt provided",
"extra_info": {},
}
},
status=400,
)
prompt = json_data["prompt"]
prompt_id = str(json_data.get("prompt_id", uuid.uuid4()))
partial_execution_targets = json_data.get("partial_execution_targets")
extra_data = dict(json_data.get("extra_data", {}))
if "client_id" in json_data:
extra_data["client_id"] = json_data["client_id"]
valid = await execution.validate_prompt(
prompt_id, prompt, partial_execution_targets
)
if not valid[0]:
modal_extra = extra_data.get("modal")
if isinstance(modal_extra, Mapping) and modal_extra.get("remote_component_ids"):
logger.warning(
"ComfyUI rejected rewritten Modal prompt prompt_id=%s error=%s node_errors=%s",
prompt_id,
valid[1],
valid[3],
)
_log_modal_rewritten_prompt_diagnostics(
prompt_id=prompt_id,
prompt=prompt,
reason="comfy_validation_failure",
level=logging.WARNING,
)
else:
logger.warning("invalid prompt: %s", valid[1])
return web.json_response(
{"error": valid[1], "node_errors": valid[3]}, status=400
)
outputs_to_execute = valid[2]
sensitive: dict[str, Any] = {}
for sensitive_key in execution.SENSITIVE_EXTRA_DATA_KEYS:
if sensitive_key in extra_data:
sensitive[sensitive_key] = extra_data.pop(sensitive_key)
extra_data["create_time"] = int(time.time() * 1000)
prompt_server.prompt_queue.put(
(number, prompt_id, prompt, extra_data, outputs_to_execute, sensitive)
)
response_payload: dict[str, Any] = {
"prompt_id": prompt_id,
"number": number,
"node_errors": valid[3],
}
if modal_response_payload:
response_payload.update(modal_response_payload)
return web.json_response(response_payload)
def _analysis_route_path(route_path: str) -> str:
"""Return the sibling HTTP route used for dry-run remote-node expansion."""
if route_path.endswith("/queue_prompt"):
return f"{route_path.removesuffix('/queue_prompt')}/analyze_remote_nodes"
return f"{route_path.rstrip('/')}/analyze_remote_nodes"
def _progress_state_route_path(route_path: str) -> str:
"""Return the sibling HTTP route used for Modal UI event replay."""
if route_path.endswith("/queue_prompt"):
return f"{route_path.removesuffix('/queue_prompt')}/progress_state"
return f"{route_path.rstrip('/')}/progress_state"
def _container_status_route_path(route_path: str) -> str:
"""Return the sibling HTTP route used for active Modal container status."""
if route_path.endswith("/queue_prompt"):
return f"{route_path.removesuffix('/queue_prompt')}/container_status"
return f"{route_path.rstrip('/')}/container_status"
def _delete_modal_caches_route_path(route_path: str) -> str:
"""Return the sibling HTTP route used to delete persistent Modal cache Dicts."""
if route_path.endswith("/queue_prompt"):
return f"{route_path.removesuffix('/queue_prompt')}/delete_caches"
return f"{route_path.rstrip('/')}/delete_caches"
def _delete_modal_volume_route_path(route_path: str) -> str:
"""Return the sibling HTTP route used to delete the configured Modal Volume."""
if route_path.endswith("/queue_prompt"):
return f"{route_path.removesuffix('/queue_prompt')}/delete_volume"
return f"{route_path.rstrip('/')}/delete_volume"
def _cancel_preparation_route_path(route_path: str) -> str:
"""Return the sibling route used to cancel queue-time remote preparation."""
if route_path.endswith("/queue_prompt"):
return f"{route_path.removesuffix('/queue_prompt')}/cancel_preparation"
return f"{route_path.rstrip('/')}/cancel_preparation"
def _remote_environments_route_path(route_path: str) -> str:
"""Return the provider-neutral environment management route."""
del route_path
return "/remote/environments"
def _remote_environment_probe_route_path(route_path: str) -> str:
"""Return the provider-neutral host capability probe route."""
del route_path
return "/remote/environments/probe"
def _remote_environment_bootstrap_route_path(route_path: str) -> str:
"""Return the provider-neutral worker bootstrap route."""
del route_path
return "/remote/environments/bootstrap"
def _remote_environment_status_route_path(route_path: str) -> str:
"""Return the provider-neutral worker status route."""
del route_path
return "/remote/environments/status"
def _remote_environment_stop_route_path(route_path: str) -> str:
"""Return the provider-neutral worker shutdown route."""
del route_path
return "/remote/environments/stop"
def _vast_registry(settings: ModalSyncSettings) -> VastLeaseRegistry | None:
"""Return persistent credential-free Vast lease state when available."""
user_directory = discover_comfyui_user_directory(settings)
if user_directory is None:
return None
return VastLeaseRegistry.for_user_directory(user_directory)
def _modal_not_found_error_types() -> tuple[type[BaseException], ...]:
"""Return Modal SDK exception classes that mean a named object is absent."""
if modal is None:
return ()
exception_namespace = getattr(modal, "exception", None)
candidates = [
getattr(exception_namespace, "NotFoundError", None),
getattr(exception_namespace, "InvalidError", None),
]
return tuple(
candidate
for candidate in candidates
if isinstance(candidate, type) and issubclass(candidate, BaseException)
)
def _modal_cache_dict_names(settings: ModalSyncSettings) -> list[str]:
"""Return the configured persistent Modal Dict names used as local-reset caches."""
return [
settings.interrupt_dict_name,
settings.node_output_cache_dict_name,
settings.session_bridge_dict_name,
settings.sync_index_dict_name,
settings.snapshot_profile_dict_name,
]
async def _call_modal_sdk(method: Any, *args: Any, **kwargs: Any) -> Any:
"""Call a Modal SDK method without blocking the aiohttp event loop."""
async_method = getattr(method, "aio", None)
if callable(async_method):
return await async_method(*args, **kwargs)
return await asyncio.to_thread(method, *args, **kwargs)
async def _delete_modal_named_object(
namespace: Any, name: str, *, object_label: str
) -> None:
"""Delete a named Modal object using the non-deprecated manager API when available."""
objects_manager = getattr(namespace, "objects", None)
manager_delete = getattr(objects_manager, "delete", None)
if callable(manager_delete):
await _call_modal_sdk(manager_delete, name, allow_missing=True)
return
instance = await _call_modal_sdk(namespace.from_name, name, create_if_missing=False)
delete_method = getattr(instance, "delete", None)
if not callable(delete_method):
raise RuntimeError(f"Modal {object_label} {name!r} does not expose delete().")
await _call_modal_sdk(delete_method, name)
async def delete_modal_cache_dicts(settings: ModalSyncSettings) -> dict[str, Any]:
"""Delete all configured Modal Dict caches and return a reset summary."""
if modal is None:
raise RuntimeError("Modal SDK is unavailable; cannot delete Modal caches.")
modal_dict = getattr(modal, "Dict", None)
if modal_dict is None:
raise RuntimeError(
"Modal SDK does not expose modal.Dict; cannot delete Modal caches."
)
deleted: list[str] = []
skipped: list[str] = []
not_found_errors = _modal_not_found_error_types()
for dict_name in _modal_cache_dict_names(settings):
try:
await _call_modal_sdk(
modal_dict.from_name,
dict_name,
create_if_missing=False,
)
except not_found_errors:
skipped.append(dict_name)
continue
await _delete_modal_named_object(modal_dict, dict_name, object_label="Dict")
deleted.append(dict_name)
logger.info("Deleted Modal cache Dicts deleted=%s skipped=%s.", deleted, skipped)
return {"deleted": deleted, "skipped": skipped}
async def delete_modal_volume(settings: ModalSyncSettings) -> dict[str, Any]:
"""Delete the configured Modal Volume and return a reset summary."""
if modal is None:
raise RuntimeError("Modal SDK is unavailable; cannot delete Modal volume.")
modal_volume = getattr(modal, "Volume", None)
if modal_volume is None:
raise RuntimeError(
"Modal SDK does not expose modal.Volume; cannot delete Modal volume."
)
try:
await _call_modal_sdk(
modal_volume.from_name,
settings.volume_name,
create_if_missing=False,
)
except _modal_not_found_error_types():
logger.info("Skipped deleting missing Modal Volume %s.", settings.volume_name)
return {"deleted": [], "skipped": [settings.volume_name]}
await _delete_modal_named_object(
modal_volume, settings.volume_name, object_label="Volume"
)
logger.info("Deleted Modal Volume %s.", settings.volume_name)
return {"deleted": [settings.volume_name], "skipped": []}
def _install_modal_interrupt_queue_bridge(prompt_server: Any) -> None:
"""Expose active remote work through every ComfyUI queue-state view."""
prompt_queue = getattr(prompt_server, "prompt_queue", None)
if prompt_queue is None or getattr(
prompt_queue, _MODAL_INTERRUPT_QUEUE_BRIDGE_ATTR, False
):
return
original_get_current_queue = getattr(prompt_queue, "get_current_queue", None)
original_get_current_queue_volatile = getattr(
prompt_queue, "get_current_queue_volatile", None
)
original_get_tasks_remaining = getattr(prompt_queue, "get_tasks_remaining", None)
original_interrupt_if_running = getattr(
prompt_queue, "interrupt_if_running", None
)
original_task_done = getattr(prompt_queue, "task_done", None)
original_wipe_queue = getattr(prompt_queue, "wipe_queue", None)
original_delete_queue_item = getattr(prompt_queue, "delete_queue_item", None)
if not any(
callable(method)
for method in (
original_get_current_queue,
original_get_current_queue_volatile,
original_get_tasks_remaining,
)
):
logger.debug(
"Prompt queue does not expose queue-state methods; skipping remote queue bridge."
)
return
preparation_prompts: dict[str, tuple[Any, ...]] = {}
preparation_cancellations: dict[str, threading.Event] = {}
preparation_lock = threading.RLock()
setattr(prompt_queue, _REMOTE_PREPARATION_PROMPTS_ATTR, preparation_prompts)
setattr(
prompt_queue,
_REMOTE_PREPARATION_CANCELLATIONS_ATTR,
preparation_cancellations,
)
setattr(prompt_queue, _REMOTE_PREPARATION_LOCK_ATTR, preparation_lock)
def preparation_items() -> list[tuple[Any, ...]]:
"""Return a stable snapshot of prompts still preparing remote capacity."""
with preparation_lock:
return list(preparation_prompts.values())
def append_missing_preparations(
running: Iterable[Any],
queued: Iterable[Any],
) -> tuple[list[Any], Any]:
"""Add preparation entries that are not already in ComfyUI's native queue."""
running_items = list(running)
queued_items = list(queued)
native_prompt_ids = _queue_item_prompt_ids((*running_items, *queued_items))
running_items.extend(
item
for item in preparation_items()
if str(item[1]) not in native_prompt_ids
)
return running_items, queued
if callable(original_get_current_queue):
def remote_get_current_queue() -> tuple[list[Any], Any]:
"""Return native work plus preparation and active remote prompt entries."""
running, queued = original_get_current_queue()
running_items, queued = append_missing_preparations(running, queued)
running_prompt_ids = _queue_item_prompt_ids(running_items)
try:
from .remote.modal_app import active_remote_modal_prompt_ids
except ImportError:
return running_items, queued
for prompt_id in sorted(
active_remote_modal_prompt_ids() - running_prompt_ids
):
running_items.append((0, prompt_id, {}, {}, [], {}))
return running_items, queued
setattr(prompt_queue, "get_current_queue", remote_get_current_queue)
if callable(original_get_current_queue_volatile):
def remote_get_current_queue_volatile() -> tuple[list[Any], Any]:
"""Include remote preparation in ComfyUI's public `/queue` response."""
running, queued = original_get_current_queue_volatile()
return append_missing_preparations(running, queued)
setattr(
prompt_queue,
"get_current_queue_volatile",
remote_get_current_queue_volatile,
)
if callable(original_get_tasks_remaining):
def remote_get_tasks_remaining() -> int:
"""Count remote preparation as work in websocket queue status."""
remaining = int(original_get_tasks_remaining())
native_prompt_ids: set[str] = set()
native_queue_method = (
original_get_current_queue_volatile
if callable(original_get_current_queue_volatile)
else original_get_current_queue
)
if callable(native_queue_method):
running, queued = native_queue_method()
native_prompt_ids = _queue_item_prompt_ids((*running, *queued))
return remaining + sum(
str(item[1]) not in native_prompt_ids
for item in preparation_items()
)
setattr(prompt_queue, "get_tasks_remaining", remote_get_tasks_remaining)
if callable(original_interrupt_if_running):
def remote_interrupt_if_running(prompt_id: str) -> bool:
"""Cancel remote preparation or interrupt matching native execution."""
if _cancel_remote_preparation(prompt_server, prompt_id):
return True
return bool(original_interrupt_if_running(prompt_id))
setattr(prompt_queue, "interrupt_if_running", remote_interrupt_if_running)
if callable(original_task_done):
def remote_task_done(item_id: Any, *args: Any, **kwargs: Any) -> Any:
"""Release background cache work after the whole prompt terminates."""
currently_running = getattr(prompt_queue, "currently_running", {})
running_item = (
currently_running.get(item_id)
if isinstance(currently_running, Mapping)
else None
)
prompt_id = (
str(running_item[1])
if isinstance(running_item, (list, tuple)) and len(running_item) > 1
else None
)
try:
return original_task_done(item_id, *args, **kwargs)
finally:
if prompt_id is not None:
finish_r2_writeback_prompt(prompt_id)
setattr(prompt_queue, "task_done", remote_task_done)
if callable(original_wipe_queue):
def remote_wipe_queue() -> Any:
"""Release reservations belonging to every discarded queued prompt."""
queued_items: list[Any] = []
if callable(original_get_current_queue):
_running, queued = original_get_current_queue()
queued_items = list(queued)
try:
return original_wipe_queue()
finally:
for queued_prompt_id in _queue_item_prompt_ids(queued_items):
finish_r2_writeback_prompt(queued_prompt_id)
setattr(prompt_queue, "wipe_queue", remote_wipe_queue)
if callable(original_delete_queue_item):
def remote_delete_queue_item(predicate: Callable[[Any], bool]) -> Any:
"""Release the exact reservation removed through ComfyUI's queue API."""
before_prompt_ids: set[str] = set()
if callable(original_get_current_queue):
_running, queued = original_get_current_queue()
before_prompt_ids = _queue_item_prompt_ids(queued)
result = original_delete_queue_item(predicate)
if result and callable(original_get_current_queue):
_running, queued = original_get_current_queue()
after_prompt_ids = _queue_item_prompt_ids(queued)
for removed_prompt_id in before_prompt_ids - after_prompt_ids:
finish_r2_writeback_prompt(removed_prompt_id)
return result
setattr(prompt_queue, "delete_queue_item", remote_delete_queue_item)
setattr(prompt_queue, _MODAL_INTERRUPT_QUEUE_BRIDGE_ATTR, True)
logger.info("Installed remote preparation bridge on ComfyUI prompt queue.")
def _queue_item_prompt_ids(items: Iterable[Any]) -> set[str]:
"""Return prompt IDs from well-formed native or synthetic queue items."""
return {
str(item[1])
for item in items
if isinstance(item, (list, tuple)) and len(item) > 1
}
def _cancel_remote_preparation(prompt_server: Any, prompt_id: str) -> bool:
"""Signal cancellation when the prompt is preparing remote capacity."""
normalized_prompt_id = str(prompt_id).strip()
prompt_queue = getattr(prompt_server, "prompt_queue", None)
cancellations = getattr(
prompt_queue,
_REMOTE_PREPARATION_CANCELLATIONS_ATTR,
None,
)
preparation_lock = getattr(prompt_queue, _REMOTE_PREPARATION_LOCK_ATTR, None)
if (
not normalized_prompt_id
or not isinstance(cancellations, dict)
or preparation_lock is None
):
return False
with preparation_lock:
cancellation_event = cancellations.get(normalized_prompt_id)
if cancellation_event is None:
return False
cancellation_event.set()
logger.info(
"Cancelled remote preparation for prompt %s.", normalized_prompt_id
)
return True
def _queued_ssh_environment_ids(
prompt_server: Any,
*,
excluding_prompt_id: str | None = None,
) -> frozenset[str]:
"""Return SSH environments reserved by prompts already in ComfyUI's queue."""
prompt_queue = getattr(prompt_server, "prompt_queue", None)
get_current_queue = getattr(prompt_queue, "get_current_queue", None)
if not callable(get_current_queue):
return frozenset()
queue_state = get_current_queue()
if not isinstance(queue_state, (list, tuple)) or len(queue_state) != 2:
return frozenset()
running, queued = queue_state
queue_items = [
item
for collection in (running, queued)
if isinstance(collection, (list, tuple))
for item in collection
]
environment_ids: set[str] = set()
for item in queue_items:
if not isinstance(item, (list, tuple)) or len(item) <= 3:
continue
prompt_id = str(item[1]) if len(item) > 1 else ""
if excluding_prompt_id is not None and prompt_id == excluding_prompt_id:
continue
extra_data = item[3]
if not isinstance(extra_data, Mapping):
continue
remote_execution = extra_data.get("remote_execution")
if not isinstance(remote_execution, Mapping):
continue
assignments = remote_execution.get("assignments")
if not isinstance(assignments, Mapping):
continue
for assignment in assignments.values():
if not isinstance(assignment, Mapping):
continue
provider = str(assignment.get("provider") or "").strip().lower()
environment_id = str(assignment.get("environment_id") or "").strip()
if provider == ExecutionProvider.SSH_DOCKER.value and environment_id:
environment_ids.add(environment_id)
return frozenset(environment_ids)
def _set_remote_preparation(
prompt_server: Any,
*,
prompt_id: str,
prompt: Mapping[str, Any],
extra_data: Mapping[str, Any],
cancellation_event: threading.Event | None = None,
) -> bool:
"""Register one pre-queue remote workflow and publish its active queue state."""
prompt_queue = getattr(prompt_server, "prompt_queue", None)
preparations = getattr(prompt_queue, _REMOTE_PREPARATION_PROMPTS_ATTR, None)
preparation_lock = getattr(prompt_queue, _REMOTE_PREPARATION_LOCK_ATTR, None)
if not isinstance(preparations, dict) or preparation_lock is None:
return False
preparation_extra_data = copy.deepcopy(dict(extra_data))
preparation_extra_data.setdefault("create_time", int(time.time() * 1000))
with preparation_lock:
preparations[prompt_id] = (
0,
prompt_id,
copy.deepcopy(dict(prompt)),
preparation_extra_data,
[],
{},
)
cancellations = getattr(
prompt_queue,
_REMOTE_PREPARATION_CANCELLATIONS_ATTR,
None,
)
if isinstance(cancellations, dict) and cancellation_event is not None:
cancellations[prompt_id] = cancellation_event
queue_updated = getattr(prompt_server, "queue_updated", None)
if callable(queue_updated):
queue_updated()
return True
def _clear_remote_preparation(prompt_server: Any, prompt_id: str) -> None:
"""Remove one pre-queue remote workflow and publish the resulting queue state."""
prompt_queue = getattr(prompt_server, "prompt_queue", None)
preparations = getattr(prompt_queue, _REMOTE_PREPARATION_PROMPTS_ATTR, None)
preparation_lock = getattr(prompt_queue, _REMOTE_PREPARATION_LOCK_ATTR, None)
if not isinstance(preparations, dict) or preparation_lock is None:
return
with preparation_lock:
removed = preparations.pop(prompt_id, None)
cancellations = getattr(
prompt_queue,
_REMOTE_PREPARATION_CANCELLATIONS_ATTR,
None,
)
if isinstance(cancellations, dict):
cancellations.pop(prompt_id, None)
if removed is not None:
queue_updated = getattr(prompt_server, "queue_updated", None)
if callable(queue_updated):
queue_updated()
def setup_modal_queue_route(
prompt_server: Any | None = None,
sync_engine: ModalAssetSyncEngine | None = None,
settings: ModalSyncSettings | None = None,
) -> None:
"""Register the `/modal/queue_prompt` route once for the active PromptServer."""
global _ROUTE_REGISTERED
if _ROUTE_REGISTERED:
return
try:
resolved_server_module = _get_server_module()
except ModuleNotFoundError:
logger.debug(
"ComfyUI server module is not available; skipping route registration."
)
return
resolved_settings = settings or get_settings()
prompt_server = prompt_server or getattr(
resolved_server_module.PromptServer, "instance", None
)
if prompt_server is None:
logger.debug(
"PromptServer.instance is not available; skipping route registration."
)
return
setup_r2_oauth_routes(prompt_server)
resolved_sync_engine = sync_engine or ModalAssetSyncEngine.from_environment(
resolved_settings
)
analysis_route_path = _analysis_route_path(resolved_settings.route_path)
progress_state_route_path = _progress_state_route_path(resolved_settings.route_path)
container_status_route_path = _container_status_route_path(
resolved_settings.route_path
)
container_stop_route_path = container_status_route_path.replace(
"/container_status", "/container_stop"
)
delete_caches_route_path = _delete_modal_caches_route_path(
resolved_settings.route_path
)
delete_volume_route_path = _delete_modal_volume_route_path(
resolved_settings.route_path
)
cancel_preparation_route_path = _cancel_preparation_route_path(
resolved_settings.route_path
)
remote_environments_route_path = _remote_environments_route_path(
resolved_settings.route_path
)
remote_environment_probe_route_path = _remote_environment_probe_route_path(
resolved_settings.route_path
)
remote_environment_bootstrap_route_path = _remote_environment_bootstrap_route_path(
resolved_settings.route_path
)
remote_environment_status_route_path = _remote_environment_status_route_path(
resolved_settings.route_path
)
remote_environment_stop_route_path = _remote_environment_stop_route_path(
resolved_settings.route_path
)
remote_host_registry = _ssh_host_registry(resolved_settings)
vast_registry = _vast_registry(resolved_settings)
_install_modal_interrupt_queue_bridge(prompt_server)
@prompt_server.routes.post(_R2_STORAGE_USAGE_ROUTE)
async def refresh_r2_storage_usage(request: web.Request) -> web.Response:
"""Refresh one configured R2 bucket's safe aggregate storage state."""
try:
payload = await request.json()
if not isinstance(payload, Mapping):
raise TypeError("R2 storage usage request must be a JSON object.")
storage = _r2_storage_from_usage_payload(payload)
except (TypeError, ValueError, json.JSONDecodeError) as exc:
return web.json_response({"error": str(exc)}, status=400)
try:
usage = await asyncio.to_thread(_refresh_r2_storage_usage, storage)
except R2CredentialError as exc:
logger.warning(
"Unable to refresh R2 storage usage configuration=%s bucket=%s: %s",
storage.configuration_id,
storage.bucket,
exc,
)
status = 423 if exc.code == R2_KEYCHAIN_UNLOCK_REQUIRED_CODE else 502
return web.json_response(
{"error": str(exc), "code": exc.code},
status=status,
)
except (R2CacheError, RuntimeError, ValueError) as exc:
logger.warning(
"Unable to refresh R2 storage usage configuration=%s bucket=%s: %s",
storage.configuration_id,
storage.bucket,
exc,
)
return web.json_response({"error": str(exc)}, status=502)
return web.json_response(
{
"configuration_id": storage.configuration_id,
"storage_usage_bytes": usage.size_bytes,
"storage_object_count": usage.object_count,
"refreshed_at": time.time(),
}
)
@prompt_server.routes.post(_R2_KEYCHAIN_UNLOCK_ROUTE)
async def unlock_r2_keychain(request: web.Request) -> web.Response:
"""Display macOS's system-owned login-keychain unlock prompt."""
del request
try:
await asyncio.to_thread(request_macos_keychain_unlock)
except R2CredentialError as exc:
logger.warning("Unable to unlock the macOS login keychain: %s", exc)
return web.json_response({"error": str(exc)}, status=409)
return web.json_response({"unlocked": True})
@prompt_server.routes.post(cancel_preparation_route_path)
async def cancel_remote_preparation(request: web.Request) -> web.Response:
"""Cancel one prompt while it is still preparing remote execution."""
payload = await request.json()
prompt_id = str(payload.get("prompt_id") or "").strip()
cancelled = _cancel_remote_preparation(prompt_server, prompt_id)
return web.json_response({"cancelled": cancelled, "prompt_id": prompt_id})
if hasattr(prompt_server.routes, "get"):
@prompt_server.routes.get(remote_environments_route_path)
async def remote_environments(request: web.Request) -> web.Response:
"""Return credential-free SSH host configuration and discovered state."""
del request
if remote_host_registry is None:
return web.json_response(
{
"error": "The ComfyUI user directory could not be resolved.",
"hosts": [],
},
status=503,
)
try:
config = await asyncio.to_thread(remote_host_registry.load)
except ValueError as exc:
return web.json_response({"error": str(exc), "hosts": []}, status=500)
return web.json_response(config.to_dict())
@prompt_server.routes.get(progress_state_route_path)
async def modal_progress_state(request: web.Request) -> web.Response:
"""Return recent Modal UI events for the requesting ComfyUI client."""
client_id = request.query.get("client_id")
return web.json_response({"events": modal_ui_events_for_client(client_id)})
@prompt_server.routes.get(container_status_route_path)
async def modal_container_status(request: web.Request) -> web.Response:
"""Return active containers and hourly billing for one selected GPU app."""
from .remote.modal_app import (
ModalBillingStatusError,
ModalContainerStatusError,
get_hourly_modal_app_billing,
list_active_modal_containers,
)
requested_modal_gpu = request.query.get(
"modal_gpu",
resolved_settings.modal_gpu,
)
try:
selected_settings = settings_for_modal_gpu(
resolved_settings,
requested_modal_gpu,
)
except ValueError as exc:
return web.json_response(
{"containers": [], "error": str(exc), "polled_at": time.time()},
status=400,
)
containers_task = asyncio.create_task(
list_active_modal_containers(resolved_settings)
)
include_billing = request.query.get("include_billing", "true").casefold() not in {
"0",
"false",
"no",
}
billing_task = (
asyncio.create_task(
get_hourly_modal_app_billing(
selected_settings.modal_gpu,
resolved_settings,
)
)
if include_billing
else None
)
try:
containers = await containers_task
except ModalContainerStatusError as exc:
if billing_task is not None:
billing_task.cancel()
await asyncio.gather(billing_task, return_exceptions=True)
logger.warning("Unable to refresh Modal container status: %s", exc)
return web.json_response(
{"containers": [], "error": str(exc), "polled_at": time.time()},
status=502,
)
billing = None
billing_error = None
if billing_task is not None:
try:
billing = await billing_task
except ModalBillingStatusError as exc:
billing_error = str(exc)
logger.warning("Unable to refresh Modal hourly billing: %s", exc)
return web.json_response(
{
"containers": [container.as_dict() for container in containers],
"billing": billing.as_dict() if billing is not None else None,
"billing_error": billing_error,
"polled_at": time.time(),
}
)
@prompt_server.routes.post(container_stop_route_path)
async def modal_container_stop(request: web.Request) -> web.Response:
"""Stop one exact active Modal container owned by this installation."""
from .remote.modal_app import (
ModalContainerStatusError,
stop_managed_modal_container,
)
try:
payload = await request.json()
container_id = str(payload.get("container_id") or "").strip()
stopped = await stop_managed_modal_container(
container_id,
resolved_settings,
)
except (ModalContainerStatusError, TypeError, ValueError) as exc:
return web.json_response({"error": str(exc)}, status=502)
return web.json_response(
{"container_id": container_id, "stopped": stopped}
)
@prompt_server.routes.get("/remote/vast/status")
async def vast_status(request: web.Request) -> web.Response:
"""Return refreshed credential-free managed Vast lease inventory."""
del request
if vast_registry is None:
return web.json_response(
{"configured": False, "leases": [], "error": "ComfyUI user directory unavailable."},
status=503,
)
try:
if not os.getenv("VAST_API_KEY"):
state = await asyncio.to_thread(vast_registry.load)
leases = state.leases
else:
api_key = str(os.getenv("VAST_API_KEY") or "").strip()
base_url = str(
os.getenv("COMFY_MODAL_VAST_API_BASE_URL") or ""
).strip()
api_client = VastApiClient(
api_key,
**({"base_url": base_url} if base_url else {}),
)
manager = VastLeaseManager.for_inventory(
api_client=api_client,
registry=vast_registry,
owner_id=resolved_settings.app_name,
)
leases = await manager.refresh_owned_leases()
except (OSError, RuntimeError, ValueError) as exc:
return web.json_response(
{"configured": bool(os.getenv("VAST_API_KEY")), "leases": [], "error": str(exc)},
status=502,
)
return web.json_response(
{
"configured": bool(os.getenv("VAST_API_KEY")),
"image_configured": bool(os.getenv("COMFY_MODAL_VAST_IMAGE")),
"leases": [lease.to_dict() for lease in leases],
}
)
if hasattr(prompt_server.routes, "put"):
@prompt_server.routes.put(remote_environments_route_path)
async def remote_environments_update(request: web.Request) -> web.Response:
"""Validate and atomically replace SSH host configuration."""
if remote_host_registry is None:
return web.json_response(
{"error": "The ComfyUI user directory could not be resolved."},
status=503,
)
try:
payload = await request.json()
if not isinstance(payload, Mapping):
raise ValueError(
"Remote environment configuration must be a JSON object."
)
config = RemoteExecutionConfig.from_dict(payload)
await asyncio.to_thread(remote_host_registry.save, config)
except (TypeError, ValueError, json.JSONDecodeError) as exc:
return web.json_response({"error": str(exc)}, status=400)
return web.json_response(config.to_dict())
@prompt_server.routes.post(remote_environment_probe_route_path)
async def remote_environment_probe(request: web.Request) -> web.Response:
"""Probe one configured SSH host and persist its discovered capabilities."""
if remote_host_registry is None:
return web.json_response(
{"error": "The ComfyUI user directory could not be resolved."},
status=503,
)
try:
payload = await request.json()
environment_id = str(payload.get("environment_id") or "").strip()
host = await asyncio.to_thread(
remote_host_registry.get_host, environment_id
)
capabilities = await asyncio.to_thread(
SshDockerController(host).probe_capabilities
)
updated_host = await asyncio.to_thread(
remote_host_registry.update_probe_result,
environment_id,
capabilities=capabilities,
health=EnvironmentHealth.READY,
last_error=None,
)
except (KeyError, TypeError, ValueError, RuntimeError) as exc:
environment_id = str(locals().get("environment_id") or "").strip()
if environment_id:
try:
await asyncio.to_thread(
remote_host_registry.update_probe_result,
environment_id,
capabilities=None,
health=EnvironmentHealth.UNAVAILABLE,
last_error=str(exc),
)
except (KeyError, ValueError):
pass
return web.json_response({"error": str(exc)}, status=502)
return web.json_response(updated_host.to_dict())
@prompt_server.routes.post("/remote/vast/verify")
async def vast_verify(request: web.Request) -> web.Response:
"""Verify the configured Vast credential without returning it."""
del request
api_key = str(os.getenv("VAST_API_KEY") or "").strip()
if not api_key:
return web.json_response(
{"verified": False, "error": "Set VAST_API_KEY first."},
status=400,
)
base_url = str(os.getenv("COMFY_MODAL_VAST_API_BASE_URL") or "").strip()
try:
client = VastApiClient(api_key, **({"base_url": base_url} if base_url else {}))
account = await client.verify_credentials()
except (OSError, RuntimeError, ValueError) as exc:
return web.json_response(
{"verified": False, "error": str(exc)},
status=502,
)
return web.json_response({"verified": True, "account": account})
@prompt_server.routes.post("/remote/vast/reap")
async def vast_reap(request: web.Request) -> web.Response:
"""Destroy only owned idle leases whose configured deadline has expired."""
del request
try:
service = VastService.from_environment(
resolved_settings,
repo_root=Path(__file__).resolve().parent,
)
destroyed = await service.lease_manager.destroy_expired()
except (OSError, RuntimeError, ValueError) as exc:
return web.json_response({"error": str(exc)}, status=502)
return web.json_response({"destroyed_instance_ids": list(destroyed)})
@prompt_server.routes.post("/remote/vast/destroy")
async def vast_destroy(request: web.Request) -> web.Response:
"""Destroy one exact idle registry-owned lease after server-side label checks."""
try:
payload = await request.json()
instance_id = int(payload.get("instance_id"))
api_key = str(os.getenv("VAST_API_KEY") or "").strip()
if not api_key:
raise RuntimeError("Set VAST_API_KEY before destroying Vast capacity.")
if vast_registry is None:
raise RuntimeError("ComfyUI user directory is unavailable.")
base_url = str(
os.getenv("COMFY_MODAL_VAST_API_BASE_URL") or ""
).strip()
api_client = VastApiClient(
api_key,
**({"base_url": base_url} if base_url else {}),
)
manager = VastLeaseManager.for_inventory(
api_client=api_client,
registry=vast_registry,
owner_id=resolved_settings.app_name,
)
destroyed = await manager.destroy_owned_lease(
instance_id,
allow_active_work=payload.get("force") is True,
)
except (OSError, RuntimeError, TypeError, ValueError) as exc:
return web.json_response({"error": str(exc)}, status=502)
return web.json_response({"instance_id": instance_id, "destroyed": destroyed})
@prompt_server.routes.post(remote_environment_bootstrap_route_path)
async def remote_environment_bootstrap(request: web.Request) -> web.Response:
"""Build the current runtime and start one compatible warm SSH worker."""
if remote_host_registry is None:
return web.json_response(
{"error": "The ComfyUI user directory could not be resolved."},
status=503,
)
environment_id = ""
try:
from .ssh_runtime import SshRuntimeManager
payload = await request.json()
environment_id = str(payload.get("environment_id") or "").strip()
worker_index = int(payload.get("worker_index", 0))
host = await asyncio.to_thread(
remote_host_registry.get_host, environment_id
)
manager = SshRuntimeManager(
controller=SshDockerController(host),
repo_root=Path(__file__).resolve().parent,
settings=resolved_settings,
)
spec = await asyncio.to_thread(manager.ensure_worker, worker_index)
except (KeyError, TypeError, ValueError, RuntimeError) as exc:
if remote_host_registry is not None and environment_id:
try:
await asyncio.to_thread(
remote_host_registry.update_probe_result,
environment_id,
capabilities=host.capabilities if "host" in locals() else None,
health=EnvironmentHealth.UNAVAILABLE,
last_error=str(exc),
)
except (KeyError, ValueError):
pass
return web.json_response({"error": str(exc)}, status=502)
await asyncio.to_thread(
remote_host_registry.update_probe_result,
environment_id,
capabilities=host.capabilities,
health=EnvironmentHealth.READY,
last_error=None,
)
return web.json_response(
{
"environment_id": environment_id,
"worker_index": worker_index,
"container_name": spec.container_name,
"image_tag": spec.image_tag,
"runtime_fingerprint": spec.identity.fingerprint,
}
)
@prompt_server.routes.post(remote_environment_status_route_path)
async def remote_environment_status(request: web.Request) -> web.Response:
"""Return managed worker state for one SSH execution environment."""
if remote_host_registry is None:
return web.json_response(
{"error": "The ComfyUI user directory could not be resolved."},
status=503,
)
try:
payload = await request.json()
environment_id = str(payload.get("environment_id") or "").strip()
host = await asyncio.to_thread(
remote_host_registry.get_host, environment_id
)
workers = await asyncio.to_thread(
SshDockerController(host).list_managed_workers
)
except (KeyError, TypeError, ValueError, RuntimeError) as exc:
return web.json_response({"error": str(exc)}, status=502)
return web.json_response(
{
"environment_id": environment_id,
"workers": [worker.to_dict() for worker in workers],
}
)
@prompt_server.routes.post(remote_environment_stop_route_path)
async def remote_environment_stop(request: web.Request) -> web.Response:
"""Stop all node-pack-managed workers on one configured SSH host."""
if remote_host_registry is None:
return web.json_response(
{"error": "The ComfyUI user directory could not be resolved."},
status=503,
)
try:
from .ssh_runtime import SshRuntimeManager
payload = await request.json()
environment_id = str(payload.get("environment_id") or "").strip()
host = await asyncio.to_thread(
remote_host_registry.get_host, environment_id
)
manager = SshRuntimeManager(
controller=SshDockerController(host),
repo_root=Path(__file__).resolve().parent,
settings=resolved_settings,
)
removed = await asyncio.to_thread(manager.stop_all_workers)
except (KeyError, TypeError, ValueError, RuntimeError) as exc:
return web.json_response({"error": str(exc)}, status=502)
return web.json_response(
{"environment_id": environment_id, "removed_containers": list(removed)}
)
@prompt_server.routes.post(analysis_route_path)
async def modal_analyze_remote_nodes(request: web.Request) -> web.Response:
"""Analyze which workflow nodes should be marked remote for the current graph."""
logger.info("Received Modal remote-node analysis request.")
try:
request_started_at = time.perf_counter()
json_data = await request.json()
prompt = json_data.get("prompt")
if not isinstance(prompt, dict):
raise ValueError(
"Modal remote-node analysis requires a 'prompt' object."
)
workflow = json_data.get("workflow")
seed_node_ids = json_data.get("seed_node_ids") or []
if not isinstance(seed_node_ids, list):
raise ValueError(
"Modal remote-node analysis requires 'seed_node_ids' to be a list."
)
analysis = analyze_remote_node_selection(
prompt=prompt,
workflow=workflow if isinstance(workflow, dict) else None,
seed_workflow_node_paths=[
str(seed_node_id) for seed_node_id in seed_node_ids
],
settings=resolved_settings,
)
logger.info(
"Modal remote-node analysis finished in %.3fs with %d requested nodes and %d additions.",
time.perf_counter() - request_started_at,
len(analysis.requested_workflow_node_paths),
len(analysis.added_workflow_node_paths),
)
return web.json_response(
{
"requested_node_ids": analysis.requested_node_ids,
"requested_workflow_node_paths": analysis.requested_workflow_node_paths,
"current_remote_node_ids": analysis.current_remote_node_ids,
"current_remote_workflow_node_paths": (
analysis.current_remote_workflow_node_paths
),
"resolved_remote_node_ids": analysis.resolved_remote_node_ids,
"resolved_workflow_node_paths": analysis.resolved_workflow_node_paths,
"added_node_ids": analysis.added_node_ids,
"added_workflow_node_paths": analysis.added_workflow_node_paths,
"sandwiched_local_node_ids": analysis.sandwiched_local_node_ids,
"reasons": [
{
"node_id": reason.node_id,
"class_type": reason.class_type,
"required_by_node_id": reason.required_by_node_id,
"required_by_class_type": reason.required_by_class_type,
"output_index": reason.output_index,
"io_type": reason.io_type,
}
for reason in analysis.reasons
],
}
)
except (TypeError, ValueError) as exc:
logger.warning("Modal remote-node analysis request was invalid: %s", exc)
return web.json_response({"error": str(exc), "node_errors": []}, status=400)
@prompt_server.routes.post(delete_caches_route_path)
async def modal_delete_caches(request: web.Request) -> web.Response:
"""Delete persistent Modal cache Dicts for the active configuration."""
del request
logger.info("Received Modal cache deletion request.")
try:
return web.json_response(await delete_modal_cache_dicts(resolved_settings))
except RuntimeError as exc:
logger.warning("Modal cache deletion request failed: %s", exc)
return web.json_response({"error": str(exc), "node_errors": []}, status=400)
@prompt_server.routes.post(delete_volume_route_path)
async def modal_delete_volume(request: web.Request) -> web.Response:
"""Delete the configured Modal Volume for the active configuration."""
del request
logger.info("Received Modal volume deletion request.")
try:
return web.json_response(await delete_modal_volume(resolved_settings))
except RuntimeError as exc:
logger.warning("Modal volume deletion request failed: %s", exc)
return web.json_response({"error": str(exc), "node_errors": []}, status=400)
@prompt_server.routes.post(resolved_settings.route_path)
async def modal_queue_prompt(request: web.Request) -> web.Response:
"""Handle prompt queue requests that include Modal remote markers."""
logger.info("Received Modal queue request.")
json_data: dict[str, Any] | None = None
workflow: dict[str, Any] | None = None
remote_node_ids: list[str] = []
request_modal_gpu: str | None = None
summary = RewriteSummary()
preparation_prompt_id: str | None = None
preparation_cancellation = threading.Event()
configurator_node_id: str | None = None
r2_writeback_prompt_id: str | None = None
prompt_queued = False
try:
request_started_at = time.perf_counter()
json_data = await request.json()
json_data.setdefault("prompt_id", str(uuid.uuid4()))
json_data.setdefault("extra_data", {})
json_data["extra_data"]["prompt_id"] = json_data["prompt_id"]
if json_data.get("client_id") is not None:
json_data["extra_data"]["client_id"] = json_data["client_id"]
client_id = (
str(json_data.get("client_id")) if json_data.get("client_id") else None
)
prompt_id = (
str(json_data.get("prompt_id")) if json_data.get("prompt_id") else None
)
extra_pnginfo = (json_data.get("extra_data") or {}).get(
"extra_pnginfo"
) or {}
workflow = extra_pnginfo.get("workflow")
remote_node_ids = sorted(
requested_remote_node_ids(
prompt=json_data.get("prompt", {}),
workflow=workflow,
settings=resolved_settings,
)
)
configurator_node_id = _remote_execution_configurator_node_id(
json_data.get("prompt", {})
)
if "prompt" in json_data and not remote_node_ids:
logger.info(
"No workflow nodes are marked for Modal execution; forwarding prompt without Modal status or rewrite."
)
response = await _queue_prompt_json(prompt_server, json_data)
logger.info(
"Modal queue request completed in %.3fs.",
time.perf_counter() - request_started_at,
)
return response
if prompt_id is not None:
begin_r2_writeback_prompt(prompt_id)
r2_writeback_prompt_id = prompt_id
if prompt_id is not None:
extra_pnginfo[MODAL_PROMPT_ID_EXTRA_PNGINFO_KEY] = prompt_id
json_data["extra_data"]["extra_pnginfo"] = extra_pnginfo
logger.debug(
"Attached prompt-scoped Modal execution metadata prompt_id=%s.",
prompt_id,
)
if _set_remote_preparation(
prompt_server,
prompt_id=prompt_id,
prompt=json_data.get("prompt", {}),
extra_data=json_data.get("extra_data", {}),
cancellation_event=preparation_cancellation,
):
preparation_prompt_id = prompt_id
try:
request_settings = settings_for_modal_gpu(
resolved_settings,
modal_gpu_from_workflow(workflow, resolved_settings.modal_gpu),
)
except ValueError as exc:
raise ModalPromptValidationError(str(exc)) from exc
logger.info(
"Resolved workflow Modal GPU selection gpu=%s prompt_id=%s.",
request_settings.modal_gpu,
prompt_id,
)
status_modal_gpu = (
None
if _prompt_uses_remote_execution_configurator(
json_data.get("prompt", {})
)
else request_settings.modal_gpu
)
request_modal_gpu = status_modal_gpu
def emit_setup_status(
message: str,
current: int | None = None,
total: int | None = None,
) -> None:
"""Forward one queue-time Modal setup update into the websocket stream."""
if preparation_cancellation.is_set():
raise SyncCancelledError(
"Remote workflow preparation was cancelled."
)
_emit_modal_status(
prompt_server=prompt_server,
phase="setup",
client_id=client_id,
prompt_id=prompt_id,
node_ids=remote_node_ids,
configurator_node_id=configurator_node_id,
modal_gpu=status_modal_gpu,
component_node_ids_by_representative=(
summary.component_node_ids_by_representative or None
),
status_message=message,
status_current=current,
status_total=total,
)
def emit_execution_plan(
assignments: dict[str, dict[str, Any]],
configurations: list[dict[str, Any]],
) -> None:
"""Publish scheduler choices before capacity acquisition can block."""
component_nodes = {
component_id: list(assignment.get("node_ids", []))
for component_id, assignment in assignments.items()
}
_emit_modal_status(
prompt_server=prompt_server,
phase="setup",
client_id=client_id,
prompt_id=prompt_id,
node_ids=remote_node_ids,
configurator_node_id=configurator_node_id,
modal_gpu=status_modal_gpu,
component_node_ids_by_representative=component_nodes,
status_message="Remote execution plan ready",
remote_execution_assignments=assignments,
remote_execution_configurations=configurations,
)
def emit_environment_setup_status(
environment_id: str,
message: str,
current: int | None = None,
total: int | None = None,
) -> None:
"""Publish setup progress for one concrete remote environment."""
if preparation_cancellation.is_set():
raise SyncCancelledError(
"Remote workflow preparation was cancelled."
)
_emit_modal_status(
prompt_server=prompt_server,
phase="setup",
client_id=client_id,
prompt_id=prompt_id,
node_ids=remote_node_ids,
configurator_node_id=configurator_node_id,
modal_gpu=status_modal_gpu,
component_node_ids_by_representative=(
summary.component_node_ids_by_representative or None
),
status_message=message,
status_current=current,
status_total=total,
execution_environment_id=environment_id,
)
if "prompt" in json_data:
emit_setup_status("Preparing remote workflow")
rewrite_started_at = time.perf_counter()
occupied_environment_ids = _queued_ssh_environment_ids(
prompt_server,
excluding_prompt_id=prompt_id,
)
if occupied_environment_ids:
logger.info(
"Queued workflow may reuse SSH capacity after earlier "
"prompts finish environments=%s.",
sorted(occupied_environment_ids),
)
rewritten_prompt, summary = await rewrite_prompt_for_modal_async(
prompt=json_data["prompt"],
workflow=workflow,
sync_engine=resolved_sync_engine,
settings=request_settings,
extra_data=json_data.get("extra_data"),
status_callback=emit_setup_status,
environment_status_callback=emit_environment_setup_status,
plan_callback=emit_execution_plan,
cancellation_check=preparation_cancellation.is_set,
occupied_environment_ids=occupied_environment_ids,
)
selected_modal_gpus = _selected_modal_gpus(
summary,
request_settings.modal_gpu,
)
status_modal_gpu = (
selected_modal_gpus[0]
if len(selected_modal_gpus) == 1
else None
)
request_modal_gpu = status_modal_gpu
logger.info(
"Modal prompt rewrite finished in %.3fs for %d remote nodes across %d components.",
time.perf_counter() - rewrite_started_at,
len(summary.remote_node_ids),
len(summary.remote_component_ids),
)
remote_node_ids = list(summary.remote_node_ids)
json_data["prompt"] = rewritten_prompt
if json_data.get("partial_execution_targets"):
rewritten_targets = {
summary.rewritten_node_id_map.get(str(target), str(target))
for target in json_data["partial_execution_targets"]
}
json_data["partial_execution_targets"] = sorted(rewritten_targets)
json_data.setdefault("extra_data", {}).setdefault("modal", {})
json_data["extra_data"]["modal"]["gpu"] = request_settings.modal_gpu
json_data["extra_data"]["modal"][
"remote_node_ids"
] = summary.remote_node_ids
json_data["extra_data"]["modal"][
"remote_component_ids"
] = summary.remote_component_ids
json_data["extra_data"]["modal"][
"component_dependency_ids_by_representative"
] = summary.component_dependency_ids_by_representative
json_data["extra_data"]["modal"][
"component_execution_stages"
] = summary.component_execution_stages
json_data["extra_data"]["modal"][
"mapped_component_ids"
] = summary.mapped_component_ids
json_data["extra_data"]["modal"][
"estimated_max_parallel_requests"
] = summary.estimated_max_parallel_requests
json_data["extra_data"]["modal"][
"max_parallel_requests_upper_bound"
] = summary.max_parallel_requests_upper_bound
json_data["extra_data"]["modal"]["synced_assets"] = [
asset.remote_path for asset in summary.synced_assets
]
json_data["extra_data"]["remote_execution"] = {
"assignments": _execution_assignments_payload(
summary,
request_settings,
),
"configurations": list(summary.remote_configurations),
}
if summary.custom_nodes_bundle is not None:
json_data["extra_data"]["modal"][
"custom_nodes_bundle"
] = summary.custom_nodes_bundle.remote_path
_emit_modal_status(
prompt_server=prompt_server,
phase="setup",
client_id=client_id,
prompt_id=prompt_id,
node_ids=remote_node_ids,
configurator_node_id=configurator_node_id,
modal_gpu=status_modal_gpu,
component_node_ids_by_representative=summary.component_node_ids_by_representative,
status_message="Submitting remote workflow",
)
response = await _queue_prompt_json(
prompt_server,
json_data,
modal_response_payload=(
{
"modal_gpu": status_modal_gpu,
"remote_execution_configurator_node_id": configurator_node_id,
"remote_execution_modal_gpus": selected_modal_gpus,
"modal_remote_node_ids": list(summary.remote_node_ids),
"modal_sandwiched_local_node_ids": list(
summary.sandwiched_local_node_ids
),
"modal_parallel_local_branch_node_ids": list(
summary.parallel_local_branch_node_ids
),
"remote_execution_assignments": _execution_assignments_payload(
summary,
request_settings,
),
"remote_execution_configurations": list(
summary.remote_configurations
),
"modal_components": [
{
"representative_node_id": representative_node_id,
"node_ids": list(component_node_ids),
}
for representative_node_id, component_node_ids in sorted(
summary.component_node_ids_by_representative.items()
)
],
}
if summary.remote_node_ids
else None
),
)
prompt_queued = response.status < 400
logger.info(
"Modal queue request completed in %.3fs.",
time.perf_counter() - request_started_at,
)
return response
except SyncCancelledError as exc:
logger.info("Remote workflow preparation cancelled: %s", exc)
if json_data is not None:
_emit_modal_status(
prompt_server=prompt_server,
phase="execution_interrupted",
client_id=str(json_data.get("client_id"))
if json_data.get("client_id")
else None,
prompt_id=str(json_data.get("prompt_id"))
if json_data.get("prompt_id")
else None,
node_ids=remote_node_ids,
configurator_node_id=configurator_node_id,
modal_gpu=request_modal_gpu,
status_message=str(exc),
)
return web.json_response(
{"error": str(exc), "node_errors": [], "cancelled": True},
status=409,
)
except FileNotFoundError as exc:
logger.exception("Modal asset sync failed.")
if json_data is not None:
_emit_modal_status(
prompt_server=prompt_server,
phase="error",
client_id=str(json_data.get("client_id"))
if json_data.get("client_id")
else None,
prompt_id=str(json_data.get("prompt_id"))
if json_data.get("prompt_id")
else None,
node_ids=remote_node_ids,
configurator_node_id=configurator_node_id,
modal_gpu=request_modal_gpu,
error_message=str(exc),
)
return web.json_response({"error": str(exc), "node_errors": []}, status=400)
except ModalPromptValidationError as exc:
logger.exception("Modal prompt validation failed.")
if json_data is not None:
_emit_modal_status(
prompt_server=prompt_server,
phase="error",
client_id=str(json_data.get("client_id"))
if json_data.get("client_id")
else None,
prompt_id=str(json_data.get("prompt_id"))
if json_data.get("prompt_id")
else None,
node_ids=remote_node_ids,
configurator_node_id=configurator_node_id,
modal_gpu=request_modal_gpu,
error_message=str(exc),
)
return web.json_response({"error": str(exc), "node_errors": []}, status=400)
except Exception as exc:
logger.exception("Modal queue handler failed.")
if json_data is not None:
_emit_modal_status(
prompt_server=prompt_server,
phase="error",
client_id=str(json_data.get("client_id"))
if json_data.get("client_id")
else None,
prompt_id=str(json_data.get("prompt_id"))
if json_data.get("prompt_id")
else None,
node_ids=remote_node_ids,
configurator_node_id=configurator_node_id,
modal_gpu=request_modal_gpu,
error_message=str(exc),
)
return web.json_response({"error": str(exc), "node_errors": []}, status=500)
finally:
if preparation_prompt_id is not None:
_clear_remote_preparation(prompt_server, preparation_prompt_id)
if r2_writeback_prompt_id is not None and not prompt_queued:
finish_r2_writeback_prompt(r2_writeback_prompt_id)
_ROUTE_REGISTERED = True
logger.info(
"Registered Modal queue route at %s, analysis route at %s, progress state route at %s, container status route at %s, cache deletion route at %s, and volume deletion route at %s",
resolved_settings.route_path,
analysis_route_path,
progress_state_route_path,
container_status_route_path,
delete_caches_route_path,
delete_volume_route_path,
)