ComfyUI/tools/magic_patch/assets/comfy-api.pyi

"""The published ComfyUI V2 Python API — the complete conversion surface.

Generated from the live public ComfyUI API modules.
If a member is not here it does not exist: do not call it, and report
the gap naming what is missing. Reached from a converted pack as:

    from comfy_api.latest import io, sdk
"""
from typing import Any


# ─── comfy_api.latest.io ──────────────────────────────────────────────────────────

class Accumulation(ComfyTypeIO):
    io_type: Any = ...

class AnyType(ComfyTypeIO):
    io_type: Any = ...

class Array(ComfyTypeIO):
    io_type: Any = ...

class Audio(ComfyTypeIO):
    io_type: Any = ...

class AudioEncoder(ComfyTypeIO):
    io_type: Any = ...

class AudioEncoderOutput(ComfyTypeIO):
    io_type: Any = ...

class Autogrow(ComfyTypeI):
    io_type: Any = ...

class BBOX(ComfyTypeIO):
    io_type: Any = ...

class BackgroundRemoval(ComfyTypeIO):
    io_type: Any = ...

class Boolean(ComfyTypeIO):
    io_type: Any = ...

class BoundingBox(ComfyTypeIO):
    io_type: Any = ...

class BoundingBoxes(ComfyTypeIO):
    io_type: Any = ...

class Clip(ComfyTypeIO):
    io_type: Any = ...

class ClipVision(ComfyTypeIO):
    io_type: Any = ...

class ClipVisionOutput(ComfyTypeIO):
    io_type: Any = ...

class Color(ComfyTypeIO):
    io_type: Any = ...

class Colors(ComfyTypeIO):
    io_type: Any = ...

class Combo(ComfyTypeIO):
    io_type: Any = ...

class ComfyNode(_ComfyNodeBaseInternal):
    """Common base class for all V3 nodes."""
    @classmethod
    # DO NOT override this class. Will break things in execution.py.
    def GET_BASE_CLASS(cls): ...
    @classmethod
    # Optionally, define this function to return a list of input names that should be evaluated.
    def check_lazy_status(cls, **kwargs) -> list[str]: ...
    @classmethod
    # Override this function with one that returns a Schema instance.
    def define_schema(cls) -> Schema: ...
    @classmethod
    # Override this function with one that performs node's actions.
    def execute(cls, **kwargs) -> NodeOutput: ...
    @classmethod
    # Optionally, define this function to fingerprint inputs; equivalent to V1's IS_CHANGED.
    def fingerprint_inputs(cls, **kwargs) -> Any: ...
    @classmethod
    # Optionally, define this function to validate inputs; equivalent to V1's VALIDATE_INPUTS.
    def validate_inputs(cls, **kwargs) -> bool | str: ...

class ComfyTypeI(_ComfyType):
    """ComfyType subclass that only has a default Input class - intended for types that only have Inputs."""
    ...

class ComfyTypeIO(ComfyTypeI):
    """ComfyType subclass that has default Input and Output classes; useful for types with both Inputs and Outputs."""
    ...

class Conditioning(ComfyTypeIO):
    io_type: Any = ...

class ControlAfterGenerate(str, Enum):
    decrement: Any = ...
    fixed: Any = ...
    increment: Any = ...
    randomize: Any = ...

class ControlNet(ComfyTypeIO):
    io_type: Any = ...

class Curve(ComfyTypeIO):
    io_type: Any = ...

# Create a ComfyType for a custom io_type.
def Custom(io_type: str) -> type[ComfyTypeIO]: ...

class Dict(ComfyTypeIO):
    io_type: Any = ...

class DynamicCombo(ComfyTypeI):
    io_type: Any = ...

class FaceAnalysis(ComfyTypeIO):
    io_type: Any = ...

class File3DAny(ComfyTypeIO):
    """General 3D file type - accepts any supported 3D format."""
    io_type: Any = ...

class File3DFBX(ComfyTypeIO):
    """FBX format 3D file - best for game engines and animation."""
    io_type: Any = ...

class File3DGLB(ComfyTypeIO):
    """GLB format 3D file - binary glTF, best for web and cross-platform."""
    io_type: Any = ...

class File3DGLTF(ComfyTypeIO):
    """GLTF format 3D file - JSON-based glTF with external resources."""
    io_type: Any = ...

class File3DKSPLAT(ComfyTypeIO):
    """KSPLAT format 3D file - 3D Gaussian splat."""
    io_type: Any = ...

class File3DOBJ(ComfyTypeIO):
    """OBJ format 3D file - simple geometry format."""
    io_type: Any = ...

class File3DPLY(ComfyTypeIO):
    """PLY format 3D file - point cloud or Gaussian splat."""
    io_type: Any = ...

class File3DPointCloudAny(ComfyTypeIO):
    """General point cloud file type - accepts any supported point cloud container (currently .ply)."""
    io_type: Any = ...

class File3DSPLAT(ComfyTypeIO):
    """SPLAT format 3D file - 3D Gaussian splat."""
    io_type: Any = ...

class File3DSPZ(ComfyTypeIO):
    """SPZ format 3D file - compressed 3D Gaussian splat."""
    io_type: Any = ...

class File3DSTL(ComfyTypeIO):
    """STL format 3D file - best for 3D printing."""
    io_type: Any = ...

class File3DSplatAny(ComfyTypeIO):
    """General 3D Gaussian splat file type - accepts any supported splat container (.ply / .spz / .splat / .ksplat)."""
    io_type: Any = ...

class File3DUSDZ(ComfyTypeIO):
    """USDZ format 3D file - Apple AR format."""
    io_type: Any = ...

class Float(ComfyTypeIO):
    io_type: Any = ...

class FlowControl(ComfyTypeIO):
    io_type: Any = ...

class FolderType(str, Enum):
    input: Any = ...
    output: Any = ...
    temp: Any = ...

class Gligen(ComfyTypeIO):
    """ModelPatcher that wraps around a 'Gligen' model."""
    io_type: Any = ...

class Guider(ComfyTypeIO):
    io_type: Any = ...

class Hidden(str, Enum):
    """Enumerator for requesting hidden variables in nodes."""
    api_key_comfy_org: Any = ...
    auth_token_comfy_org: Any = ...
    comfy_usage_source: Any = ...
    dynprompt: Any = ...
    extra_pnginfo: Any = ...
    prompt: Any = ...
    unique_id: Any = ...

class HiddenHolder:
    def __init__(self, unique_id: str, prompt: Any, extra_pnginfo: Any, dynprompt: Any, auth_token_comfy_org: str, api_key_comfy_org: str, comfy_usage_source: str = None, **kwargs): ...
    @classmethod
    def from_dict(cls, d: dict | None): ...
    @classmethod
    def from_v3_data(cls, v3_data: V3Data | None) -> HiddenHolder: ...

class Histogram(ComfyTypeIO):
    """A histogram represented as a list of bin counts."""
    io_type: Any = ...

class HookKeyframes(ComfyTypeIO):
    io_type: Any = ...

class Hooks(ComfyTypeIO):
    io_type: Any = ...

class Image(ComfyTypeIO):
    io_type: Any = ...

class ImageCompare(ComfyTypeI):
    io_type: Any = ...

class Input(_IO_V3):
    """Base class for a V3 Input."""
    def __init__(self, id: str, display_name: str = None, optional=False, tooltip: str = None, lazy: bool = None, extra_dict=None, raw_link: bool = None, advanced: bool = None): ...
    def as_dict(self): ...
    def get_all(self) -> list[Input]: ...
    def get_io_type(self): ...

