feat: add versioned lossless serverless media transport

api/config_routes.py

@@ -97,10 +97,11 @@ _SECRET_FIELDS = {"api_key", "clear_api_key"}
 _SERVERLESS_LOCAL_FIELDS = {"host", "port", "listen_port", "cuda_device", "extra_args"}
 _SERVERLESS_ONLY_FIELDS = {
     "provider", "endpoint_id", "jobs_per_run", "concurrency", "timeout_seconds", "api_key_configured",
+    "media_profile",
 }
 _SERVERLESS_CONFIG_FIELDS = {
     "worker_id", "type", "provider", "endpoint_id", "jobs_per_run", "concurrency",
-    "timeout_seconds", "enabled", "name", "api_key", "clear_api_key",
+    "timeout_seconds", "enabled", "name", "api_key", "clear_api_key", "media_profile",
 }
 
 
@@ -199,7 +200,13 @@ def _clean_bulk_type_conversions(old_workers, new_workers):
         if not isinstance(worker, dict) or worker.get("id") is None:
             continue
         old = old_by_id.get(str(worker["id"]))
-        if old is None or old.get("type") != "serverless" or worker.get("type") == "serverless":
+        if old is None or old.get("type") != "serverless":
+            continue
+        if worker.get("type") == "serverless":
+            # Older clients do not know this field. Only an explicit selection
+            # may change the endpoint contract, including on bulk saves.
+            if "media_profile" not in worker and "media_profile" in old:
+                worker["media_profile"] = old["media_profile"]
             continue
         converted = dict(worker)
         for field in _SERVERLESS_ONLY_FIELDS:
@@ -218,7 +225,10 @@ def _bulk_serverless_changes(old_workers, new_workers):
             if new is None or new.get("type") != "serverless":
                 _ensure_pool_idle(worker_id)
                 secret_ids.add(worker_id)
-            elif old.get("endpoint_id") != new.get("endpoint_id"):
+            elif (
+                old.get("endpoint_id") != new.get("endpoint_id")
+                or old.get("media_profile", "legacy-image-v1") != new.get("media_profile", "legacy-image-v1")
+            ):
                 _ensure_pool_idle(worker_id)
         elif new is not None and new.get("type") == "serverless":
             secret_ids.add(worker_id)
@@ -401,7 +411,7 @@ async def _update_worker_endpoint(request, secret_changes):
                 candidate.setdefault("provider", "runpod")
                 candidate.setdefault("enabled", False)
                 # Legacy "concurrency" is accepted (see _SERVERLESS_CONFIG_FIELDS) but ignored.
-                for field in ("provider", "endpoint_id", "jobs_per_run", "timeout_seconds", "enabled", "name"):
+                for field in ("provider", "endpoint_id", "jobs_per_run", "timeout_seconds", "enabled", "name", "media_profile"):
                     if field in data:
                         candidate[field] = data[field]
                 candidate = _clean_serverless_worker(candidate)
@@ -424,8 +434,12 @@ async def _update_worker_endpoint(request, secret_changes):
                     current_type == "serverless"
                     and existing.get("endpoint_id") != validated.get("endpoint_id")
                 )
+                media_profile_changed = (
+                    current_type == "serverless"
+                    and existing.get("media_profile", "legacy-image-v1") != validated.get("media_profile", "legacy-image-v1")
+                )
                 if current_type == "serverless" and (
-                    endpoint_changed or next_secret != prior_secret
+                    endpoint_changed or media_profile_changed or next_secret != prior_secret
                 ):
                     _ensure_pool_idle(worker_id)
 

api/orchestration/prompt_transform.py

@@ -401,6 +401,20 @@ def prune_prompt_for_worker(prompt_obj, use_worker_sink=False):
 
 def prepare_delegate_master_prompt(prompt_obj, collector_ids):
     """Prune master prompt so it only executes post-collector nodes in delegate mode."""
+    # Typed collectors need no fabricated IMAGE placeholder. Sever all media
+    # links, including STRING scalar branches, before retained-config analysis.
+    # Their expected_media control was frozen by participant overrides.
+    if any((prompt_obj.get(cid) or {}).get("class_type") == "DistributedMediaCollector"
+           for cid in collector_ids):
+        from ...serverless_bridge.contract import SOCKET_KINDS
+        prompt_obj = json.loads(json.dumps(prompt_obj))
+        for cid in collector_ids:
+            node = prompt_obj.get(cid) or {}
+            if node.get("class_type") == "DistributedMediaCollector":
+                inputs = node.setdefault("inputs", {})
+                inputs.setdefault("expected_media", json.dumps([name for name in SOCKET_KINDS if name in inputs]))
+                for name in SOCKET_KINDS:
+                    inputs.pop(name, None)
     downstream = _find_downstream_nodes(prompt_obj, collector_ids)
     nodes_to_keep = set(collector_ids)
     nodes_to_keep.update(downstream)
@@ -505,6 +519,7 @@ def generate_job_id_map(prompt_index, prefix):
     job_map = {}
     distributed_nodes = (
         prompt_index.nodes_for_class("DistributedCollector")
+        + prompt_index.nodes_for_class("DistributedMediaCollector")
         + prompt_index.nodes_for_class("UltimateSDUpscaleDistributed")
         + prompt_index.nodes_for_class("DistributedImageListStart")
     )
@@ -537,8 +552,10 @@ def _override_collector_nodes(
     enabled_json,
     delegate_master,
 ):
-    """Configure DistributedCollector nodes for master or worker role."""
-    for node_id in prompt_index.nodes_for_class("DistributedCollector"):
+    """Configure legacy and typed collectors without changing their media."""
+    collector_ids = (prompt_index.nodes_for_class("DistributedCollector")
+                     + prompt_index.nodes_for_class("DistributedMediaCollector"))
+    for node_id in collector_ids:
         node = prompt_copy.get(node_id)
         if not isinstance(node, dict):
             continue