class Int(ComfyTypeIO):
    io_type: Any = ...

class Latent(ComfyTypeIO):
    """Latents are stored as a dictionary."""
    io_type: Any = ...

class LatentOperation(ComfyTypeIO):
    io_type: Any = ...

class LatentUpscaleModel(ComfyTypeIO):
    io_type: Any = ...

class Load3D(ComfyTypeIO):
    """3D models are stored as a dictionary."""
    io_type: Any = ...

class Load3DAnimation(Load3D):
    io_type: Any = ...

class Load3DCamera(ComfyTypeIO):
    io_type: Any = ...

class Load3DModelInfo(ComfyTypeIO):
    io_type: Any = ...

class LoraModel(ComfyTypeIO):
    io_type: Any = ...

class LossMap(ComfyTypeIO):
    io_type: Any = ...

MAX_RESOLUTION: Any = ...  # = 16384

class Mask(ComfyTypeIO):
    io_type: Any = ...

class MatchType(ComfyTypeIO):
    io_type: Any = ...

class Mesh(ComfyTypeIO):
    io_type: Any = ...

class Model(ComfyTypeIO):
    io_type: Any = ...

class ModelPatch(ComfyTypeIO):
    io_type: Any = ...

class MultiCombo(ComfyTypeI):
    """Multiselect Combo input (dropdown for selecting potentially more than one value)."""
    io_type: Any = ...

class MultiType(MultiType, ComfyTypeIO):
    io_type: Any = ...

class NodeInfoV1:
    """NodeInfoV1(input: 'dict' = None, input_order: 'dict[str, list[str]]' = None, is_input_list: 'bool' = None, output: 'list[str]' = None, output_is_list: 'list[bool]' = None, output_name: 'list[str]' = None, output_tooltips: 'list[str]' = None, output_matchtypes: 'list[str]' = None, name: 'str' = None, display_name: 'str' = None, description: 'str' = None, python_module: 'Any' = None, category: 'str' = None, output_node: 'bool' = None, deprecated: 'bool' = None, experimental: 'bool' = None, dev_only: 'bool' = None, api_node: 'bool' = None, price_badge: 'dict | None' = None, search_aliases: 'list[str]' = None, essentials_category: 'str' = None, has_intermediate_output: 'bool' = None)"""
    def __init__(self, input: dict = None, input_order: dict[str, list[str]] = None, is_input_list: bool = None, output: list[str] = None, output_is_list: list[bool] = None, output_name: list[str] = None, output_tooltips: list[str] = None, output_matchtypes: list[str] = None, name: str = None, display_name: str = None, description: str = None, python_module: Any = None, category: str = None, output_node: bool = None, deprecated: bool = None, experimental: bool = None, dev_only: bool = None, api_node: bool = None, price_badge: dict | None = None, search_aliases: list[str] = None, essentials_category: str = None, has_intermediate_output: bool = None) -> None: ...
    api_node: Any = ...
    category: Any = ...
    deprecated: Any = ...
    description: Any = ...
    dev_only: Any = ...
    display_name: Any = ...
    essentials_category: Any = ...
    experimental: Any = ...
    has_intermediate_output: Any = ...
    input: Any = ...
    input_order: Any = ...
    is_input_list: Any = ...
    name: Any = ...
    output: Any = ...
    output_is_list: Any = ...
    output_matchtypes: Any = ...
    output_name: Any = ...
    output_node: Any = ...
    output_tooltips: Any = ...
    price_badge: Any = ...
    python_module: Any = ...
    search_aliases: Any = ...

class NodeOutput(_NodeOutputInternal):
    """Standardized output of a node; can pass in any number of args and/or a UIOutput into 'ui' kwarg."""
    def __init__(self, *args: Any, ui: _UIOutput | dict = None, expand: dict = None, block_execution: str = None): ...
    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> NodeOutput: ...
    @property
    def result(self) -> Any: ...

class NodeReplace:
    """Defines a possible node replacement, mapping inputs and outputs of the old node to the new node."""
    def __init__(self, new_node_id: str, old_node_id: str, old_widget_ids: list[str] | None = None, input_mapping: list[InputMap] | None = None, output_mapping: list[OutputMap] | None = None): ...
    # Create serializable representation of the node replacement.
    def as_dict(self): ...

class Noise(ComfyTypeIO):
    io_type: Any = ...

class NumberDisplay(str, Enum):
    gradient_slider: Any = ...
    number: Any = ...
    slider: Any = ...

class Output(_IO_V3):
    def __init__(self, id: str = None, display_name: str = None, tooltip: str = None, is_output_list=False): ...
    def as_dict(self): ...
    def get_io_type(self): ...

class Photomaker(ComfyTypeIO):
    io_type: Any = ...

class Point(ComfyTypeIO):
    io_type: Any = ...

class PriceBadge:
    """PriceBadge(expr: 'str', depends_on: 'PriceBadgeDepends' = <factory>, engine: 'str' = 'jsonata')"""
    def __init__(self, expr: str, depends_on: PriceBadgeDepends = ..., engine: str = 'jsonata') -> None: ...
    def as_dict(self, schema_inputs: list['Input']) -> dict[str, Any]: ...
    engine: Any = ...
    def validate(self) -> None: ...

class PriceBadgeDepends:
    """PriceBadgeDepends(widgets: 'list[str]' = <factory>, inputs: 'list[str]' = <factory>, input_groups: 'list[str]' = <factory>)"""
    def __init__(self, widgets: list[str] = ..., inputs: list[str] = ..., input_groups: list[str] = ...) -> None: ...
    def as_dict(self, schema_inputs: list['Input']) -> dict[str, Any]: ...
    def validate(self) -> None: ...

class Range(ComfyTypeIO):
    io_type: Any = ...

class RemoteOptions:
    def __init__(self, route: str, refresh_button: bool, control_after_refresh: Literal['first', 'last'] = 'first', timeout: int = None, max_retries: int = None, refresh: int = None): ...
    def as_dict(self): ...

class SEGS(ComfyTypeIO):
    io_type: Any = ...

class SVG(ComfyTypeIO):
    io_type: Any = ...

class Sampler(ComfyTypeIO):
    io_type: Any = ...

class Schema:
    """Definition of V3 node properties."""
    def __init__(self, node_id: str, display_name: str = None, category: str = 'sd', inputs: list[Input] = ..., outputs: list[Output] = ..., hidden: list[Hidden] = ..., description: str = '', search_aliases: list[str] = ..., is_input_list: bool = False, is_output_node: bool = False, is_deprecated: bool = False, is_experimental: bool = False, is_dev_only: bool = False, is_api_node: bool = False, price_badge: PriceBadge | None = None, not_idempotent: bool = False, enable_expand: bool = False, accept_all_inputs: bool = False, essentials_category: str | None = None, has_intermediate_output: bool = False) -> None: ...
    accept_all_inputs: Any = ...
    category: Any = ...
    description: Any = ...
    display_name: Any = ...
    enable_expand: Any = ...
    essentials_category: Any = ...
    # Add hidden based on selected schema options, and give outputs without ids default ids.
    def finalize(self): ...
    def get_v1_info(self, cls) -> NodeInfoV1: ...
    has_intermediate_output: Any = ...
    is_api_node: Any = ...
    is_deprecated: Any = ...
    is_dev_only: Any = ...
    is_experimental: Any = ...
    is_input_list: Any = ...
    is_output_node: Any = ...
    not_idempotent: Any = ...
    price_badge: Any = ...
    # Validate the schema:
    def validate(self): ...

class Sigmas(ComfyTypeIO):
    io_type: Any = ...

class Splat(ComfyTypeIO):
    io_type: Any = ...

class String(ComfyTypeIO):
    io_type: Any = ...

class StyleModel(ComfyTypeIO):
    io_type: Any = ...

class TimestepsRange(ComfyTypeIO):
    """Range defined by start and endpoint, between 0.0 and 1.0."""
    io_type: Any = ...

class Tracks(ComfyTypeIO):
    io_type: Any = ...

class UploadType(str, Enum):
    audio: Any = ...
    image: Any = ...
    model: Any = ...
    video: Any = ...

class UpscaleModel(ComfyTypeIO):
    io_type: Any = ...

class V3Data(dict):
    ...

class Vae(ComfyTypeIO):
    io_type: Any = ...

class Video(ComfyTypeIO):
    io_type: Any = ...

class Voxel(ComfyTypeIO):
    io_type: Any = ...

class WanCameraEmbedding(ComfyTypeIO):
    io_type: Any = ...

class Webcam(ComfyTypeIO):
    io_type: Any = ...

class WidgetInput(Input):
    """Base class for a V3 Input with widget."""
    def __init__(self, id: str, display_name: str = None, optional=False, tooltip: str = None, lazy: bool = None, default: Any = None, socketless: bool = None, widget_type: str = None, force_input: bool = None, extra_dict=None, raw_link: bool = None, advanced: bool = None): ...
    def as_dict(self): ...
    def get_io_type(self): ...

def add_to_dict_v1(i: Input, d: dict): ...

# Decorator to mark nested classes as ComfyType; io_type will be bound to the class.
def comfytype(io_type: str, **kwargs): ...

# ─── comfy_api.latest.sdk ─────────────────────────────────────────────────────────

class Ref:
    """Base opaque resource handle. ``kind`` is the io_type (IMAGE, LATENT...)."""
    def __init__(self, kind: str, id: str) -> None: ...
    # Return a bounded, inert description of this opaque value.
    async def describe(self, max_value_chars: int = 32768) -> dict[str, Any]: ...
    # Optional early free. **A node never has to call this.**
    async def release(self) -> None: ...

class ClosureRef(_TypedRef):
    """Handle to a retained node closure (D21)."""
    KIND: Any = ...
    # Expose a ``latent_operation`` closure as LATENT_OPERATION.
    async def as_latent_operation(self) -> 'LatentOperationRef': ...
    # Expose a ``custom_sampler`` closure as a host-owned SAMPLER.
    async def as_sampler(self) -> 'SamplerRef': ...
    # Attach this closure to its declared canonical model phase.
    async def attach_model(self, model: 'ModelRef') -> 'ModelRef': ...
    # Wrap a sampler's model calls with a ``model_sigma`` closure.
    async def wrap_sampler(self, sampler: 'SamplerRef', *, start_percent: Optional[float] = None, end_percent: Optional[float] = None) -> 'SamplerRef': ...

class ClosuresDomain:
    def __init__(self, *args, **kwargs): ...
    async def attach_model(self, closure: ClosureRef, model: ModelRef) -> ModelRef: ...
    async def attach_sampler(self, closure: ClosureRef, sampler: SamplerRef, *, start_percent: Optional[float] = None, end_percent: Optional[float] = None) -> SamplerRef: ...
    async def create_latent_operation(self, closure: ClosureRef) -> LatentOperationRef: ...
    async def create_sampler(self, closure: ClosureRef) -> SamplerRef: ...
    async def retain(self, kind: str, fn: Callable, *, captures: Optional[dict] = None) -> ClosureRef: ...

class AnimaDomain:
    """Vendor-specific Anima model adapters."""
    def __init__(self, *args, **kwargs): ...
    async def apply_lllite(self, model: ModelRef, weights: AssetRef, image: ImageRef, *, strength: float = 1.0, start_percent: float = 0.0, end_percent: float = 1.0, preserve_wrapper: bool = True) -> ModelRef: ...

class TensorRef(_TypedRef):
    KIND: Any = ...
    async def raw(self) -> Any: ...

class UpscaleModelRef(_TypedRef):
    KIND: Any = ...
    async def upscale(self, images: ImageRef, per_batch: int = 16, downscale_ratio: float = 1.0, downscale_method: str = 'lanczos', precision: str = 'float32', tile_size: Optional[int] = None, channels_last: bool = False) -> ImageRef: ...

class ValueRef(_TypedRef):
    """A handle whose value is buffer-safe structured data."""
    KIND: Any = ...
    @classmethod
    def from_value(cls, v: Any) -> 'Ref': ...
    # Read a dict/list of tensors and JSON scalars.
    async def value(self) -> Any: ...

class ImageRef(TensorRef):
    KIND: Any = ...
    # Return the number of images without exposing their pixel buffers.
    async def batch_size(self) -> int: ...
    async def invert(self) -> 'ImageRef': ...
    async def op(self, name: str, **params: Any) -> 'ImageRef': ...
    # Return the first three channels of an image as an opaque ref.
    async def rgb(self) -> 'ImageRef': ...
    async def scale(self, factor: float) -> 'ImageRef': ...
    # Select an ordered, bounded set of images from a BHWC batch.
    async def select_batch(self, indices: list[int]) -> 'ImageRef': ...
    # Return image height and width without exposing its pixel buffer.
    async def spatial_shape(self) -> tuple[int, int]: ...
    # Clone this image onto a named ComfyUI-managed device.
    async def to_device(self, device: str = 'auto') -> 'ImageRef': ...

class ImageClassifierRef(_TypedRef):
    KIND: Any = ...
    # Classify a host-side image batch and return bounded label scores.
    async def classify(self, images: ImageRef, use_accelerator: bool = True, top_k: int = 5) -> list[list[dict[str, Any]]]: ...
    # Run a multi-label classifier and retain its score matrix host-side.
    async def predict_scores(self, images: ImageRef) -> 'ClassifierScoresRef': ...

class ClassifierScoresRef(_TypedRef):
    """Opaque bounded batch-by-class scores from an image classifier."""
    KIND: Any = ...
    # Page score/index pairs above a threshold in one class range.
    async def select_above(self, batch_index: int, start: int, end: int, threshold: float, offset: int = 0, limit: int = 512) -> dict[str, Any]: ...
    async def shape(self) -> tuple[int, int]: ...

class InpaintModelRef(_TypedRef):
    """Opaque prompt-free image inpainting model."""
    KIND: Any = ...
    # Fill the masked image region while keeping model weights host-side.
    async def inpaint(self, image: ImageRef, mask: MaskRef) -> ImageRef: ...

class ImagePreprocessorRef(_TypedRef):
    """Opaque host-created image preprocessor with one bounded operation."""
    KIND: Any = ...
    async def apply(self, image: ImageRef, mask: Optional[MaskRef] = None) -> ImageRef: ...

class IpAdapterEmbedsRef(_TypedRef):
    """Opaque image embeddings produced by a host IP-Adapter encoder."""
    KIND: Any = ...
    async def combine(self, others: list['IpAdapterEmbedsRef'], method: str = 'concat') -> 'IpAdapterEmbedsRef': ...