@@ -551,6 +568,9 @@ def _override_collector_nodes(
         inputs["multi_job_id"] = job_id_map.get(node_id, node_id)
         inputs["is_worker"] = not is_master
         inputs["enabled_worker_ids"] = enabled_json
+        if node.get("class_type") == "DistributedMediaCollector":
+            from ...serverless_bridge.contract import SOCKET_KINDS
+            inputs["expected_media"] = json.dumps([socket for socket in SOCKET_KINDS if socket in inputs])
         if is_master:
             inputs["delegate_only"] = bool(delegate_master)
             inputs.pop("master_url", None)

api/orchestration/runpod_client.py

@@ -18,7 +18,7 @@ class AmbiguousSubmission(RunpodError):
     """Submission may have been accepted: never retry it automatically."""
 
 
-def serialize_request(workflow, images, timeout_seconds):
+def serialize_request(workflow, images, timeout_seconds, *, media=None):
     """Serialize and size-check the exact JSON body used by Runpod ``/run``."""
     payload = {
         "input": {"workflow": workflow, "images": images},
@@ -27,6 +27,10 @@ def serialize_request(workflow, images, timeout_seconds):
             "ttl": timeout_seconds * 1000,
         },
     }
+    if media is not None:
+        from ...serverless_bridge.handler import _validate_request
+        payload["input"]["media"] = media
+        _validate_request(payload["input"])
     body = json.dumps(
         payload,
         allow_nan=False,
@@ -108,8 +112,8 @@ class RunpodClient:
     async def health(self):
         return await self._request("GET", "health")
 
-    async def submit(self, workflow, images, timeout_seconds):
-        body = serialize_request(workflow, images, timeout_seconds)
+    async def submit(self, workflow, images, timeout_seconds, *, media=None):
+        body = serialize_request(workflow, images, timeout_seconds, media=media)
         result = await self._request("POST", "run", request_body=body)
         try:
             return checked_id(result.get("id"))

api/orchestration/serverless_media_prompt.py

@@ -0,0 +1,191 @@
+"""Pure compilation of prepared typed-media collectors for serverless workers.
+
+The legacy image compiler remains separate. This compiler emits one MediaExportV1
+execution target per connected socket, retaining only its upstream computation.
+It never imports ComfyUI or inspects/executes runtime node classes.
+"""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from graphlib import CycleError, TopologicalSorter
+
+from ...serverless_bridge.contract import SOCKET_KINDS, identifier, validate_bindings
+from .serverless_prompt import (
+    _PRIMITIVE_NODES,
+    _REMOVED_SCALAR_NODES,
+    _is_link,
+    _is_unsupported_distributed,
+    _make_scalar_resolver,
+    _prune_graph,
+)
+
+
+_COLLECTOR = "DistributedMediaCollector"
+
+
+def _validate_identifier(value, context):
+    # Validate the original token before any lookup. In particular, neither
+    # integer node IDs nor whitespace in prepared identities is normalized.
+    try:
+        identifier(value)
+    except ValueError:
+        raise ValueError(f"{context} must be a valid nonempty media identifier") from None
+
+
+def _validate_prompt(prepared_prompt):
+    """Validate the entire source graph, including branches later discarded."""
+    if not isinstance(prepared_prompt, dict) or not prepared_prompt:
+        raise ValueError("prepared prompt must be a nonempty API graph mapping")
+
+    collectors = []
+    for node_id, node in prepared_prompt.items():
+        _validate_identifier(node_id, "prompt node ID")
+        if not isinstance(node, dict):
+            raise ValueError(f"prompt node {node_id!r} must be an object")
+        class_type = node.get("class_type")
+        if not isinstance(class_type, str) or not class_type.strip():
+            raise ValueError(f"prompt node {node_id!r} must have a nonempty class_type")
+        if class_type == _COLLECTOR:
+            collectors.append(node_id)
+        elif class_type == "DistributedCollector" or _is_unsupported_distributed(class_type):
+            raise ValueError(f"unsupported distributed or batch node: {class_type}")
+        inputs = node.get("inputs", {})
+        if not isinstance(inputs, dict):
+            raise ValueError(f"prompt node {node_id!r} inputs must be an object")
+        if any(not isinstance(name, str) or not name for name in inputs):
+            raise ValueError(f"prompt node {node_id!r} input names must be nonempty strings")
+
+    if not collectors:
+        raise ValueError("prepared prompt contains no DistributedMediaCollector nodes")
+
+    dependencies = {}
+    for node_id, node in prepared_prompt.items():
+        parents = set()
+        for input_name, value in node.get("inputs", {}).items():
+            # Ordinary inputs may contain literal lists. As in the legacy
+            # compiler, exactly two elements denotes an API link, not media.
+            if not _is_link(value):
+                continue
+            source_id, slot = value
+            context = f"link at node {node_id!r} input {input_name!r}"
+            _validate_identifier(source_id, context + " source ID")
+            if source_id not in prepared_prompt:
+                raise ValueError(f"invalid {context}: source {source_id!r} does not exist")
+            if type(slot) is not int or slot < 0:
+                raise ValueError(f"invalid {context}: output slot must be a nonnegative integer")
+            source_class = prepared_prompt[source_id]["class_type"]
+            if source_class in _REMOVED_SCALAR_NODES or source_class in _PRIMITIVE_NODES:
+                if slot != 0:
+                    raise ValueError(f"invalid output slot {slot} for scalar node {source_id!r}")
+            elif source_class == _COLLECTOR and slot >= len(SOCKET_KINDS):
+                raise ValueError(f"invalid output slot {slot} for collector {source_id!r}")
+            parents.add(source_id)
+        dependencies[node_id] = parents
+
+    for collector_id in collectors:
+        inputs = prepared_prompt[collector_id].get("inputs", {})
+        for field in ("multi_job_id", "worker_id"):
+            _validate_identifier(inputs.get(field), f"collector {collector_id!r} {field}")
+        connected = [socket for socket in SOCKET_KINDS if socket in inputs]
+        if not connected:
+            raise ValueError(f"collector {collector_id!r} requires at least one connected media socket")
+        for socket in connected:
+            value = inputs[socket]
+            if not _is_link(value):
+                raise ValueError(f"collector {collector_id!r} {socket} must be linked; use PrimitiveString for literal text")
+            if prepared_prompt[value[0]]["class_type"] in _REMOVED_SCALAR_NODES:
+                # Export sockets must remain actual links after scalar folding.
+                # A PrimitiveString wrapper supports distributed text values.
+                raise ValueError(f"collector {collector_id!r} {socket} cannot link directly to a distributed scalar; use a typed node or PrimitiveString")
+
+    # Iterative topological sorting checks ordinary and unused cycles too, not
+    # only the primitive chains that the shared scalar resolver traverses.
+    try:
+        ordered = list(TopologicalSorter(dependencies).static_order())
+    except CycleError:
+        raise ValueError("cycle in prepared media prompt") from None
+    collector_set = set(collectors)
+    has_collector_ancestor = {}
+    for node_id in ordered:
+        has_collector_ancestor[node_id] = any(
+            parent in collector_set or has_collector_ancestor[parent]
+            for parent in dependencies[node_id]
+        )
+        if node_id in collector_set and has_collector_ancestor[node_id]:
+            raise ValueError("collector-to-collector dependence is unsupported")
+
+    return sorted(collectors)
+
+
+def _build_exports(source_nodes, collector_ids, request_id):
+    bindings = []
+    exports = {}
+    reserved = set(source_nodes)
+    for collector_id in collector_ids:
+        inputs = source_nodes[collector_id].get("inputs", {})
+        for socket, kind in SOCKET_KINDS.items():
+            if socket not in inputs:
+                continue
+            # Hashing avoids separator ambiguity and keeps long valid source
+            # identities within the bridge's identifier length bound. Identity
+            # is stable across request IDs and dictionary insertion orders.
+            identity = json.dumps([collector_id, socket], separators=(",", ":"))
+            digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()
+            base_id = f"serverless:media:{digest}"
+            node_id = base_id
+            suffix = 0
+            while node_id in reserved:
+                suffix += 1
+                node_id = f"{base_id}:{suffix}"
+            reserved.add(node_id)
+            binding = {
+                "binding_id": f"b:{digest}",
+                "collector_id": collector_id,
+                "job_id": inputs["multi_job_id"],
+                "participant_id": inputs["worker_id"],
+                "node_id": node_id,
+                "socket": socket,
+                "kind": kind,
+            }
+            bindings.append(binding)
+            exports[node_id] = {
+                "class_type": "MediaExportV1",
+                "inputs": {
+                    "request_id": request_id,
+                    "binding_json": json.dumps(binding),
+                    socket: copy.deepcopy(inputs[socket]),
+                },
+            }
+    validate_bindings(bindings)
+    return bindings, exports
+
+
+def compile_media_prompt(prepared_prompt: dict, request_id: str) -> tuple[dict, list[dict]]:
+    """Return a worker graph and canonical typed-media export bindings.
+
+    Each DistributedMediaCollector must have prepared literal ``multi_job_id``
+    and ``worker_id`` identities and at least one linked media input. Present
+    media inputs (including texts) must be API links; literal text belongs in a
+    PrimitiveString node. Nonmedia collector controls are not exported.
+
+    Only MediaExportV1 targets and their ancestors survive. Required ordinary
+    output/utility nodes are preserved regardless of their runtime OUTPUT_NODE
+    flag; callers must dispatch exactly the returned binding ``node_id`` values
+    as partial_execution_targets. Distributed scalar dependencies use the same
+    resolver as the legacy compiler. Direct scalar-to-media links need a typed
+    wrapper so that each export input remains a link.
+
+    Validation and compilation never modify the caller's source graph. This
+    structural compiler does not infer socket types or unknown nodes' output
+    counts; ComfyUI performs runtime type validation at the endpoint.
+    """
+    _validate_identifier(request_id, "request_id")
+    collector_ids = _validate_prompt(prepared_prompt)
+    bindings, exports = _build_exports(prepared_prompt, collector_ids, request_id)
+    compiled = _prune_graph(
+        prepared_prompt, bindings, exports, _make_scalar_resolver(prepared_prompt)
+    )
+    return {node_id: compiled[node_id] for node_id in sorted(compiled)}, bindings

api/orchestration/serverless_runtime.py

@@ -146,6 +146,10 @@ class ServerlessManager:
 
     async def start(self, execution_id, prompt_id, prepared_jobs):
         self.ensure_capacity(len(prepared_jobs))
+        from .serverless_typed_results import validate_prepared_job
+        # Validate the entire batch before creating even the first local slot.
+        for prepared in prepared_jobs:
+            validate_prepared_job(prepared)
         new_runs = []
         # Journal every local slot before starting any billed submission.
         for prepared in prepared_jobs:
@@ -177,14 +181,30 @@ class ServerlessManager:
         return any(run["worker"]["id"] == participant_id and
                    any(m["job_id"] == collector_job_id for m in run["manifest"]) for run in self.runs.values())
 
-    async def _deliver(self, run, images=None, error=None):
-        async with self.server.distributed_jobs_lock:
+    async def _deliver(self, run, images=None, error=None, *, media=None):
+        if media is not None and not error:
+            # Validate all destinations before exposing any partial success.
             for binding in run["manifest"]:
+                if binding["participant_id"] != run["worker"]["id"]:
+                    raise ValueError("Serverless media participant identity mismatch")
+                values = media[binding["job_id"]][binding["socket"]]
+                if not isinstance(values, list) or not values:
+                    raise ValueError("Missing serverless media values")
+        async with self.server.distributed_jobs_lock:
+            delivered = set()
+            for binding in run["manifest"]:
+                job_id = binding["job_id"]
+                if error or media is not None:
+                    if job_id in delivered:
+                        continue
+                    delivered.add(job_id)
                 queue = self.server.distributed_pending_jobs.get(binding["job_id"])
                 if queue is None:
                     continue  # A cancelled/finished collector must never be resurrected.
                 if error:
                     queue.put_nowait({"worker_id": run["worker"]["id"], "serverless_error": error, "is_last": True})
+                elif media is not None:
+                    queue.put_nowait({"worker_id": run["worker"]["id"], "media": media[job_id], "is_last": True})
                 else:
                     batch = images[binding["job_id"]]
                     for index, tensor in enumerate(batch):
@@ -218,13 +238,18 @@ class ServerlessManager:
         submitted = False
         provider_terminal = False
         try:
+            from .serverless_typed_results import validate_prepared_job, collect_typed_results
+            typed = validate_prepared_job(run)
             async with asyncio.timeout(max(0, run["deadline"] - time.monotonic())):
                 async with semaphore:
                     await self._reserve_provider_capacity(run)
                     client = await self._client(run["worker"])
                     run["client"] = client
                     submitted = True
-                    job_id = await client.submit(run["workflow"], run["images"], run["worker"]["timeout_seconds"])
+                    if typed:
+                        job_id = await client.submit(run["workflow"], run["images"], run["worker"]["timeout_seconds"], media=run["media"])
+                    else:
+                        job_id = await client.submit(run["workflow"], run["images"], run["worker"]["timeout_seconds"])
                     run["provider_job_id"] = job_id
                     # Persist immediately: all subsequent actions use this exact ID.
                     self.ledger.update(rid, job_id=job_id, status="IN_QUEUE", updated_at=time.time())
@@ -240,8 +265,12 @@ class ServerlessManager:
                             if not isinstance(output, dict) or output.get("error") or output.get("errors"):
                                 raise RunpodError("Serverless workflow reported an output error; inspect the provider job")
                             self.ledger.update(rid, status="DOWNLOADING", updated_at=time.time())
-                            images = await decode_results(output, run["manifest"])
-                            await self._deliver(run, images=images)
+                            if typed:
+                                media = await collect_typed_results(output, run["media"], run["manifest"])
+                                await self._deliver(run, media=media)
+                            else:
+                                images = await decode_results(output, run["manifest"])
+                                await self._deliver(run, images=images)
                             self.ledger.update(rid, status="COMPLETED", updated_at=time.time())
                             return
                         self.ledger.update(rid, status=status, updated_at=time.time())
@@ -276,6 +305,7 @@ class ServerlessManager:
             # base64 inputs, workflow metadata, or a client holding credentials.
             run.pop("workflow", None)
             run.pop("images", None)
+            run.pop("media", None)
             run["client"] = None
             # A released local semaphore alone cannot release a durable reservation.
             await self._notify_capacity()
@@ -359,6 +389,7 @@ class ServerlessManager:
                         self.ledger.update(rid, status="CANCELLED", updated_at=time.time())
                         run.pop("workflow", None)
                         run.pop("images", None)
+                        run.pop("media", None)
                         await self._deliver(run, error="Serverless job cancelled before submission")
                     continue
                 # Do not send two cancellation requests for the same live job in

api/orchestration/serverless_typed_inputs.py

@@ -0,0 +1,225 @@
+"""Bounded local native inputs for comfy-media-v1 requests.
+
+Only stock LoadAudio/LoadVideo schemas are adapted. Original nodes and links
+remain in the private request graph for endpoint validation; only the existing
+image preparer rewrites filenames. No upload, transcode, resample or provider
+operation belongs here. The caller still enforces the complete request-body cap.
+"""
+import hashlib
+import io
+import json
+from pathlib import Path
+
+from ...serverless_bridge import codec, contract, video_codec
+from . import serverless_media
+
+
+_LOADERS = {"LoadAudio": ("audio", "AUDIO"), "LoadVideo": ("file", "VIDEO")}
+Error = serverless_media.ServerlessMediaError
+
+
+def _loader_references(workflow):
+    if not isinstance(workflow, dict):
+        raise Error("Serverless workflow must be an API graph")
+    loaders = {}
+    for node_id, node in workflow.items():
+        if not isinstance(node, dict):
+            continue
+        class_type = node.get("class_type")
+        if not isinstance(class_type, str) or class_type not in _LOADERS:
+            continue
+        contract.identifier(node_id)
+        key, kind = _LOADERS[class_type]
+        inputs = node.get("inputs")
+        if not isinstance(inputs, dict) or set(inputs) != {key}:
+            raise Error(f"{class_type} inputs must contain only its native filename; extra controls are unsupported")
+        if not isinstance(inputs[key], str):
+            raise Error(f"{class_type} requires a literal filename; linked inputs are unsupported")
+        normalized, _suffix = serverless_media._normalize_media_reference(inputs[key])
+        loaders[node_id] = (kind, normalized)
+
+    references = []
+    used = set()
+    for node_id, node in workflow.items():
+        if not isinstance(node, dict) or not isinstance(node.get("inputs"), dict):
+            continue
+        for key, value in node["inputs"].items():
+            if not isinstance(value, (list, tuple)) or not value or not isinstance(value[0], str):
+                continue
+            if value[0] not in loaders:
+                continue
+            if type(value) is not list or len(value) != 2 or type(value[1]) is not int or value[1] != 0:
+                raise Error("Native media loader links require output slot 0")
+            contract.identifier(node_id)
+            contract.identifier(key)
+            references.append((node_id, key, value[0]))
+            used.add(value[0])
+    if used != set(loaders):
+        raise Error("Native media loader has no consumer")
+    return loaders, references
+
+
+def _source_path(value, folder_paths):
+    """Resolve native annotations; authorization/opening is owned by _read_path."""
+    annotation = serverless_media._MEDIA_ANNOTATION.search(value)
+    unannotated = value[:annotation.start()] if annotation else value
+    if Path(unannotated).is_absolute():
+        return unannotated
+    try:
+        return folder_paths.get_annotated_filepath(value)
+    except Exception:
+        raise Error("Native media input path is missing or invalid") from None
+
+
+def _decode_audio(data, limit):
+    """Match core nodes_audio.load, bounding each frame *before* accumulation.
+
+    Float32 output bytes are charged even for narrow integer PCM. Native float64
+    remains float64, as in f32_pcm (there is no silent dtype conversion). Raw
+    retained arrays and each live AV plane allocation have a separate <= limit
+    guard, so wider/padded raw frames cannot bypass the output budget. PyAV owns
+    the one frame being decoded; it is checked before to_ndarray or torch.cat.
+    Concatenation/conversion use bounded temporary copies, not an RSS guarantee.
+    """
+    if limit <= 0:
+        raise Error("AUDIO decoded input exceeds byte limit")
+    try:
+        import av
+        import torch
+        from comfy_extras.nodes_audio import f32_pcm
+    except ImportError:
+        raise Error("Native AUDIO decoding is unavailable") from None
+
+    try:
+        with av.open(io.BytesIO(data)) as container:
+            if not container.streams.audio:
+                raise Error("Native AUDIO input has no audio stream")
+            stream = container.streams.audio[0]
+            rate = stream.codec_context.sample_rate
+            channels = stream.channels
+            if type(rate) is not int or rate <= 0 or type(channels) is not int or channels <= 0:
+                raise Error("Native AUDIO stream shape or sample rate is invalid")
+            frames = []
+            output_bytes = raw_bytes = 0
+            dtype = None
+            for frame in container.decode(streams=stream.index):
+                samples = frame.samples
+                if (type(samples) is not int or samples <= 0 or frame.sample_rate != rate
+                        or len(frame.layout.channels) != channels):
+                    raise Error("Native AUDIO frame shape or sample rate changed")
+                if frame.format.name not in {"s16", "s16p", "s32", "s32p", "flt", "fltp", "dbl", "dblp"}:
+                    raise Error("Unsupported native AUDIO PCM format")
+                itemsize = frame.format.bytes
+                count = samples * channels
+                # Native f32_pcm promotes integer PCM to float32, not float64.
+                decoded_bytes = count * max(4, itemsize)
+                frame_bytes = count * itemsize
+                plane_bytes = sum(plane.buffer_size for plane in frame.planes)
+                if (decoded_bytes > limit - output_bytes or frame_bytes > limit - raw_bytes
+                        or plane_bytes > limit - raw_bytes):
+                    raise Error("AUDIO decoded input or raw frame exceeds byte limit")
+                output_bytes += decoded_bytes
+                raw_bytes += frame_bytes
+                array = frame.to_ndarray()
+                if array.ndim != 2 or array.size != count or array.nbytes != frame_bytes:
+                    raise Error("Native AUDIO frame shape is invalid")
+                buf = torch.from_numpy(array)
+                if buf.dtype not in (torch.int16, torch.int32, torch.float32, torch.float64):
+                    raise Error("Unsupported native AUDIO PCM dtype")
+                if dtype is not None and buf.dtype != dtype:
+                    raise Error("Native AUDIO frame dtype changed")
+                dtype = buf.dtype
+                if buf.shape[0] != channels:
+                    buf = buf.view(-1, channels).t()
+                if tuple(buf.shape) != (channels, samples):
+                    raise Error("Native AUDIO frame shape is invalid")
+                frames.append(buf)
+            if not frames:
+                raise Error("Native AUDIO input has no decoded frames")
+            waveform = f32_pcm(torch.cat(frames, dim=1)).unsqueeze(0)
+        return {"waveform": waveform, "sample_rate": rate}
+    except Error:
+        raise
+    except Exception:
+        raise Error("Native AUDIO input could not be decoded safely") from None
+
+
+def _artifact_id(kind, descriptor, payload):
+    digest = hashlib.sha256()
+    digest.update(kind.encode("ascii") + b"\0")
+    # Include interpretation (not just waveform bytes), notably sample_rate.
+    digest.update(json.dumps(descriptor, sort_keys=True, separators=(",", ":"), allow_nan=False).encode())
+    digest.update(b"\0")
+    digest.update(payload)
+    return "input-" + kind.lower() + "-" + digest.hexdigest()
+
+
+async def prepare_media_inputs(workflow: dict) -> tuple[list, list, list]:
+    """Return legacy images, inline typed artifacts, and every consumer reference.
+
+    Paths are read once into bounded bytes through the video adapter's no-follow
+    directory-fd reader. Decoders receive BytesIO, never a second path open.
+    Source files and unique decoded artifacts have independent aggregate caps.
+    Repeated content of the same kind is decoded once and charged once.
+    """
+    loaders, references = _loader_references(workflow)
+    image_view = {node_id: node for node_id, node in workflow.items() if node_id not in loaders}
+    # Retain the strict scan for every other loader before expensive decoding.
+    serverless_media._iter_image_references(image_view)
+    if not loaders:
+        return await serverless_media.prepare_images(image_view), [], []
+    try:
+        import folder_paths
+    except ImportError:
+        raise Error("ComfyUI media directories are unavailable") from None
+    roots = list(serverless_media._folder_roots(folder_paths).values())
+    artifacts = {}
+    source_artifacts = {}
+    source_digests = set()
+    filenames = {}
+    loader_artifacts = {}
+    file_total = decoded_total = 0
+    for node_id, (kind, filename) in loaders.items():
+        # Avoid retaining file buffers merely to deduplicate repeat references.
+        if (kind, filename) in filenames:
+            loader_artifacts[node_id] = filenames[kind, filename]
+            continue
+        path = _source_path(filename, folder_paths)
+        data = video_codec._read_path(path, contract.MAX_ARTIFACT_BYTES, roots)
+        source_digest = hashlib.sha256(data).hexdigest()
+        if source_digest not in source_digests:
+            if len(data) > contract.MAX_TOTAL_BYTES - file_total:
+                raise Error("Native media source files exceed total byte limit")
+            file_total += len(data)
+            source_digests.add(source_digest)
+        source_key = (kind, source_digest)
+        if source_key not in source_artifacts:
+            if len(artifacts) >= contract.MAX_ARTIFACTS:
+                raise Error("Native media input artifact count exceeds limit")
+            limit = min(contract.MAX_ARTIFACT_BYTES, contract.MAX_TOTAL_BYTES - decoded_total)
+            if kind == "AUDIO":
+                value = _decode_audio(data, limit)
+                descriptor, payload = codec.encode_payload(kind, value, max_bytes=limit)
+                del value
+            else:
+                from comfy_api.latest import InputImpl
+                value = InputImpl.VideoFromFile(io.BytesIO(data))
+                descriptor, payload = video_codec.encode_video(value, max_bytes=limit, allowed_roots=roots)
+                del value
+            aid = _artifact_id(kind, descriptor, payload)
+            if aid not in artifacts:
+                if len(payload) > limit:
+                    raise Error("Native media decoded inputs exceed total byte limit")
+                artifacts[aid] = contract.artifact_from_payload(
+                    aid, kind, descriptor, payload, max_bytes=contract.MAX_ARTIFACT_BYTES)
+                decoded_total += len(payload)
+            source_artifacts[source_key] = aid
+            del payload
+        aid = source_artifacts[source_key]
+        loader_artifacts[node_id] = filenames[kind, filename] = aid
+        del data
+    refs = [{"node_id": node_id, "input_name": key, "artifact_id": loader_artifacts[source]}
+            for node_id, key, source in references]
+    # Shared node dictionaries deliberately propagate only legacy image rewrites.
+    images = await serverless_media.prepare_images(image_view)
+    return images, list(artifacts.values()), refs

api/orchestration/serverless_typed_results.py

@@ -0,0 +1,79 @@
+"""Explicit media-v1 preflight and atomic, lossless result materialization."""
+from urllib.parse import urljoin
+
+import aiohttp
+
+from ...serverless_bridge import contract
+from ...serverless_bridge.contract import MAX_ARTIFACT_BYTES, PROFILE, validate_bindings
+from .serverless_media import (
+    MAX_REDIRECTS, _CHUNK_SIZE, _DOWNLOAD_TIMEOUT_SECONDS,
+    _PublicResolver, _validate_artifact_url,
+)
+
+
+def validate_prepared_job(prepared):
+    """Fail before journalling/submission; legacy jobs keep their original path."""
+    profile = prepared["worker"].get("media_profile", "legacy-image-v1")
+    media = prepared.get("media")
+    if profile == "legacy-image-v1" and media is None:
+        return False
+    if profile != PROFILE or media is None:
+        raise ValueError("Serverless media profile/request mismatch")
+    from .runpod_client import serialize_request
+    # The shared endpoint validator also checks exporter controls and input refs.
+    serialize_request(prepared["workflow"], prepared["images"],
+                      prepared["worker"]["timeout_seconds"], media=media)
+    expected = validate_bindings(prepared["manifest"])
+    if expected != validate_bindings(media["outputs"]):
+        raise ValueError("Serverless media manifest identity mismatch")
+    if any(binding["participant_id"] != prepared["worker"]["id"] for binding in expected.values()):
+        raise ValueError("Serverless media participant identity mismatch")
+    return True
+
+
+async def download_public_artifact(url, *, max_bytes=MAX_ARTIFACT_BYTES):
+    """Contract-sized public HTTPS download, never using a provider session/key."""
+    if type(max_bytes) is not int or not 0 <= max_bytes <= MAX_ARTIFACT_BYTES:
+        raise ValueError("Media artifact exceeds the permitted size")
+    current_url = _validate_artifact_url(url)
+    connector = aiohttp.TCPConnector(resolver=_PublicResolver(), use_dns_cache=False, limit=4)
+    timeout = aiohttp.ClientTimeout(total=_DOWNLOAD_TIMEOUT_SECONDS, connect=10, sock_read=15)
+    try:
+        async with aiohttp.ClientSession(connector=connector, timeout=timeout, trust_env=False,
+                                         cookie_jar=aiohttp.DummyCookieJar()) as session:
+            for redirect_count in range(MAX_REDIRECTS + 1):
+                async with session.get(current_url, allow_redirects=False) as response:
+                    if response.status in {301, 302, 303, 307, 308}:
+                        location = response.headers.get("Location")
+                        if not location or redirect_count >= MAX_REDIRECTS:
+                            raise ValueError("Media artifact redirect is invalid")
+                        current_url = _validate_artifact_url(urljoin(current_url, location))
+                        continue
+                    if response.status != 200:
+                        raise ValueError("Media artifact download failed")
+                    payload = bytearray()
+                    async for chunk in response.content.iter_chunked(_CHUNK_SIZE):
+                        if not isinstance(chunk, bytes) or len(payload) + len(chunk) > max_bytes:
+                            raise ValueError("Media artifact exceeds the permitted size")
+                        payload.extend(chunk)
+                    return bytes(payload)
+    except Exception:
+        # Signed URLs and transport exception details must not reach the ledger.
+        raise ValueError("Media artifact download failed or exceeded its byte limit") from None
+    raise ValueError("Media artifact download failed")
+
+
+async def collect_typed_results(output, media, manifest, *, download=None):
+    """Decode every binding before returning any collector's native value lists."""
+    if not isinstance(output, dict) or output.get("error") or output.get("errors"):
+        raise ValueError("Invalid serverless media output")
+    decoded = await contract.decode_response(output.get("media"), request_id=media["request_id"],
+                                    bindings=manifest,
+                                    download=download if download is not None else download_public_artifact)
+    grouped = {}
+    for binding in manifest:
+        values = decoded[binding["binding_id"]]
+        if not isinstance(values, list) or not values:
+            raise ValueError("Missing serverless media values")
+        grouped.setdefault(binding["job_id"], {}).setdefault(binding["socket"], []).extend(values)
+    return grouped

api/queue_orchestration.py

@@ -1,4 +1,5 @@
 import asyncio
+import copy
 import time
 import uuid
 
@@ -85,6 +86,10 @@ def _resolve_enabled_workers(config, requested_ids=None):
         if worker.get("type") == "serverless":
             from ..utils.serverless_config import validate_serverless_worker
             pool = validate_serverless_worker(worker)
+            if pool.get("media_profile") == "comfy-media-v1":
+                from ..serverless_bridge.contract import identifier
+                # Do not repair malformed media identities by stripping them.
+                identifier(pool["id"])
             for slot in range(pool["jobs_per_run"]):
                 workers.append(dict(pool, pool_id=worker_id, id=f"serverless:{worker_id}:{slot + 1}",
                                     name=f"{pool.get('name', worker_id)} / job {slot + 1}"))
@@ -147,13 +152,47 @@ def _is_load_balance_enabled(value):
 
 
 def _prompt_requests_load_balance(prompt_index):
-    for node_id in prompt_index.nodes_for_class("DistributedCollector"):
+    collector_ids = (prompt_index.nodes_for_class("DistributedCollector")
+                     + prompt_index.nodes_for_class("DistributedMediaCollector"))
+    for node_id in collector_ids:
         inputs = prompt_index.inputs_by_node.get(node_id, {})
         if _is_load_balance_enabled(inputs.get("load_balance", False)):
             return True
     return False
 
 
+def _validate_media_workers(prompt_index, workers):
+    """Reject incompatible selections before probing, pruning or dispatching.
+
+    Typed collectors currently have only a master/serverless transport. Check
+    all selected workers, not just reachable or load-balanced participants, so
+    a bad selection cannot silently turn into a different execution mode.
+    """
+    typed = bool(prompt_index.nodes_for_class("DistributedMediaCollector"))
+    if typed and (prompt_index.nodes_for_class("DistributedCollector")
+                  or prompt_index.nodes_for_class("UltimateSDUpscaleDistributed")):
+        raise ValueError(
+            "DistributedMediaCollector cannot be mixed with DistributedCollector or "
+            "UltimateSDUpscaleDistributed in one prompt"
+        )
+    for worker in workers:
+        if worker.get("type") != "serverless":
+            if typed:
+                raise ValueError(
+                    "DistributedMediaCollector does not support selected direct workers; "
+                    "use master execution or comfy-media-v1 serverless pools"
+                )
+            continue
+        profile = worker.get("media_profile", "legacy-image-v1")
+        if profile not in ("legacy-image-v1", "comfy-media-v1"):
+            raise ValueError("Unsupported serverless media_profile")
+        if typed != (profile == "comfy-media-v1"):
+            raise ValueError(
+                "DistributedMediaCollector requires comfy-media-v1 serverless pools; "
+                "legacy collectors require legacy-image-v1"
+            )
+
+
 async def _prepare_worker_payload(
     worker,
     prompt_index,
@@ -169,11 +208,34 @@ async def _prepare_worker_payload(
 ):
     """Prepare one worker prompt payload with bounded concurrency and media-sync timeout."""
     async with worker_prep_semaphore:
+        _validate_media_workers(prompt_index, [worker])
         worker_prompt = prompt_index.copy_prompt()
         if worker.get("type") == "serverless":
+            from .orchestration.runpod_client import serialize_request
+            if worker.get("media_profile", "legacy-image-v1") == "comfy-media-v1":
+                from ..serverless_bridge.contract import PROFILE, VERSION
+                from .orchestration.serverless_media_prompt import compile_media_prompt
+                from .orchestration.serverless_typed_inputs import prepare_media_inputs
+                # The legacy pruner only recognizes Collector/Upscale roots.
+                # The typed compiler validates the complete private graph and
+                # prunes it itself, retaining every connected typed branch.
+                worker_prompt = apply_participant_overrides(
+                    worker_prompt, worker["id"], enabled_ids, job_id_map,
+                    "", delegate_master, prompt_index,
+                )
+                request_id = str(uuid.uuid4())
+                workflow, manifest = compile_media_prompt(worker_prompt, request_id)
+                images, artifacts, refs = await prepare_media_inputs(workflow)
+                media = {
+                    "version": VERSION, "profile": PROFILE, "request_id": request_id,
+                    "artifacts": artifacts, "inputs": refs, "outputs": copy.deepcopy(manifest),
+                }
+                serialize_request(workflow, images, worker["timeout_seconds"], media=media)
+                return worker, {"worker": worker, "workflow": workflow, "manifest": manifest,
+                                "images": images, "media": media}
+
             from .orchestration.serverless_prompt import compile_serverless_prompt
             from .orchestration.serverless_media import prepare_images
-            from .orchestration.runpod_client import serialize_request
             worker_prompt = apply_participant_overrides(
                 prune_prompt_for_worker(worker_prompt), worker["id"], enabled_ids,
                 job_id_map, "", delegate_master, prompt_index,
@@ -266,21 +328,28 @@ async def orchestrate_distributed_execution(
     ) = _resolve_orchestration_limits(config)
     requested_ids = enabled_worker_ids if enabled_worker_ids is not None else None
     prompt_index = PromptIndex(prompt_obj)
+    typed_media_mode = bool(prompt_index.nodes_for_class("DistributedMediaCollector"))
+    if typed_media_mode:
+        from ..serverless_bridge.contract import identifier
+        # PromptIndex's JSON copy would otherwise turn integer keys into strings.
+        for node_id in prompt_obj:
+            identifier(node_id)
     image_list_start_ids = prompt_index.nodes_for_class("DistributedImageListStart")
     image_list_mode = bool(image_list_start_ids)
     if image_list_mode:
         # Validate before probing or queueing anything so malformed boundaries and
         # old worker releases fail before any upstream image is materialized.
-        validate_image_list_regions(prompt_obj)
         other_distributed = (
             prompt_index.nodes_for_class("DistributedCollector")
+            + prompt_index.nodes_for_class("DistributedMediaCollector")
             + prompt_index.nodes_for_class("UltimateSDUpscaleDistributed")
         )
         if other_distributed:
             raise ValueError(
-                "Distributed Image List v1 cannot be mixed with Distributed Collector or "
-                "Ultimate SD Upscale nodes in one prompt"
+                "Distributed Image List v1 cannot be mixed with Distributed Collector, "
+                "DistributedMediaCollector or Ultimate SD Upscale nodes in one prompt"
             )
+        validate_image_list_regions(prompt_obj)
         configured_serverless = [
             str(worker.get("id")) for worker in config.get("workers", [])
             if worker.get("enabled", False)
@@ -300,7 +369,14 @@ async def orchestrate_distributed_execution(
                 "serverless pools are unsupported"
             )
     else:
+        if typed_media_mode and requested_ids is not None:
+            configured_ids = {worker.get("id") for worker in config.get("workers", [])}
+            if any(worker_id not in configured_ids for worker_id in requested_ids):
+                raise ValueError("Unknown explicitly selected media worker ID")
         workers = _resolve_enabled_workers(config, requested_ids)
+        _validate_media_workers(prompt_index, workers)
+        if typed_media_mode and delegate_master and not workers:
+            raise RuntimeError("Delegate-only DistributedMediaCollector requires a selected comfy-media-v1 pool")
     load_balance_requested = _prompt_requests_load_balance(prompt_index)
     trace_debug(
         execution_trace_id,
@@ -316,8 +392,8 @@ async def orchestrate_distributed_execution(
         ),
     )
 
-    # Keep the legacy Collector fallback, but never override image-list
-    # orchestrator-only mode merely because its direct workers are unavailable.
+    # Keep the legacy Collector fallback. Image-list and typed-media delegate
+    # modes must not silently enable master participation.
     if not image_list_mode and not workers and delegate_master:
         trace_debug(
             execution_trace_id,
@@ -332,6 +408,8 @@ async def orchestrate_distributed_execution(
         trace_execution_id=execution_trace_id,
         probe_concurrency=worker_probe_concurrency,
     )
+    if typed_media_mode and delegate_master and not active_workers:
+        raise RuntimeError("No selected comfy-media-v1 worker is reachable for delegate-only DistributedMediaCollector")
     if image_list_mode:
         from .orchestration.dispatch import preflight_image_list_workers
         if not active_workers and delegate_master:
@@ -430,7 +508,8 @@ async def orchestrate_distributed_execution(
     )
 
     if delegate_master and not image_list_mode:
-        collector_ids = find_nodes_by_class(master_prompt, "DistributedCollector")
+        collector_ids = (find_nodes_by_class(master_prompt, "DistributedCollector")
+                         + find_nodes_by_class(master_prompt, "DistributedMediaCollector"))
         upscale_nodes = find_nodes_by_class(master_prompt, "UltimateSDUpscaleDistributed")
         if upscale_nodes:
             debug_log(
@@ -477,6 +556,15 @@ async def orchestrate_distributed_execution(
     manager = None
     if serverless_jobs:
         from .orchestration.serverless_runtime import get_manager
+        # Repeat the manager's whole-batch identity/profile preflight *before*
+        # any direct dispatch, master queueing or collector queue allocation.
+        # Pure legacy jobs are a no-op in that validator; keep their path free
+        # of typed imports (and compatible with existing legacy media adapters).
+        for prepared in serverless_jobs:
+            if (prepared["worker"].get("media_profile", "legacy-image-v1") != "legacy-image-v1"
+                    or prepared.get("media") is not None):
+                from .orchestration.serverless_typed_results import validate_prepared_job
+                validate_prepared_job(prepared)
         manager = get_manager()
         manager.ensure_capacity(len(serverless_jobs))
 
@@ -522,8 +610,10 @@ async def orchestrate_distributed_execution(
             await manager.cancel(execution_id=execution_trace_id)
             async with prompt_server.distributed_jobs_lock:
                 for prepared in serverless_jobs:
-                    for binding in prepared["manifest"]:
-                        q = prompt_server.distributed_pending_jobs.get(binding["job_id"])
+                    # Typed manifests have several sockets per collector, but
+                    # each participant gets only one terminal collector event.
+                    for job_id in {binding["job_id"] for binding in prepared["manifest"]}:
+                        q = prompt_server.distributed_pending_jobs.get(job_id)
                         if q is not None:
                             q.put_nowait({"worker_id": prepared["worker"]["id"], "is_last": True,
                                           "serverless_error": "Serverless job setup failed; no automatic retry was attempted"})