class IpAdapterRef(_TypedRef):
    """Opaque, host-created IP-Adapter pipeline."""
    KIND: Any = ...
    # Apply this pipeline to a model using bounded image inputs.
    async def apply(self, model: ModelRef, image: ImageRef, negative_image: Optional[ImageRef] = None, attn_mask: Optional[MaskRef] = None, style_image: Optional[ImageRef] = None, composition_image: Optional[ImageRef] = None, weight: float = 0.7, weight_type: str = 'channel penalty', start_percent: float = 0.0, end_percent: float = 1.0, combine_embeds: str = 'concat', weight_faceidv2: float = 1.0, embeds_scaling: str = 'V only', unfold_batch: bool = False, layer_weights: Optional[str] = None, weight_style: float = 1.0, weight_composition: float = 1.0, expand_style: bool = False) -> ModelRef: ...
    # Apply already encoded image embeddings to a model.
    async def apply_embeds(self, model: ModelRef, positive: IpAdapterEmbedsRef, negative: Optional[IpAdapterEmbedsRef] = None, attn_mask: Optional[MaskRef] = None, weight: float = 1.0, weight_type: str = 'linear', start_percent: float = 0.0, end_percent: float = 1.0, embeds_scaling: str = 'V only') -> ModelRef: ...
    # Apply the canonical tiled IP-Adapter operation.
    async def apply_tiled(self, model: ModelRef, image: ImageRef, negative_image: Optional[ImageRef] = None, attn_mask: Optional[MaskRef] = None, weight: float = 0.7, weight_type: str = 'linear', start_percent: float = 0.0, end_percent: float = 1.0, combine_embeds: str = 'concat', embeds_scaling: str = 'V only', sharpening: float = 0.0, unfold_batch: bool = False) -> tuple[ModelRef, ImageRef, MaskRef]: ...
    # Encode one image into positive and negative IP-Adapter embeddings.
    async def encode(self, image: ImageRef, weight: float = 1.0, mask: Optional[MaskRef] = None) -> tuple[IpAdapterEmbedsRef, IpAdapterEmbedsRef]: ...

class InteractionDomain:
    def __init__(self, *args, **kwargs): ...
    async def request(self, kind: str, payload: Any, *, reuse_last: bool = False, remember: bool = False, timeout: float = 540.0) -> Any: ...

class InterpolationStatesRef(_TypedRef):
    """A frame-interpolation skip policy projected from another pack."""
    KIND: Any = ...
    # Return which source-frame pairs must not be interpolated.
    async def skip_mask(self, pair_count: int) -> list[bool]: ...

class MaskRef(TensorRef):
    KIND: Any = ...
    # Dilate or erode a mask through core's canonical morphology node.
    async def grow(self, amount: int, tapered_corners: bool = False) -> 'MaskRef': ...

class MattingModelRef(_TypedRef):
    """Opaque fixed-architecture image matting model."""
    KIND: Any = ...
    # Refine a coarse trimap into an alpha mask.
    async def refine(self, image: ImageRef, trimap: MaskRef, max_megapixels: float = 2.0) -> MaskRef: ...

class LatentOperationRef(_TypedRef):
    """Handle to a host-owned LATENT_OPERATION callable."""
    KIND: Any = ...

class LatentRef(ValueRef):
    KIND: Any = ...
    # Composite a source latent using core's bounded latent operation.
    async def composite(self, source: 'LatentRef', *, x: int = 0, y: int = 0, resize_source: bool = False, mask: Optional['MaskRef'] = None) -> 'LatentRef': ...
    @classmethod
    # Create a bounded zero latent without granting raw tensor access.
    def empty(cls, width: int, height: int, batch_size: int = 1, channels: int = 4, spatial_downscale_ratio: Optional[int] = None) -> 'LatentRef': ...
    async def minimax_h3_token_count(self, conditioning: 'CondRef') -> dict[str, Any]: ...
    # Return the latent's optional noise mask as an opaque mask ref.
    async def noise_mask(self) -> Optional['MaskRef']: ...
    # Generate latent-shaped noise with a host-owned CPU/GPU RNG.
    async def random_noise(self, seed: int, source: str = 'cpu', batch_size: Optional[int] = None) -> TensorRef: ...
    # Repeat a latent through core's canonical batch operation.
    async def repeat_batch(self, amount: int) -> 'LatentRef': ...
    # Resize latent spatial cells with ComfyUI's canonical interpolators.
    async def resize(self, width: int, height: int, method: str = 'bilinear') -> 'LatentRef': ...
    # Return the latent sample height and width without exposing buffers.
    async def spatial_shape(self) -> tuple[int, int]: ...

class LlmDomain:
    """Provider-neutral bounded chat and function-tool contract."""
    def __init__(self, *args, **kwargs): ...
    async def chat(self, provider: str, profile: str, model: str, messages: list[dict[str, Any]], *, tools: Optional[list[dict[str, Any]]] = None, temperature: float = 0.8, max_tokens: int = 512, thinking: bool = False, response_format: str | dict[str, Any] = '', timeout_seconds: float = 600.0, vendor_options: Optional[dict[str, Any]] = None) -> dict[str, Any]: ...

class LlamaCppDomain:
    """Bounded llama.cpp vendor adapter over managed GGUF weights."""
    def __init__(self, *args, **kwargs): ...
    async def generate(self, model: LlamaCppModelRef, system: str, prompt: str, image: Optional[ImageRef] = None, video: Optional[ImageRef] = None, max_tokens: int = 512, temperature: float = 0.7, top_p: float = 0.9, repetition_penalty: float = 1.0, seed: int = 1) -> str: ...
    async def load_chat_model(self, model_weight: str, mmproj_weight: Optional[str] = None, *, family: str = 'qwen3_vl', device: str = 'auto', context_length: int = 8192, batch_size: int = 512, gpu_layers: int = -1, image_max_tokens: int = 4096, top_k: int = 0, pool_size: int = 4194304, cache: bool = True) -> LlamaCppModelRef: ...

class LlamaCppModelRef(_TypedRef):
    """Opaque vendor-owned llama.cpp chat/VLM session."""
    KIND: Any = ...
    async def generate(self, system: str, prompt: str, image: Optional[ImageRef] = None, video: Optional[ImageRef] = None, max_tokens: int = 512, temperature: float = 0.7, top_p: float = 0.9, repetition_penalty: float = 1.0, seed: int = 1) -> str: ...

class CondRef(ValueRef):
    KIND: Any = ...
    async def combine(self, other: 'CondRef') -> 'CondRef': ...
    async def concat(self, other: 'CondRef') -> 'CondRef': ...
    # Whether tile-relative conditioning must be cropped per image.
    async def has_spatial_metadata(self) -> bool: ...
    async def sequence_length(self) -> int: ...
    # Crop 2D spatial conditioning to a latent-space window.
    async def spatial_crop(self, *, x: int, y: int, width: int, height: int, source_width: int, source_height: int, target_width: Optional[int] = None, target_height: Optional[int] = None) -> 'CondRef': ...
    # Attach one opaque CLIP-vision result to every conditioning row.
    async def with_clip_vision_output(self, output: 'ClipVisionOutputRef') -> 'CondRef': ...
    # Attach model-formatted latent ``c_concat`` conditioning.
    async def with_concat_latent(self, model: 'ModelRef', latent: 'LatentRef', extra_latent: Optional['LatentRef'] = None) -> 'CondRef': ...
    # Attach core conditioning-mask metadata without exposing tensors.
    async def with_mask(self, mask: MaskRef, strength: float = 1.0, set_area_to_bounds: bool = False) -> 'CondRef': ...
    # Attach closed, scalar micro-conditioning metadata.
    async def with_metadata(self, *, width: Optional[int] = None, height: Optional[int] = None, crop_w: Optional[int] = None, crop_h: Optional[int] = None, target_width: Optional[int] = None, target_height: Optional[int] = None) -> 'CondRef': ...
    # Clone conditioning with a normalized sampling-percent range.
    async def with_timestep_range(self, start: float, end: float) -> 'CondRef': ...
    # Zero embeddings while preserving the conditioning structure.
    async def zero_out(self) -> 'CondRef': ...

class GligenRef(_TypedRef):
    KIND: Any = ...
    async def apply_batched(self, conditioning: 'CondRef', clip: 'ClipRef', text: str, boxes: list[tuple[int, int, int | float, int | float]]) -> 'CondRef': ...

class GuiderRef(_TypedRef):
    """Opaque handle to a host-owned sampling guider."""
    KIND: Any = ...
    # Clone this guider and crop model-owned spatial inputs for tiles.
    async def spatial_crop_inputs(self, *, regions: list[tuple[int, int, int, int]], source_width: int, source_height: int, target_width: int, target_height: int) -> 'GuiderRef': ...

class SamplerRef(_TypedRef):
    """Opaque handle to a host-owned sampler."""
    KIND: Any = ...
    @classmethod
    # Select a core sampler with its small, validated option set.
    def named(cls, name: str, *, eta: Optional[float] = None, ge_gamma: Optional[float] = None) -> 'SamplerRef': ...
    @classmethod
    def self_refine_video(cls, stochastic_steps: list[dict[str, int]], certain_percentage: float, uncertainty_threshold: float, seed: int, verbose: bool = False, latent: Optional['LatentRef'] = None) -> 'SamplerRef': ...

class SigmasRef(_TypedRef):
    """Opaque one-dimensional host-owned sampling schedule."""
    KIND: Any = ...
    # Return the number of sampling intervals in this schedule.
    async def steps(self) -> int: ...
    # Return one finite scalar from a bounded sampling schedule.
    async def value_at(self, index: int) -> float: ...

class SamModelRef(_TypedRef):
    KIND: Any = ...
    # Segment one host-side image from bounded boxes and point hints.
    async def segment(self, image: ImageRef, boxes: list[Optional[list[float]]], point_coords: Optional[list[list[list[float]]]] = None, point_labels: Optional[list[list[int]]] = None, multimask_output: bool = True) -> tuple[MaskRef, list[list[float]]]: ...
    # Propagate frame-zero boxes through a host-side SAM2 video batch.
    async def segment_video(self, frames: ImageRef, boxes: list[list[float]]) -> MaskRef: ...

class SemanticSegmentationRef(_TypedRef):
    """Opaque fixed-architecture semantic segmentation model."""
    KIND: Any = ...
    # Return the union of selected semantic class IDs as a mask.
    async def mask(self, image: ImageRef, classes: list[int]) -> MaskRef: ...

class HuggingFaceWeight:
    """A public Hugging Face weight file required by a node."""
    def __init__(self, repo_id: str, filename: str, folder: str, revision: str = 'main', sha256: Optional[str] = None, on_demand: bool = False) -> None: ...
    @property
    def catalogue_name(self) -> Any: ...
    on_demand: Any = ...
    revision: Any = ...
    sha256: Any = ...

class ModelRef(_TypedRef):
    KIND: Any = ...
    async def apply_dit_block_lora(self, asset: 'AssetRef', strength_model: float, block_weights: list[dict[str, Any]]) -> tuple['ModelRef', str]: ...
    # Apply one resolved LoRA without exposing model weights or paths.
    async def apply_lora(self, asset: 'AssetRef', clip: Optional['ClipRef'], strength_model: float, strength_clip: float) -> tuple['ModelRef', Optional['ClipRef']]: ...
    async def apply_ltx2_lora(self, asset: 'AssetRef', strength_model: float, block_weights: list[dict[str, Any]], video: float, video_to_audio: float, audio: float, audio_to_video: float, other: float) -> tuple['ModelRef', str, str]: ...
    # Return the model's canonical base family, or ``unknown``.
    async def family(self) -> str: ...
    # Text-ground objects in an image with a compatible vision MODEL.
    async def ground_image(self, image: ImageRef, conditioning: CondRef, *, threshold: float = 0.5, refine_iterations: int = 2, individual_masks: bool = True, max_detections: int = 64) -> tuple[MaskRef, list[list[dict[str, float]]]]: ...
    # Whether the model uses ComfyUI's FLOW model family.
    async def is_flow(self) -> bool: ...
    # Whether the model's sampling schedule uses zero terminal SNR.
    async def is_zero_terminal_snr(self) -> bool: ...
    async def latent_scale_factor(self) -> float: ...
    async def lora_weight_differences(self, original: 'ModelRef', include_bias: bool = False) -> 'WeightDiffCursorRef': ...
    # Apply a named transform, returning a NEW model ref.
    async def patch(self, transform: str, **params: Any) -> 'ModelRef': ...
    # Return one bounded scheduler delta in latent-value units.
    async def sampling_sigma_delta(self, *, steps: int, sampler_name: str, scheduler: str, start_step: int, end_step: int, denoise: float = 1.0, sigma_schedule: Optional[dict] = None) -> float: ...
    async def scheduled_cfg_guider(self, positive: 'CondRef', negative: 'CondRef', cfg: float, start_percent: float = 0.0, end_percent: float = 1.0, *, bounds: Optional[dict] = None) -> 'GuiderRef': ...
    # Project one sampling percentage through the model's schedule.
    async def sigma_for_percent(self, percent: float, actual_endpoints: bool = False) -> float: ...
    # Clone the model with model-owned spatial inputs cropped to tiles.
    async def spatial_crop_inputs(self, *, regions: list[tuple[int, int, int, int]], source_width: int, source_height: int, target_width: int, target_height: int) -> 'ModelRef': ...
    # Return the transforms supported by the active host.
    async def transforms(self) -> list[dict]: ...
    # Return a model's scalar UNet context dimension when published.
    async def unet_context_dim(self) -> Optional[int]: ...

class ModelsDomain:
    def __init__(self, *args, **kwargs): ...
    async def download_huggingface_weights(self, repo_id: str, filename: str, folder: str, revision: str = 'main', sha256: Optional[str] = None) -> str: ...
    async def generate_text(self, generator: str, input_text: str, max_new_tokens: int = 128, weight: Optional[str] = None) -> str: ...
    async def list_controlnet(self) -> list[str]: ...
    async def list_diffusion_models(self, include_connectors: bool = False) -> list[str]: ...
    async def list_vae(self) -> list[str]: ...
    async def load_advanced_controlnet(self, name: str, model: Optional[ModelRef] = None, timestep_keyframe: Optional[TimestepKeyframeRef] = None) -> ControlNetRef: ...
    async def load_background_removal_model(self, model: str) -> BackgroundRemovalModelRef: ...
    async def load_brushnet(self, model: str, dtype: str = 'float16') -> BrushNetRef: ...
    async def load_checkpoint(self, name: str, weight_dtype: str = 'default', compute_dtype: str = 'default', cublas_linear: bool = False) -> tuple[ModelRef, Optional[ClipRef], Optional[VaeRef]]: ...
    async def load_clip_vision(self, model: str) -> ClipVisionRef: ...
    async def load_clipseg(self, model: str) -> ClipSegRef: ...
    async def load_controlnet(self, name: str, model: Optional[ModelRef] = None) -> ControlNetRef: ...
    async def load_controlnet_plusplus(self, name: str, control_type: str = 'none') -> ControlNetRef: ...
    async def load_diffusion_model(self, name: str, extra_name: Optional[str] = None, weight_dtype: str = 'default', compute_dtype: str = 'default', cublas_linear: bool = False) -> ModelRef: ...
    async def load_gguf_model(self, name: str, extra_name: Optional[str] = None, dequant_dtype: str = 'default', patch_dtype: str = 'default', patch_on_device: bool = False) -> ModelRef: ...
    async def load_gguf_text_encoders(self, names: Sequence[str], clip_type: str) -> ClipRef: ...
    async def load_image_classifier(self, model: str, architecture: str, labels: list[str]) -> ImageClassifierRef: ...
    async def load_inpaint_model(self, model: str, architecture: str = 'big-lama') -> InpaintModelRef: ...
    async def load_ipadapter(self, model: str, clip_vision: ClipVisionRef) -> IpAdapterRef: ...
    async def load_language_model(self, weights: list[str], family: str, device: str = 'default', cache: bool = True) -> ClipRef: ...
    async def load_object_detector(self, model: str) -> ObjectDetectorRef: ...
    async def load_onnx_detector(self, model: str) -> OnnxDetectorRef: ...
    async def load_onnx_image_classifier(self, model: str, input_layout: str = 'NHWC', channel_order: str = 'BGR', resize_mode: str = 'fit_pad', input_scale: float = 255.0, pad_color: tuple[float, float, float] = (1.0, 1.0, 1.0), mean: tuple[float, float, float] = (0.0, 0.0, 0.0), std: tuple[float, float, float] = (1.0, 1.0, 1.0), activation: str = 'identity', resize_filter: str = 'lanczos') -> ImageClassifierRef: ...
    async def load_powerpaint(self, model: str, base_clip: str, powerpaint_clip: str, dtype: str = 'float16') -> PowerPaintRef: ...
    async def load_sam(self, model: str, architecture: str = 'vit_b', device_mode: str = 'AUTO') -> SamModelRef: ...
    async def load_segformer(self, model: str, variant: str, num_labels: int) -> SemanticSegmentationRef: ...
    async def load_text_encoder(self, model: str, model_type: str, device: str = 'default') -> ClipRef: ...
    async def load_transparent_vae_decoder(self, model: str, family: str) -> TransparentVaeDecoderRef: ...
    async def load_upscale_model(self, name: str) -> UpscaleModelRef: ...
    async def load_vae(self, name: str, device: str = 'default', weight_dtype: str = 'default') -> VaeRef: ...
    async def load_vitmatte(self, model: str, variant: str) -> MattingModelRef: ...
    async def load_vqa(self, model: str, architecture: str, precision: str = 'fp16', device: str = 'cuda') -> VqaModelRef: ...
    async def memory_cleanup(self, empty_cache: bool = True, collect_cycles: bool = True, unload_all_models: bool = False) -> tuple[int, int]: ...

class OnnxDetectorRef(_TypedRef):
    KIND: Any = ...
    # Run a catalogued ONNX object detector on one host-side image.
    async def detect(self, image: ImageRef) -> list[dict[str, Any]]: ...

class ObjectDetectorRef(_TypedRef):
    KIND: Any = ...
    async def detect(self, image: ImageRef, threshold: float = 0.5, class_name: str = 'all', max_detections: int = 100) -> list[list[dict[str, Any]]]: ...

class PowerPaintRef(_TypedRef):
    """Opaque PowerPaint model and token-extended CLIP pipeline."""
    KIND: Any = ...
    async def apply(self, model: ModelRef, vae: VaeRef, image: ImageRef, mask: MaskRef, positive: CondRef, negative: CondRef, fitting: float = 1.0, function: str = 'text guided', scale: float = 1.0, start_step: int = 0, end_step: int = 10000, save_memory: str = 'none') -> tuple[ModelRef, CondRef, CondRef, LatentRef]: ...

class ClipRef(_TypedRef):
    KIND: Any = ...
    # Describe bounded token IDs without exposing tokenizer objects.
    async def describe_tokens(self, tokens: dict) -> dict: ...
    # The two steps above in one call, for the common case.
    async def encode(self, text: str) -> 'CondRef': ...
    # Mirrors ``clip.encode_from_tokens_scheduled``, ``add_dict`` included.
    async def encode_from_tokens_scheduled(self, tokens: dict, add_dict: dict | None = None) -> 'CondRef': ...
    # Encode token-weight pairs with one named CLIP component.
    async def encode_token_weights_component(self, component: str, tokens: list) -> tuple[TensorRef, Optional[TensorRef]]: ...
    # Generate bounded text with a canonical Comfy text encoder.
    async def generate_text(self, prompt: str, image: Optional[ImageRef] = None, video: Optional[ImageRef] = None, max_length: int = 256, do_sample: bool = False, temperature: float = 1.0, top_k: Optional[int] = 50, top_p: float = 0.95, min_p: float = 0.0, repetition_penalty: float = 1.0, seed: Optional[int] = None, presence_penalty: float = 0.0, thinking: bool = False, use_default_template: bool = True, num_beams: int = 1) -> str: ...
    async def lora_weight_differences(self, original: 'ClipRef', include_bias: bool = False) -> 'WeightDiffCursorRef': ...
    # Scale selected CLIP/T5 attention projection weights.
    async def scale_attention_weights(self, *, clip_l: Optional[list[float]] = None, clip_g: Optional[list[float]] = None, t5xxl: Optional[list[float]] = None, query: bool = True, key: bool = True, value: bool = True, output: bool = True) -> 'ClipRef': ...
    # Clone this CLIP and stop encoding at a bounded hidden layer.
    async def set_last_layer(self, stop_at_clip_layer: int) -> 'ClipRef': ...
    # Mirrors ``clip.tokenize``, including its per-model kwargs.
    async def tokenize(self, text: str, **kwargs: Any) -> dict: ...
    # Clone this encoder with one host-registered attention function.
    async def with_attention_impl(self, mode: str) -> 'ClipRef': ...

class ClipSegRef(_TypedRef):
    KIND: Any = ...
    # Return native-resolution sigmoid CLIPSeg predictions.
    async def predict_mask(self, images: ImageRef, text: str, use_accelerator: bool = True) -> MaskRef: ...
    async def segment(self, images: ImageRef, text: str, threshold: float = 0.5, binary_mask: bool = True, combine_mask: bool = False, use_accelerator: bool = True, blur_sigma: float = 0.0, previous_mask: Optional[MaskRef] = None, invert: bool = False, image_background_level: float = 0.5) -> tuple[MaskRef, ImageRef]: ...

class ClipVisionRef(_TypedRef):
    KIND: Any = ...
    async def encode_image(self, image: ImageRef, crop: bool = True) -> ClipVisionOutputRef: ...

class ClipVisionOutputRef(_TypedRef):
    KIND: Any = ...
    # Concatenate opaque penultimate vision tokens along their token axis.
    async def concat(self, other: 'ClipVisionOutputRef') -> 'ClipVisionOutputRef': ...
    async def image_embeds(self) -> TensorRef: ...

class ControlNetRef(_TypedRef):
    KIND: Any = ...
    # Apply this ControlNet while all referenced data stays host-owned.
    async def apply(self, positive: CondRef, negative: CondRef, image: ImageRef, strength: float = 1.0, start_percent: float = 0.0, end_percent: float = 1.0, vae: Optional[VaeRef] = None) -> tuple[CondRef, CondRef]: ...
    # Apply Advanced-ControlNet scheduling, weights, and effect masks.
    async def apply_advanced(self, positive: CondRef, negative: CondRef, image: ImageRef, strength: float = 1.0, start_percent: float = 0.0, end_percent: float = 1.0, vae: Optional[VaeRef] = None, mask: Optional[MaskRef] = None, timestep_keyframe: Optional[TimestepKeyframeRef] = None, weights: Optional[ControlNetWeightsRef] = None) -> tuple[CondRef, CondRef]: ...
    async def compile(self, *, backend: str = 'inductor', mode: str = 'default', fullgraph: bool = False) -> 'ControlNetRef': ...
    async def with_union_type(self, type_number: Optional[int]) -> 'ControlNetRef': ...

class ControlNetWeightsRef(_TypedRef):
    """Opaque Advanced-ControlNet weight policy."""
    KIND: Any = ...
    @classmethod
    def from_list(cls, weights: list[float], uncond_multiplier: float = 1.0, extras: Any = None) -> tuple['ControlNetWeightsRef', 'TimestepKeyframeRef']: ...
    @classmethod
    def scaled_soft(cls, base_multiplier: float = 0.825, uncond_multiplier: float = 1.0) -> tuple['ControlNetWeightsRef', 'TimestepKeyframeRef']: ...

class StyleModelRef(_TypedRef):
    KIND: Any = ...
    async def apply(self, clip_vision_output: ClipVisionOutputRef, conditioning: CondRef, strength: float = 1.0) -> CondRef: ...

class SystemDomain:
    def __init__(self, *args, **kwargs): ...
    async def monitor(self) -> dict[str, Any]: ...
    async def stats(self) -> dict[str, Any]: ...

class VaeRef(_TypedRef):
    KIND: Any = ...
    async def compile(self, *, backend: str = 'inductor', mode: str = 'default', fullgraph: bool = False, encoder: bool = True, decoder: bool = True) -> 'VaeRef': ...
    async def decode(self, latent: 'LatentRef') -> 'ImageRef': ...
    # Decode an audio latent and attach the VAE's output sample rate.
    async def decode_audio(self, latent: 'LatentRef') -> 'AudioRef': ...
    # Decode to an opaque BHWC tensor without assuming RGB channels.
    async def decode_tensor(self, latent: 'LatentRef') -> 'TensorRef': ...
    # Tiled :meth:`decode_tensor` with the canonical VAE tile options.
    async def decode_tensor_tiled(self, latent: 'LatentRef', tile_size: int = 512, overlap: int = 64, temporal_size: int = 64, temporal_overlap: int = 8) -> 'TensorRef': ...
    # Decode through ComfyUI's bounded, pixel-sized tiled operation.
    async def decode_tiled(self, latent: 'LatentRef', tile_size: int = 512, overlap: int = 64, temporal_size: int = 64, temporal_overlap: int = 8) -> 'ImageRef': ...
    # Decode image/video latents and flatten video batches to frames.
    async def decode_video(self, latent: 'LatentRef', *, tiled: bool = False, tile_size: int = 512, overlap: int = 64, temporal_size: int = 4096, temporal_overlap: int = 16) -> 'ImageRef': ...
    async def downscale_index_formula(self) -> Optional[tuple[int, int, int]]: ...
    # Mirrors ``vae.encode`` exactly. The caller owns any channel slicing.
    async def encode(self, image: 'ImageRef') -> 'LatentRef': ...
    # Run core VAEEncodeForInpaint with a bounded mask grow amount.
    async def encode_for_inpaint(self, image: 'ImageRef', mask: 'MaskRef', grow_mask_by: int = 6) -> 'LatentRef': ...
    # Run core InpaintModelConditioning without exposing model tensors.
    async def encode_inpaint_conditioning(self, image: 'ImageRef', mask: 'MaskRef', positive: 'CondRef', negative: 'CondRef', noise_mask: bool = True) -> tuple['CondRef', 'CondRef', 'LatentRef']: ...
    async def encode_tiled(self, image: 'ImageRef', tile_x: Optional[int] = None, tile_y: Optional[int] = None, overlap: Optional[int] = None, tile_t: Optional[int] = None, overlap_t: Optional[int] = None) -> 'LatentRef': ...
    # Encode frames after applying this VAE's temporal frame constraint.
    async def encode_video(self, image: 'ImageRef') -> tuple['LatentRef', int]: ...
    # The CPU dtype a node should use before VAE-specific preprocessing.
    async def input_dtype(self) -> str: ...
    # Return the VAE's bounded latent channel/compression metadata.
    async def latent_layout(self) -> dict[str, Optional[int]]: ...
    async def merge(self, other: 'VaeRef', ratio: float = 0.5) -> 'VaeRef': ...
    async def patch_triton(self, *, fuse_norm_silu: bool = True, channels_last: bool = True, int8_conv: bool = False, autotune: bool = False) -> 'VaeRef': ...

class AudioRef(ValueRef):
    KIND: Any = ...

class BackgroundRemovalModelRef(_TypedRef):
    """Opaque ComfyUI background-removal model."""
    KIND: Any = ...
    # Generate a foreground alpha mask through core's canonical model.
    async def mask(self, image: ImageRef) -> MaskRef: ...

class BrushNetRef(_TypedRef):
    """Opaque BrushNet weights loaded by the canonical host extension."""
    KIND: Any = ...
    async def apply(self, model: ModelRef, vae: VaeRef, image: ImageRef, mask: MaskRef, positive: CondRef, negative: CondRef, scale: float = 1.0, start_step: int = 0, end_step: int = 10000) -> tuple[ModelRef, CondRef, CondRef, LatentRef]: ...

class VideoRef(_TypedRef):
    KIND: Any = ...
    # Return encoded bytes and trim metadata, never the source path.
    async def encoded_source(self) -> ValueRef: ...

class VqaModelRef(_TypedRef):
    """Opaque visual question-answering model."""
    KIND: Any = ...
    async def answer(self, image: ImageRef, question: str, max_new_tokens: int = 32) -> str: ...

class TimestepKeyframeRef(_TypedRef):
    """Opaque Advanced-ControlNet timestep-keyframe schedule."""
    KIND: Any = ...

class TransparentVaeDecoderRef(_TypedRef):
    """Opaque canonical decoder for Layer Diffusion transparency weights."""
    KIND: Any = ...
    # Return a consistent RGBA batch and decoded alpha masks.
    async def decode(self, latent: LatentRef, image: ImageRef, frames: int = 1, sub_batch_size: int = 16) -> tuple[ImageRef, MaskRef]: ...

class WeightDiffCursorRef(_TypedRef):
    """Execution-scoped iterator over host-owned model weight differences."""
    KIND: Any = ...
    async def next(self) -> Optional[dict[str, Any]]: ...

class AssetRef(_TypedRef):
    """A file/model resolved by name+hash, tenant-scoped. Never a raw path."""
    KIND: Any = ...

class AssetsDomain:
    def __init__(self, *args, **kwargs): ...
    async def delete_input(self, name: str) -> bool: ...
    async def digest(self, ref: AssetRef, algorithm: str = 'sha256') -> str: ...
    async def exists(self, folder: str, name: str) -> bool: ...
    async def latest(self, folder: str, prefix: str = '', suffix: str = '') -> Optional[str]: ...
    async def list(self, folder: str, prefix: str = '', recursive: bool = True) -> list[str]: ...
    async def load_image(self, ref: AssetRef) -> ImageRef: ...
    async def load_latent(self, ref: AssetRef) -> LatentRef: ...
    async def load_state_dict(self, ref: AssetRef, return_metadata: bool = False) -> Any: ...
    async def path(self, ref: AssetRef) -> str: ...
    async def read_bytes(self, ref: AssetRef) -> bytes: ...
    async def read_range(self, ref: AssetRef, offset: int = 0, length: int = 8388608) -> bytes: ...
    async def resolve(self, folder: str, name: str) -> AssetRef: ...
    async def size(self, ref: AssetRef) -> int: ...

class OutputDomain:
    def __init__(self, *args, **kwargs): ...
    async def save_animation(self, images: ImageRef, fps: float = 8.0, filename_prefix: str = 'animation/ComfyUI', format: str = 'webp', loop_count: int = 0, lossless: bool = True, quality: int = 90, save_output: bool = True) -> dict: ...
    async def save_image_sequence(self, images: ImageRef, filename_prefix: str = 'sequence/ComfyUI', format: str = 'png', bit_depth: int = 8, save_output: bool = True) -> dict: ...
    async def save_images(self, images: ImageRef, filename_prefix: str = 'ComfyUI', subfolder: str = '', compress_level: int = 4, caption: Optional[str] = None, caption_extension: str = '.txt', save_metadata: bool = True, extra_metadata: Optional[dict[str, Any]] = None, image_format: str = 'png', quality: int = 95, filenames: Optional[list[str]] = None, lossless: bool = False, optimize: bool = False) -> dict: ...
    async def save_images_with_alpha(self, images: ImageRef, mask: MaskRef, filename_prefix: str = 'ComfyUI', subfolder: str = '', compress_level: int = 4) -> dict: ...
    async def save_latent(self, latent: LatentRef, filename_prefix: str = 'latents/LatentSender', preview_method: str = 'Latent2RGB-SDXL') -> dict: ...
    async def save_model(self, model: ModelRef, filename_prefix: str, model_key_prefix: str = 'model.diffusion_model.') -> str: ...
    async def save_state_dict(self, state_dict: ValueRef, filename_prefix: str, metadata: Optional[dict[str, str]] = None) -> str: ...
    async def save_text(self, text: str, filename_prefix: str = 'text', subfolder: str = '', extension: str = '.txt') -> str: ...
    async def save_video(self, images: ImageRef, audio: Optional[AudioRef] = None, fps: float = 25.0, filename_prefix: str = 'video/ComfyUI', format: str = 'auto', codec: str = 'auto', encoder_options: Optional[dict[str, Any]] = None, loop_count: int = 0, bit_depth: int = 8, save_output: bool = True, save_metadata: bool = True) -> dict: ...
    async def save_workflow_json(self, filename: str, mode: str = 'new_only') -> str: ...
    async def write_text(self, text: str, filename: str, folder: str = 'output', mode: str = 'overwrite', insert_newline: bool = False) -> str: ...

class GraphDomain:
    def __init__(self, *args, **kwargs): ...
    async def block(self, reason: Optional[str] = None) -> Any: ...
    async def current_node_id(self) -> str: ...
    async def expand_loop(self, flow: Any, values: list[Any]) -> dict[str, Any]: ...
    async def expand_nodes(self, nodes: list[dict[str, Any]], outputs: list[dict[str, Any]]) -> dict[str, Any]: ...
    async def input_label(self, input_name: str, default: str = '') -> str: ...
    async def widget_values(self, node_id: int | str = 0, node_title: str = '', node_name: str = '', linked_input: str = 'any_input') -> dict[str, Any]: ...

class Context:
    def __init__(self, *args, **kwargs): ...

class CivitaiDomain:
    """Bounded read-only projection of the Civitai public model API."""
    def __init__(self, *args, **kwargs): ...
    async def model_version(self, model_version_id: int) -> dict[str, Any]: ...
    async def model_version_by_hash(self, hash_value: str, refresh: bool = False) -> dict[str, Any]: ...
    async def search_models(self, username: str, query: Optional[str] = None, limit: int = 20, nsfw: bool = False) -> dict[str, Any]: ...

class OllamaDomain:
    """Bounded Ollama vendor API; endpoint is loopback or an admin profile."""
    def __init__(self, *args, **kwargs): ...
    async def chat(self, endpoint: str, model: str, messages: list[dict[str, Any]], images: Optional[ImageRef] = None, think: bool = False, options: Optional[dict[str, Any]] = None, keep_alive: int = 5, keep_alive_unit: str = 'minutes', format: str | dict[str, Any] = '', timeout_seconds: float = 600.0, tools: Optional[list[dict[str, Any]]] = None) -> dict[str, Any]: ...
    async def generate(self, endpoint: str, model: str, system: str, prompt: str, images: Optional[ImageRef] = None, context: Optional[list[int]] = None, think: bool = False, options: Optional[dict[str, Any]] = None, keep_alive: int = 5, keep_alive_unit: str = 'minutes', format: str | dict[str, Any] = '', timeout_seconds: float = 600.0) -> dict[str, Any]: ...
    async def list_models(self, endpoint: str) -> list[str]: ...

class WanVideoDomain:
    """Bounded metadata for WanVideo vendor-owned opaque model handles."""
    def __init__(self, *args, **kwargs): ...
    async def transformer_dim(self, model: Ref) -> int: ...

class WebSearchDomain:
    """Fixed-profile web search with bounded normalized results."""
    def __init__(self, *args, **kwargs): ...
    async def search(self, query: str, *, provider_profile: str = 'duckduckgo', limit: int = 5, vendor_options: Optional[dict[str, Any]] = None) -> list[dict[str, str]]: ...

class IntegrationsDomain:
    """Vendor pass-throughs with vendor-shaped, less-stable contracts."""
    def __init__(self, *args, **kwargs): ...

class ExecutionDomain:
    def __init__(self, *args, **kwargs): ...
    async def interrupt(self) -> bool: ...

def ctx() -> 'Context': ...

def current_context() -> 'Context': ...

class RefResolver:
    def __init__(self, *args, **kwargs): ...
    async def create(self, kind: str, obj: Any) -> Ref: ...
    async def release(self, ref: Ref) -> None: ...
    async def resolve(self, ref: Ref) -> Any: ...

class ExecutionBackend:
    def __init__(self, *args, **kwargs): ...
    # Run the node. Default just awaits ``local_call`` (in-process). The
    async def dispatch(self, plan: ExecutionPlan, local_call: Callable[[], Awaitable[Any]], runtime: Optional[Runtime] = None) -> Any: ...

class ExecutionPlan:
    """What the execution seam hands the backend to decide placement."""
    def __init__(self, prompt_id: str, node_id: str, node_type: str, tier: str = 'default', permissions: tuple[str, ...] = (), required_weights: tuple[HuggingFaceWeight, ...] = (), node_module: str = '', inputs: Optional[dict] = None, input_mode: str = 'refs', prompt: Any = None, extra_pnginfo: Any = None, dynamic_prompt: Any = None, method: str = 'execute') -> None: ...
    dynamic_prompt: Any = ...
    extra_pnginfo: Any = ...
    input_mode: Any = ...
    inputs: Any = ...
    method: Any = ...
    node_module: Any = ...
    permissions: Any = ...
    prompt: Any = ...
    required_weights: Any = ...
    tier: Any = ...

class OpNotSupported(NotImplementedError):
    """Raised by ``apply`` for an op this provider does not implement. Carries"""
    def __init__(self, op: str) -> None: ...

providers: Providers = ...