feat: add Runpod serverless image collector pools

api/__init__.py

@@ -1,4 +1,5 @@
 from . import config_routes  # noqa: F401
+from . import serverless_routes  # noqa: F401
 from . import tunnel_routes  # noqa: F401
 from . import worker_routes  # noqa: F401
 from . import job_routes  # noqa: F401

api/config_routes.py

@@ -1,5 +1,6 @@
 import json
 from contextlib import asynccontextmanager
+from typing import Any
 
 from aiohttp import web
 import server
@@ -24,6 +25,7 @@ except ImportError:
             save_config(config)
 from ..utils.logging import debug_log, log
 from ..utils.network import handle_api_error, normalize_host
+from ..utils.serverless_config import public_config, validate_serverless_worker
 
 
 def _positive_int(value):
@@ -86,22 +88,188 @@ def _apply_field_patch(target: dict, data: dict, field_rules: list) -> None:
             target[key] = normalizer(value) if (normalizer and value is not None) else value
 
 
+_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",
+}
+_SERVERLESS_CONFIG_FIELDS = {
+    "worker_id", "type", "provider", "endpoint_id", "jobs_per_run", "concurrency",
+    "timeout_seconds", "enabled", "name", "api_key", "clear_api_key",
+}
+
+
+class _ActivePoolCheckError(RuntimeError):
+    pass
+
+
+def _error_response(message, status):
+    return web.json_response({"status": "error", "message": message}, status=status)
+
+
+def _is_secret_field_name(key):
+    normalized = "".join(character for character in key.casefold() if character.isalnum())
+    return normalized in {"apikey", "clearapikey"} or "credential" in normalized
+
+
+def _contains_secret_fields(value, *, allow_root_serverless_secrets=False, depth=0):
+    if isinstance(value, dict):
+        for key, item in value.items():
+            field_name = key.casefold() if isinstance(key, str) else ""
+            if field_name and _is_secret_field_name(field_name):
+                if not (allow_root_serverless_secrets and depth == 0 and key in _SECRET_FIELDS):
+                    return True
+            if _contains_secret_fields(
+                item, allow_root_serverless_secrets=allow_root_serverless_secrets, depth=depth + 1
+            ):
+                return True
+    elif isinstance(value, (list, tuple)):
+        return any(
+            _contains_secret_fields(
+                item, allow_root_serverless_secrets=allow_root_serverless_secrets, depth=depth + 1
+            )
+            for item in value
+        )
+    return False
+
+
+def _serverless_secret_store():
+    from .orchestration.serverless_runtime import get_secret_store
+    return get_secret_store()
+
+
+def _serverless_pool_is_active(worker_id):
+    from .orchestration.serverless_runtime import get_manager
+    manager = get_manager()
+    tasks = getattr(manager, "tasks", {})
+    runs = getattr(manager, "runs", {})
+    for run_id, task in tasks.items():
+        if task.done():
+            continue
+        run = runs.get(run_id)
+        worker = run.get("worker") if isinstance(run, dict) else None
+        if not isinstance(worker, dict):
+            continue
+        pool_id = worker.get("pool_id", worker.get("id"))
+        if str(pool_id) == str(worker_id):
+            return True
+    return False
+
+
+def _ensure_pool_idle(worker_id):
+    try:
+        active = _serverless_pool_is_active(worker_id)
+    except Exception:
+        raise _ActivePoolCheckError from None
+    if active:
+        raise _ActivePoolCheckError
+
+
+def _worker_by_id(workers, worker_id):
+    for worker in workers:
+        if isinstance(worker, dict) and worker.get("id") == worker_id:
+            return worker
+    return None
+
+
+def _worker_type(worker):
+    return worker.get("type", "local") if isinstance(worker, dict) else None
+
+
+def _strip_sensitive_fields(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {
+            key: _strip_sensitive_fields(item)
+            for key, item in value.items()
+            if not (isinstance(key, str) and _is_secret_field_name(key))
+        }
+    if isinstance(value, list):
+        return [_strip_sensitive_fields(item) for item in value]
+    return value
+
+
+def _clean_serverless_worker(worker):
+    clean = _strip_sensitive_fields(dict(worker))
+    for field in _SERVERLESS_LOCAL_FIELDS | {"api_key_configured"}:
+        clean.pop(field, None)
+    return clean
+
+
+def _validate_bulk_workers(workers):
+    validated = []
+    for worker in workers:
+        if isinstance(worker, dict) and worker.get("type") == "serverless":
+            clean = _clean_serverless_worker(worker)
+            validated.append(validate_serverless_worker(clean))
+        else:
+            validated.append(worker)
+    return validated
+
+
+def _clean_bulk_type_conversions(old_workers, new_workers):
+    old_by_id = {str(w.get("id")): w for w in old_workers if isinstance(w, dict) and w.get("id") is not None}
+    for index, worker in enumerate(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":
+            continue
+        converted = dict(worker)
+        for field in _SERVERLESS_ONLY_FIELDS:
+            converted.pop(field, None)
+        new_workers[index] = converted
+    return new_workers
+
+
+def _bulk_serverless_changes(old_workers, new_workers):
+    old_by_id = {str(w.get("id")): w for w in old_workers if isinstance(w, dict) and w.get("id") is not None}
+    new_by_id = {str(w.get("id")): w for w in new_workers if isinstance(w, dict) and w.get("id") is not None}
+    secret_ids = set()
+    for worker_id, old in old_by_id.items():
+        new = new_by_id.get(worker_id)
+        if old.get("type") == "serverless":
+            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")
+                  or old.get("concurrency", 1) != new.get("concurrency", 1)):
+                _ensure_pool_idle(worker_id)
+        elif new is not None and new.get("type") == "serverless":
+            secret_ids.add(worker_id)
+    return secret_ids
+
+
+def _restore_secret(store, worker_id, prior_secret):
+    try:
+        if prior_secret is None:
+            store.delete(worker_id)
+        else:
+            store.set(worker_id, prior_secret)
+    except Exception:
+        pass
+
+
 @server.PromptServer.instance.routes.get("/distributed/config")
 async def get_config_endpoint(request):
-    config = load_config()
-    return web.json_response(config)
+    try:
+        config = load_config()
+        return web.json_response(public_config(config, _serverless_secret_store()))
+    except Exception:
+        return _error_response("Unable to read config", 500)
 
 
 @server.PromptServer.instance.routes.post("/distributed/config")
 async def update_config_endpoint(request):
-    """Bulk config update with schema validation."""
+    """Bulk config update with schema validation and credential redaction."""
     try:
         data = await request.json()
-    except Exception as e:
-        return await handle_api_error(request, f"Invalid JSON payload: {e}", 400)
+    except Exception:
+        return _error_response("Invalid JSON payload", 400)
 
     if not isinstance(data, dict):
-        return await handle_api_error(request, "Config payload must be an object", 400)
+        return _error_response("Config payload must be an object", 400)
+    if _contains_secret_fields(data):
+        return _error_response("API keys cannot be set through bulk config", 400)
 
     validated_settings = {}
     validated_root = {}
@@ -118,9 +286,15 @@ async def update_config_endpoint(request):
             continue
 
         if validator and not validator(value):
-            errors.append(f"{key}: value {value!r} failed validation")
+            errors.append(f"{key}: failed validation")
             continue
 
+        if key == "workers":
+            try:
+                value = _validate_bulk_workers(value)
+            except (TypeError, ValueError):
+                return _error_response("Invalid serverless worker configuration", 400)
+
         if key in _SETTINGS_FIELDS:
             validated_settings[key] = value
         else:
@@ -133,15 +307,45 @@ async def update_config_endpoint(request):
             "message": "; ".join(errors),
         }, status=400)
 
+    store = None
+    changed_secrets = []
+    secret_changes_applied = False
     try:
         async with config_transaction() as config:
+            if "workers" in validated_root:
+                old_workers = config.get("workers", [])
+                validated_root["workers"] = _clean_bulk_type_conversions(
+                    old_workers, validated_root["workers"]
+                )
+                changed_ids = _bulk_serverless_changes(old_workers, validated_root["workers"])
+                for worker in validated_root["workers"]:
+                    if isinstance(worker, dict) and worker.get("type") == "serverless" and worker.get("enabled", False):
+                        store = store or _serverless_secret_store()
+                        if worker["id"] in changed_ids or not store.has(worker["id"]):
+                            return _error_response("Save an API key before enabling this pool", 400)
+                if changed_ids:
+                    store = store or _serverless_secret_store()
+                    for worker_id in sorted(changed_ids):
+                        prior_secret = store.get(worker_id)
+                        changed_secrets.append((worker_id, prior_secret))
+                        if prior_secret is not None:
+                            store.delete(worker_id)
+                            secret_changes_applied = True
+
             settings = config.setdefault("settings", {})
             settings.update(validated_settings)
             for key, value in validated_root.items():
                 config[key] = value
-            return web.json_response({"status": "success", "config": config})
-    except Exception as e:
-        return await handle_api_error(request, e)
+            response_config = public_config(config, store or _serverless_secret_store())
+
+        return web.json_response({"status": "success", "config": response_config})
+    except _ActivePoolCheckError:
+        return _error_response("Wait for active serverless jobs before changing this pool", 409)
+    except Exception:
+        if secret_changes_applied and store is not None:
+            for worker_id, prior_secret in changed_secrets:
+                _restore_secret(store, worker_id, prior_secret)
+        return _error_response("Unable to update config", 500)
 
 
 @server.PromptServer.instance.routes.get("/distributed/queue_status/{job_id}")
@@ -164,79 +368,205 @@ async def queue_status_endpoint(request):
 
 @server.PromptServer.instance.routes.post("/distributed/config/update_worker")
 async def update_worker_endpoint(request):
+    store = None
+    prior_secret = None
+    secret_changed = False
+    worker_id = None
     try:
-        data = await request.json()
+        try:
+            data = await request.json()
+        except Exception:
+            return _error_response("Invalid JSON payload", 400)
+        if not isinstance(data, dict):
+            return _error_response("Worker payload must be an object", 400)
+
         worker_id = data.get("worker_id")
-        
         if worker_id is None:
-            return await handle_api_error(request, "Missing worker_id", 400)
+            return _error_response("Missing worker_id", 400)
 
+        if _contains_secret_fields(data, allow_root_serverless_secrets=True):
+            return _error_response("Invalid worker credentials", 400)
+
+        result = None
         async with config_transaction() as config:
-            worker_found = False
             workers = config.setdefault("workers", [])
+            existing = _worker_by_id(workers, worker_id)
+            current_type = _worker_type(existing)
+            target_type = data.get("type", current_type or "local")
 
-            for worker in workers:
-                if worker["id"] == worker_id:
-                    _apply_field_patch(worker, data, _WORKER_FIELDS)
-                    worker_found = True
-                    break
+            if target_type == "serverless":
+                unknown = set(data) - _SERVERLESS_CONFIG_FIELDS - _SERVERLESS_LOCAL_FIELDS
+                if unknown:
+                    return _error_response("Unknown serverless worker field", 400)
+                if not isinstance(worker_id, str) or not worker_id.strip():
+                    return _error_response("worker_id must be a non-empty string", 400)
+                if "api_key" in data and not isinstance(data["api_key"], str):
+                    return _error_response("Invalid worker credentials", 400)
+                if "clear_api_key" in data and type(data["clear_api_key"]) is not bool:
+                    return _error_response("Invalid worker credentials", 400)
+                supplied_key = data.get("api_key", "")
+                clear_key = data.get("clear_api_key", False)
+                if supplied_key and clear_key:
+                    return _error_response("Choose either api_key or clear_api_key", 400)
+                if "enabled" in data and type(data["enabled"]) is not bool:
+                    return _error_response("enabled must be a boolean", 400)
+                if "name" in data and not isinstance(data["name"], str):
+                    return _error_response("name must be a string", 400)
 
-            if not worker_found:
-                # If worker not found and all required fields are provided, create new worker
-                if all(key in data for key in ["name", "port", "cuda_device"]):
-                    new_worker = {
-                        "id": worker_id,
-                        "name": data["name"],
-                        "host": normalize_host(data.get("host", "localhost")),
-                        "port": data["port"],
-                        "cuda_device": data["cuda_device"],
-                        "enabled": data.get("enabled", False),
-                        "extra_args": data.get("extra_args", ""),
-                        "type": data.get("type", "local")
-                    }
-                    workers.append(new_worker)
+                candidate = dict(existing or {})
+                candidate["id"] = worker_id
+                candidate["type"] = "serverless"
+                candidate.setdefault("provider", "runpod")
+                candidate.setdefault("enabled", False)
+                for field in ("provider", "endpoint_id", "jobs_per_run", "concurrency", "timeout_seconds", "enabled", "name"):
+                    if field in data:
+                        candidate[field] = data[field]
+                candidate = _clean_serverless_worker(candidate)
+                validated = validate_serverless_worker(candidate)
+                if "enabled" in validated and type(validated["enabled"]) is not bool:
+                    return _error_response("enabled must be a boolean", 400)
+                if "name" in validated and not isinstance(validated["name"], str):
+                    return _error_response("name must be a string", 400)
+
+                store = _serverless_secret_store()
+                prior_secret = store.get(worker_id)
+                converting_to_serverless = current_type != "serverless"
+                next_secret = None if converting_to_serverless else prior_secret
+                if supplied_key:
+                    next_secret = supplied_key
+                elif clear_key:
+                    next_secret = None
+
+                endpoint_changed = (
+                    current_type == "serverless"
+                    and existing.get("endpoint_id") != validated.get("endpoint_id")
+                )
+                concurrency_changed = (
+                    current_type == "serverless"
+                    and existing.get("concurrency", 1) != validated.get("concurrency", 1)
+                )
+                if current_type == "serverless" and (
+                    endpoint_changed or concurrency_changed or next_secret != prior_secret
+                ):
+                    _ensure_pool_idle(worker_id)
+
+                if validated.get("enabled", False) and next_secret is None:
+                    return _error_response("Save an API key before enabling this pool; disable it before clearing the key", 400)
+
+                if next_secret != prior_secret:
+                    if next_secret is None:
+                        store.delete(worker_id)
+                    else:
+                        store.set(worker_id, next_secret)
+                    secret_changed = True
+
+                if existing is None:
+                    workers.append(validated)
                 else:
-                    return await handle_api_error(
-                        request,
-                        f"Worker {worker_id} not found and missing required fields for creation",
-                        404,
-                    )
+                    existing.clear()
+                    existing.update(validated)
+                result = {"status": "success", "api_key_configured": next_secret is not None}
+
+            elif existing is not None:
+                cleaned_existing = _strip_sensitive_fields(existing)
+                existing.clear()
+                existing.update(cleaned_existing)
+                if current_type == "serverless":
+                    if _contains_secret_fields(data):
+                        return _error_response("Invalid worker credentials", 400)
+                    _ensure_pool_idle(worker_id)
+                    store = _serverless_secret_store()
+                    prior_secret = store.get(worker_id)
+                    if prior_secret is not None:
+                        store.delete(worker_id)
+                        secret_changed = True
+                    for field in _SERVERLESS_ONLY_FIELDS:
+                        existing.pop(field, None)
+                elif _contains_secret_fields(data):
+                    return _error_response("API keys are only accepted for serverless workers", 400)
+                _apply_field_patch(existing, data, _WORKER_FIELDS)
+                result = {"status": "success"}
+            elif all(key in data for key in ["name", "port", "cuda_device"]):
+                if _contains_secret_fields(data):
+                    return _error_response("API keys are only accepted for serverless workers", 400)
+                new_worker = {
+                    "id": worker_id,
+                    "name": data["name"],
+                    "host": normalize_host(data.get("host", "localhost")),
+                    "port": data["port"],
+                    "cuda_device": data["cuda_device"],
+                    "enabled": data.get("enabled", False),
+                    "extra_args": data.get("extra_args", ""),
+                    "type": data.get("type", "local")
+                }
+                workers.append(new_worker)
+                result = {"status": "success"}
+            else:
+                return _error_response(
+                    "Worker not found and missing required fields for creation", 404
+                )
+
+        return web.json_response(result)
+    except _ActivePoolCheckError:
+        if secret_changed and store is not None:
+            _restore_secret(store, worker_id, prior_secret)
+        return _error_response("Wait for active serverless jobs before changing this pool", 409)
+    except ValueError:
+        if secret_changed and store is not None:
+            _restore_secret(store, worker_id, prior_secret)
+        return _error_response("Invalid worker configuration", 400)
+    except Exception:
+        if secret_changed and store is not None:
+            _restore_secret(store, worker_id, prior_secret)
+        return _error_response("Unable to update worker configuration", 500)
 
-            return web.json_response({"status": "success"})
-    except Exception as e:
-        return await handle_api_error(request, e, 400)
 
 @server.PromptServer.instance.routes.post("/distributed/config/delete_worker")
 async def delete_worker_endpoint(request):
+    store = None
+    prior_secret = None
+    secret_changed = False
+    worker_id = None
     try:
-        data = await request.json()
+        try:
+            data = await request.json()
+        except Exception:
+            return _error_response("Invalid JSON payload", 400)
+        if not isinstance(data, dict):
+            return _error_response("Worker payload must be an object", 400)
         worker_id = data.get("worker_id")
-        
         if worker_id is None:
-            return await handle_api_error(request, "Missing worker_id", 400)
-            
+            return _error_response("Missing worker_id", 400)
+
+        result = None
         async with config_transaction() as config:
             workers = config.get("workers", [])
-
-            # Find and remove the worker
-            worker_index = -1
-            for i, worker in enumerate(workers):
-                if worker["id"] == worker_id:
-                    worker_index = i
-                    break
-
-            if worker_index == -1:
-                return await handle_api_error(request, f"Worker {worker_id} not found", 404)
-
-            # Remove the worker
-            removed_worker = workers.pop(worker_index)
-
-            return web.json_response({
+            removed = _worker_by_id(workers, worker_id)
+            if removed is None:
+                return _error_response("Worker not found", 404)
+            if _worker_type(removed) == "serverless":
+                _ensure_pool_idle(worker_id)
+                store = _serverless_secret_store()
+                prior_secret = store.get(str(worker_id))
+                if prior_secret is not None:
+                    store.delete(str(worker_id))
+                    secret_changed = True
+            workers.remove(removed)
+            result = {
                 "status": "success",
-                "message": f"Worker {removed_worker.get('name', worker_id)} deleted"
-            })
-    except Exception as e:
-        return await handle_api_error(request, e, 400)
+                "message": "Worker deleted" if _worker_type(removed) == "serverless"
+                else f"Worker {removed.get('name', worker_id)} deleted",
+            }
+
+        return web.json_response(result)
+    except _ActivePoolCheckError:
+        if secret_changed and store is not None:
+            _restore_secret(store, str(worker_id), prior_secret)
+        return _error_response("Wait for active serverless jobs before changing this pool", 409)
+    except Exception:
+        if secret_changed and store is not None:
+            _restore_secret(store, str(worker_id), prior_secret)
+        return _error_response("Unable to delete worker", 500)
 
 @server.PromptServer.instance.routes.post("/distributed/config/update_setting")
 async def update_setting_endpoint(request):

api/orchestration/dispatch.py

@@ -55,6 +55,11 @@ async def worker_ws_is_active(worker):
 
 async def _probe_worker_active(worker, use_websocket, semaphore):
     async with semaphore:
+        if worker.get("type") == "serverless":
+            from .serverless_runtime import get_manager
+            # Authentication/configuration failures must surface, not silently skip paid work.
+            status = await get_manager().health(worker)
+            return worker, status["ok"]
         is_active = await (worker_ws_is_active(worker) if use_websocket else worker_is_active(worker))
         return worker, is_active
 
@@ -203,6 +208,10 @@ def _extract_queue_remaining(payload):
 
 async def _probe_worker_queue(worker, semaphore, probe_timeout):
     async with semaphore:
+        if worker.get("type") == "serverless":
+            from .serverless_runtime import get_manager
+            status = await get_manager().health(worker)
+            return {"worker": worker, "queue_remaining": status["queueRemaining"]}
         worker_url = build_worker_url(worker)
         payload = await probe_worker(worker_url, timeout=probe_timeout)
         if payload is None:

api/orchestration/runpod_client.py

@@ -0,0 +1,97 @@
+"""Small Runpod queue API client. Never retries a potentially billable POST."""
+import asyncio
+import json
+import re
+
+import aiohttp
+
+API_BASE = "https://api.runpod.ai/v2"
+MAX_REQUEST_BYTES = 10 * 1024 * 1024
+_ID = re.compile(r"[A-Za-z0-9_-]{1,200}\Z")
+
+
+class RunpodError(RuntimeError):
+    """Sanitized provider failure safe to display in the panel."""
+
+
+class AmbiguousSubmission(RunpodError):
+    """Submission may have been accepted: never retry it automatically."""
+
+
+def checked_id(value):
+    if not isinstance(value, str) or not _ID.fullmatch(value):
+        raise ValueError("Invalid Runpod endpoint or job ID")
+    return value
+
+
+class RunpodClient:
+    def __init__(self, session, endpoint_id, api_key):
+        self.session = session
+        self.endpoint_id = checked_id(endpoint_id)
+        if not isinstance(api_key, str) or not api_key or any(c in api_key for c in "\r\n"):
+            raise ValueError("A valid Runpod API key is required")
+        self._api_key = api_key
+
+    async def _request(self, method, path, payload=None):
+        # Base origin is code-owned, never a user-supplied URL. Do not follow redirects
+        # with credentials, or include response bodies/transport errors in exceptions.
+        url = f"{API_BASE}/{self.endpoint_id}/{path}"
+        attempts = 3 if method == "GET" else 1
+        for attempt in range(attempts):
+            try:
+                async with self.session.request(
+                    method, url,
+                    headers={"Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json"},
+                    json=payload,
+                    timeout=aiohttp.ClientTimeout(total=30),
+                    allow_redirects=False,
+                ) as response:
+                    status = response.status
+                    if status != 200:
+                        if method == "GET" and (status == 429 or status >= 500) and attempt + 1 < attempts:
+                            await asyncio.sleep(0.5 * (2 ** attempt))
+                            continue
+                        if method == "POST" and path == "run" and status >= 500:
+                            raise AmbiguousSubmission("Runpod submission returned a server error; acceptance is unknown. Check the provider before retrying.")
+                        raise RunpodError(f"Runpod {path.split('/')[0]} request failed (HTTP {status})")
+                    try:
+                        result = await response.json()
+                    except (ValueError, aiohttp.ContentTypeError):
+                        if method == "POST" and path == "run":
+                            raise AmbiguousSubmission("Runpod submission returned an unreadable response; check the provider before retrying.") from None
+                        raise RunpodError("Runpod returned an invalid JSON response") from None
+                    if not isinstance(result, dict):
+                        if method == "POST" and path == "run":
+                            raise AmbiguousSubmission("Runpod submission did not return a job record; check the provider before retrying.")
+                        raise RunpodError("Runpod returned an invalid response")
+                    return result
+            except (asyncio.TimeoutError, aiohttp.ClientError, OSError):
+                if method == "POST" and path == "run":
+                    raise AmbiguousSubmission("Runpod submission connection was interrupted; acceptance is unknown. Check the provider before retrying.") from None
+                if method == "GET" and attempt + 1 < attempts:
+                    await asyncio.sleep(0.5 * (2 ** attempt))
+                    continue
+                raise RunpodError(f"Runpod {path.split('/')[0]} connection failed") from None
+        raise RunpodError("Runpod request failed")
+
+    async def health(self):
+        return await self._request("GET", "health")
+
+    async def submit(self, workflow, images, timeout_seconds):
+        payload = {
+            "input": {"workflow": workflow, "images": images},
+            "policy": {"executionTimeout": timeout_seconds * 1000, "ttl": timeout_seconds * 1000},
+        }
+        if len(json.dumps(payload, allow_nan=False).encode("utf-8")) > MAX_REQUEST_BYTES:
+            raise ValueError("Serverless request exceeds the 10 MiB safety limit; reduce input media size")
+        result = await self._request("POST", "run", payload)
+        try:
+            return checked_id(result.get("id"))
+        except ValueError:
+            raise AmbiguousSubmission("Runpod submission returned no valid job ID; check the provider before retrying.") from None
+
+    async def status(self, job_id):
+        return await self._request("GET", f"status/{checked_id(job_id)}")
+
+    async def cancel(self, job_id):
+        return await self._request("POST", f"cancel/{checked_id(job_id)}")

api/orchestration/serverless_ledger.py

@@ -0,0 +1,57 @@
+"""Private, atomic job-ID journal. Contains identifiers/status only, never payloads."""
+import json
+import os
+from pathlib import Path
+import tempfile
+import threading
+
+TERMINAL = {"COMPLETED", "FAILED", "CANCELLED", "TIMED_OUT"}
+FIELDS = {"record_id", "execution_id", "worker_id", "participant_id", "endpoint_id", "job_id", "status", "created_at", "updated_at", "error", "prompt_id"}
+
+
+class JobLedger:
+    def __init__(self, path):
+        self.path = Path(path)
+        self._lock = threading.RLock()
+
+    def _read(self):
+        if self.path.is_symlink():
+            raise RuntimeError("Unsafe serverless job journal path")
+        if not self.path.exists():
+            return {}
+        try:
+            data = json.loads(self.path.read_text(encoding="utf-8"))
+            if not isinstance(data, dict) or any(not isinstance(v, dict) or set(v) - FIELDS for v in data.values()):
+                raise ValueError()
+            return data
+        except (ValueError, OSError):
+            raise RuntimeError("Serverless job journal is unreadable; preserve it for recovery before submitting more work") from None
+
+    def all(self):
+        with self._lock:
+            return list(self._read().values())
+
+    def update(self, record_id, **fields):
+        if set(fields) - FIELDS:
+            raise ValueError("Unexpected serverless journal field")
+        with self._lock:
+            data = self._read()
+            data.setdefault(record_id, {"record_id": record_id}).update(fields)
+            # Keep every unresolved job, plus a bounded terminal history.
+            completed = sorted((k for k, v in data.items() if v.get("status") in TERMINAL), key=lambda k: data[k].get("updated_at", 0))
+            for key in completed[:-200]:
+                data.pop(key)
+            if self.path.parent.is_symlink():
+                raise RuntimeError("Unsafe serverless journal directory")
+            self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            fd, temporary = tempfile.mkstemp(prefix=".jobs-", dir=self.path.parent)
+            try:
+                with os.fdopen(fd, "w", encoding="utf-8") as stream:
+                    json.dump(data, stream, allow_nan=False)
+                    stream.flush()
+                    os.fsync(stream.fileno())
+                os.replace(temporary, self.path)
+            finally:
+                if os.path.exists(temporary):
+                    os.unlink(temporary)
+            return dict(data[record_id])

api/orchestration/serverless_media.py

@@ -0,0 +1,571 @@
+"""Bounded image transport for generic Runpod-backed ComfyUI jobs."""
+
+import asyncio
+import base64
+import binascii
+import hashlib
+import io
+import ipaddress
+import os
+import re
+import socket
+import stat
+import warnings
+from pathlib import Path, PurePosixPath
+from urllib.parse import urljoin, urlsplit
+
+import aiohttp
+import numpy as np
+from PIL import Image, UnidentifiedImageError
+
+
+MAX_INPUT_IMAGE_BYTES = 25 * 1024 * 1024
+MAX_INPUT_TOTAL_BYTES = 7 * 1024 * 1024
+MAX_OUTPUT_FILES = 1024
+MAX_OUTPUT_IMAGE_BYTES = 128 * 1024 * 1024
+MAX_OUTPUT_TOTAL_BYTES = 256 * 1024 * 1024
+MAX_IMAGE_PIXELS = 40_000_000
+MAX_OUTPUT_ARRAY_BYTES = 256 * 1024 * 1024
+MAX_REDIRECTS = 3
+_DOWNLOAD_TIMEOUT_SECONDS = 60
+_CHUNK_SIZE = 64 * 1024
+
+_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp"}
+_UNSUPPORTED_IMAGE_EXTENSIONS = {".gif", ".tif", ".tiff", ".avif", ".heic", ".ico", ".svg"}
+_AUDIO_EXTENSIONS = {".wav", ".mp3", ".flac", ".m4a", ".aac", ".ogg", ".opus", ".aiff", ".aif", ".wma"}
+_VIDEO_EXTENSIONS = {".mp4", ".avi", ".mov", ".mkv", ".webm"}
+_MEDIA_EXTENSIONS = _IMAGE_EXTENSIONS | _UNSUPPORTED_IMAGE_EXTENSIONS | _AUDIO_EXTENSIONS | _VIDEO_EXTENSIONS
+_MEDIA_ANNOTATION = re.compile(r"\s+\[([A-Za-z0-9_-]{1,64})\]$")
+_PREFIX_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,199}\Z", re.ASCII)
+_SAFE_FILENAME_RE = re.compile(r"[A-Za-z0-9_.-]{1,255}\Z", re.ASCII)
+
+
+class ServerlessMediaError(ValueError):
+    """Sanitized input, artifact, or download validation failure."""
+
+
+def _folder_roots(folder_paths):
+    roots = {}
+    for kind, getter_name in (
+        ("input", "get_input_directory"),
+        ("output", "get_output_directory"),
+        ("temp", "get_temp_directory"),
+    ):
+        getter = getattr(folder_paths, getter_name, None)
+        if not callable(getter):
+            raise ServerlessMediaError("ComfyUI media directories are unavailable")
+        try:
+            roots[kind] = Path(getter()).resolve(strict=True)
+        except Exception:
+            raise ServerlessMediaError("ComfyUI media directories are unavailable") from None
+    return roots
+
+
+def _inside_roots(path, roots):
+    for root in roots.values():
+        try:
+            path.relative_to(root)
+            return True
+        except ValueError:
+            continue
+    return False
+
+
+def _normalize_media_reference(value):
+    if not isinstance(value, str) or not value or value != value.strip() or "\x00" in value:
+        raise ServerlessMediaError("Image input reference is invalid")
+    value = value.replace("\\", "/")
+    explicit_root = None
+    annotation = _MEDIA_ANNOTATION.search(value)
+    if annotation:
+        marker = annotation.group(1).lower()
+        if marker not in {"input", "output", "temp"}:
+            raise ServerlessMediaError("Image input annotation is unsupported")
+        explicit_root = marker
+        value = value[: annotation.start()]
+    if "://" in value:
+        raise ServerlessMediaError("Remote media references are not supported as image inputs")
+    if re.match(r"^[A-Za-z]:", value) or value.startswith("//"):
+        raise ServerlessMediaError("Image input path is outside the allowed ComfyUI media directories")
+
+    path = PurePosixPath(value)
+    if any(part == ".." for part in path.parts):
+        raise ServerlessMediaError("Image input path traversal is not allowed")
+    if not path.parts or value in {".", ""}:
+        raise ServerlessMediaError("Image input reference is invalid")
+
+    if explicit_root:
+        value = f"{value} [{explicit_root}]"
+    suffix = Path(path.name).suffix.lower()
+    return value, suffix
+
+
+def _resolve_input_path(value, suffix, folder_paths, roots):
+    if suffix not in _IMAGE_EXTENSIONS:
+        if suffix in _MEDIA_EXTENSIONS:
+            raise ServerlessMediaError("Audio and video inputs are not supported by serverless image transport")
+        raise ServerlessMediaError("Unsupported serverless image file type")
+
+    try:
+        source = Path(value)
+        if source.is_absolute():
+            annotation = _MEDIA_ANNOTATION.search(value)
+            unannotated = value[:annotation.start()] if annotation else value
+            resolved = Path(unannotated).resolve(strict=True)
+        else:
+            resolver = getattr(folder_paths, "get_annotated_filepath", None)
+            if not callable(resolver):
+                raise ServerlessMediaError("ComfyUI media path resolver is unavailable")
+            resolved = Path(resolver(value)).resolve(strict=True)
+    except ServerlessMediaError:
+        raise
+    except Exception:
+        raise ServerlessMediaError("Image input file is missing or invalid") from None
+
+    if not _inside_roots(resolved, roots):
+        raise ServerlessMediaError("Image input path is outside the allowed ComfyUI media directories")
+    try:
+        mode = resolved.stat().st_mode
+    except OSError:
+        raise ServerlessMediaError("Image input file is missing or invalid") from None
+    if not stat.S_ISREG(mode):
+        raise ServerlessMediaError("Image input is not a regular file")
+    return resolved
+
+
+def _iter_image_references(workflow):
+    if not isinstance(workflow, dict):
+        raise ServerlessMediaError("Serverless workflow must be an API graph")
+    references = []
+    for node_id, node in workflow.items():
+        if not isinstance(node, dict):
+            continue
+        class_type = node.get("class_type")
+        inputs = node.get("inputs", {})
+        if class_type == "LoadImage" and not isinstance(inputs, dict):
+            raise ServerlessMediaError("LoadImage requires a literal image filename; linked image inputs are unsupported")
+        if not isinstance(inputs, dict):
+            continue
+        if class_type == "LoadImage":
+            value = inputs.get("image")
+            if not isinstance(value, str):
+                raise ServerlessMediaError("LoadImage requires a literal image filename; linked image inputs are unsupported")
+            references.append((node_id, inputs, "image", value, True))
+
+        for key in ("image", "file"):
+            if key == "image" and class_type == "LoadImage":
+                continue
+            value = inputs.get(key)
+            if not isinstance(value, str):
+                continue
+            normalized, suffix = _normalize_media_reference(value)
+            if suffix in _IMAGE_EXTENSIONS:
+                references.append((node_id, inputs, key, normalized, False))
+            elif suffix in _AUDIO_EXTENSIONS | _VIDEO_EXTENSIONS:
+                raise ServerlessMediaError("Audio and video inputs are not supported by serverless image transport")
+            elif suffix in _UNSUPPORTED_IMAGE_EXTENSIONS or (key == "image" and suffix):
+                raise ServerlessMediaError("Unsupported serverless image file type")
+
+        for key in ("audio", "video"):
+            value = inputs.get(key)
+            if not isinstance(value, str):
+                continue
+            _normalized, suffix = _normalize_media_reference(value)
+            if suffix in _MEDIA_EXTENSIONS:
+                raise ServerlessMediaError("Audio and video inputs are not supported by serverless image transport")
+    return references
+
+
+def _read_limited(path, expected_size):
+    try:
+        descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+        with os.fdopen(descriptor, "rb") as handle:
+            info = os.fstat(handle.fileno())
+            if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_INPUT_IMAGE_BYTES:
+                raise ServerlessMediaError("Image input exceeds the permitted size")
+            data = handle.read(MAX_INPUT_IMAGE_BYTES + 1)
+    except ServerlessMediaError:
+        raise
+    except OSError:
+        raise ServerlessMediaError("Image input file could not be read safely") from None
+    if len(data) > MAX_INPUT_IMAGE_BYTES or len(data) != expected_size:
+        raise ServerlessMediaError("Image input exceeds the permitted size or changed while being read")
+    return data
+
+
+async def prepare_images(workflow: dict) -> list[dict]:
+    """Pack referenced local images and rewrite only their graph input references.
+
+    ``workflow`` is expected to be a private API-graph copy owned by the caller.
+    Returned ``name`` values are flat generated names suitable for ``input.images``.
+    """
+    references = _iter_image_references(workflow)
+    if not references:
+        return []
+
+    # Import only in ComfyUI, keeping this module importable in the serverless client.
+    try:
+        import folder_paths
+    except Exception:
+        raise ServerlessMediaError("ComfyUI media directories are unavailable") from None
+
+    roots = _folder_roots(folder_paths)
+    resolved_refs = []
+    unique_paths = {}
+    total_source_bytes = 0
+    for node_id, inputs, key, value, _is_load_image in references:
+        normalized, suffix = _normalize_media_reference(value)
+        if key == "image" and suffix not in _IMAGE_EXTENSIONS:
+            if suffix in _AUDIO_EXTENSIONS | _VIDEO_EXTENSIONS:
+                raise ServerlessMediaError("Audio and video inputs are not supported by serverless image transport")
+            raise ServerlessMediaError("Unsupported serverless image file type")
+        path = _resolve_input_path(normalized, suffix, folder_paths, roots)
+        if path not in unique_paths:
+            try:
+                size = path.stat().st_size
+            except OSError:
+                raise ServerlessMediaError("Image input file is missing or invalid") from None
+            if size < 0 or size > MAX_INPUT_IMAGE_BYTES:
+                raise ServerlessMediaError("Image input exceeds the permitted size")
+            total_source_bytes += size
+            if total_source_bytes > MAX_INPUT_TOTAL_BYTES:
+                raise ServerlessMediaError("Total serverless image inputs exceed the permitted size")
+            unique_paths[path] = {"size": size, "suffix": suffix}
+        resolved_refs.append((inputs, key, path))
+
+    packed_by_name = {}
+    path_to_name = {}
+    for path, metadata in unique_paths.items():
+        data = _read_limited(path, metadata["size"])
+        digest = hashlib.sha256(data).hexdigest()
+        name = f"serverless_{digest}{metadata['suffix']}"
+        path_to_name[path] = name
+        packed_by_name.setdefault(name, {"name": name, "image": base64.b64encode(data).decode("ascii")})
+
+    for inputs, key, path in resolved_refs:
+        inputs[key] = path_to_name[path]
+    return [packed_by_name[name] for name in sorted(packed_by_name)]
+
+
+def _validate_artifact_url(url):
+    if not isinstance(url, str) or not url or any(ord(char) < 0x20 or char.isspace() for char in url):
+        raise ServerlessMediaError("Serverless image artifact URL is invalid")
+    try:
+        parsed = urlsplit(url)
+        port = parsed.port
+    except ValueError:
+        raise ServerlessMediaError("Serverless image artifact URL is invalid") from None
+    if (
+        parsed.scheme.lower() != "https"
+        or not parsed.hostname
+        or parsed.username is not None
+        or parsed.password is not None
+        or port not in (None, 443)
+        or parsed.fragment
+    ):
+        raise ServerlessMediaError("Serverless image artifact URL must be a public HTTPS URL")
+    host = parsed.hostname.rstrip(".").lower()
+    if host in {"localhost", "metadata", "metadata.google.internal"} or host.endswith((".localhost", ".local", ".internal")):
+        raise ServerlessMediaError("Serverless image artifact URL must be a public HTTPS URL")
+    try:
+        address = ipaddress.ip_address(host)
+    except ValueError:
+        return url
+    if not address.is_global:
+        raise ServerlessMediaError("Serverless image artifact URL must resolve to a public address")
+    return url
+
+
+class _PublicResolver(aiohttp.abc.AbstractResolver):
+    """Resolve once, reject non-public answers, and pin the checked address."""
+
+    async def _getaddrinfo(self, host, port, family):
+        return await asyncio.get_running_loop().getaddrinfo(
+            host, port, family=family, type=socket.SOCK_STREAM
+        )
+
+    async def resolve(self, host, port=0, family=socket.AF_INET):
+        try:
+            results = await self._getaddrinfo(host, port, family)
+        except (OSError, asyncio.TimeoutError):
+            raise OSError("Public artifact host could not be resolved") from None
+        if not results:
+            raise OSError("Public artifact host could not be resolved")
+
+        resolved = []
+        seen = set()
+        for result_family, _socktype, proto, _canonname, sockaddr in results:
+            try:
+                address = ipaddress.ip_address(sockaddr[0])
+            except (ValueError, TypeError, IndexError):
+                raise OSError("Public artifact host resolved to an invalid address") from None
+            if not address.is_global:
+                raise OSError("Public artifact host resolved to a non-public address")
+            key = (str(address), result_family)
+            if key in seen:
+                continue
+            seen.add(key)
+            resolved.append({
+                "hostname": host,
+                "host": str(address),
+                "port": port,
+                "family": result_family,
+                "proto": proto,
+                "flags": 0,
+            })
+        if not resolved:
+            raise OSError("Public artifact host could not be resolved")
+        return resolved
+
+    async def close(self):
+        return None
+
+
+async def download_public_artifact(url: str, max_bytes: int = MAX_OUTPUT_IMAGE_BYTES) -> bytes:
+    """Download a signed public HTTPS artifact without provider credentials."""
+    if isinstance(max_bytes, bool) or not isinstance(max_bytes, int) or not 0 < max_bytes <= MAX_OUTPUT_IMAGE_BYTES:
+        raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
+    current_url = _validate_artifact_url(url)
+    resolver = _PublicResolver()
+    connector = aiohttp.TCPConnector(resolver=resolver, 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):
+                _validate_artifact_url(current_url)
+                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 ServerlessMediaError("Serverless image artifact redirect is invalid")
+                        current_url = _validate_artifact_url(urljoin(current_url, location))
+                        continue
+                    if response.status != 200:
+                        raise ServerlessMediaError("Serverless image artifact download failed")
+                    payload = bytearray()
+                    async for chunk in response.content.iter_chunked(_CHUNK_SIZE):
+                        if not isinstance(chunk, bytes):
+                            raise ServerlessMediaError("Serverless image artifact response is invalid")
+                        if len(payload) + len(chunk) > max_bytes:
+                            raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
+                        payload.extend(chunk)
+                    if not payload:
+                        raise ServerlessMediaError("Serverless image artifact is empty")
+                    return bytes(payload)
+    except ServerlessMediaError:
+        raise
+    except Exception:
+        raise ServerlessMediaError("Serverless image artifact download failed") from None
+    raise ServerlessMediaError("Serverless image artifact download failed")
+
+
+def _validated_manifest(manifest):
+    if not isinstance(manifest, list) or len(manifest) > MAX_OUTPUT_FILES:
+        raise ServerlessMediaError("Serverless image manifest is invalid")
+    normalized = []
+    seen_prefixes = set()
+    prefixes = []
+    for item in manifest:
+        if not isinstance(item, dict):
+            raise ServerlessMediaError("Serverless image manifest is invalid")
+        if item.get("kind") != "image":
+            raise ServerlessMediaError("Only image collectors are supported by serverless media transport")
+        for key in ("collector_id", "job_id", "node_id"):
+            value = item.get(key)
+            if not isinstance(value, (str, int)) or isinstance(value, bool) or not str(value):
+                raise ServerlessMediaError("Serverless image manifest is invalid")
+        prefix = item.get("prefix")
+        if not isinstance(prefix, str) or not _PREFIX_RE.fullmatch(prefix) or prefix in {".", ".."}:
+            raise ServerlessMediaError("Serverless image manifest contains an unsafe filename prefix")
+        folded = prefix.casefold()
+        if folded in seen_prefixes:
+            raise ServerlessMediaError("Serverless image manifest prefixes must be unique")
+        seen_prefixes.add(folded)
+        prefixes.append(prefix)
+        normalized.append({
+            "collector_id": str(item["collector_id"]),
+            "job_id": str(item["job_id"]),
+            "node_id": str(item["node_id"]),
+            "prefix": prefix,
+        })
+    for first in prefixes:
+        for second in prefixes:
+            if first != second and second.casefold().startswith(first.casefold() + "_"):
+                raise ServerlessMediaError("Serverless image manifest prefixes are ambiguous")
+    return normalized
+
+
+def _artifact_filename_matches(filename, prefixes):
+    if not isinstance(filename, str):
+        return None
+    basename = filename.replace("\\", "/").rsplit("/", 1)[-1]
+    matches = [item for item in prefixes if basename.startswith(item["prefix"] + "_")]
+    if len(matches) > 1:
+        raise ServerlessMediaError("Serverless image artifact matches multiple collectors")
+    return matches[0] if matches else None
+
+
+def _canonical_artifact_identity(entry, filename):
+    artifact_path = entry.get("artifact_path")
+    artifact_type = entry.get("artifact_type", "output")
+    if artifact_path is not None:
+        if not isinstance(artifact_path, str) or not artifact_path or "\x00" in artifact_path:
+            raise ServerlessMediaError("Serverless image artifact identity is invalid")
+        normalized_path = os.path.normcase(os.path.normpath(artifact_path.replace("\\", "/")))
+    else:
+        subfolder = entry.get("subfolder", "")
+        if not isinstance(subfolder, str) or "\x00" in subfolder:
+            raise ServerlessMediaError("Serverless image artifact identity is invalid")
+        normalized_path = os.path.normcase(os.path.normpath(f"{subfolder}/{filename}"))
+    return str(artifact_type), normalized_path
+
+
+def _decode_base64_data(value, max_bytes=MAX_OUTPUT_IMAGE_BYTES):
+    if not isinstance(value, str) or not value:
+        raise ServerlessMediaError("Serverless image artifact data is invalid")
+    max_bytes = min(max_bytes, MAX_OUTPUT_IMAGE_BYTES)
+    max_encoded = ((max_bytes + 2) // 3) * 4
+    if len(value) > max_encoded:
+        raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
+    padding = len(value) - len(value.rstrip("="))
+    estimated_size = (len(value) // 4) * 3 - padding
+    if estimated_size > max_bytes:
+        raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
+    try:
+        payload = base64.b64decode(value, validate=True)
+    except (binascii.Error, ValueError):
+        raise ServerlessMediaError("Serverless image artifact data is invalid") from None
+    if not payload:
+        raise ServerlessMediaError("Serverless image artifact is empty")
+    if len(payload) > max_bytes:
+        raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
+    return payload
+
+
+def _decode_image_array(payload, total_array_bytes):
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", Image.DecompressionBombWarning)
+            image = Image.open(io.BytesIO(payload))
+            width, height = image.size
+            if width <= 0 or height <= 0 or width * height > MAX_IMAGE_PIXELS:
+                raise ServerlessMediaError("Serverless image dimensions exceed the permitted limit")
+            has_alpha = image.mode in {"RGBA", "LA"} or "transparency" in image.info
+            mode = "RGBA" if has_alpha else "RGB"
+            output_bytes = width * height * (4 if has_alpha else 3) * 4
+            if total_array_bytes + output_bytes > MAX_OUTPUT_ARRAY_BYTES:
+                raise ServerlessMediaError("Decoded serverless images exceed the permitted memory limit")
+            converted = image.convert(mode)
+            converted.load()
+            array = np.array(converted, dtype=np.float32, copy=True)
+            np.multiply(array, 1.0 / 255.0, out=array)
+            return array, output_bytes
+    except ServerlessMediaError:
+        raise
+    except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError, Image.DecompressionBombWarning):
+        raise ServerlessMediaError("Serverless image artifact could not be decoded") from None
+    except Exception:
+        raise ServerlessMediaError("Serverless image artifact could not be decoded") from None
+
+
+async def collect_images(output: dict, manifest: list, download=None) -> dict[str, list[np.ndarray]]:
+    """Validate and decode expected image artifacts, grouped by serverless job ID."""
+    if not isinstance(output, dict):
+        raise ServerlessMediaError("Serverless job output is invalid")
+    nested = output.get("output")
+    data = nested if isinstance(nested, dict) else output
+    for candidate in (output, data):
+        if candidate.get("error") or candidate.get("errors"):
+            raise ServerlessMediaError("Serverless job reported an execution error")
+
+    expected = _validated_manifest(manifest)
+    if not expected:
+        return {}
+    artifacts = data.get("images", [])
+    if not isinstance(artifacts, list) or len(artifacts) > MAX_OUTPUT_FILES:
+        raise ServerlessMediaError("Serverless job returned too many image files")
+
+    selected = {item["prefix"]: [] for item in expected}
+    identities = set()
+    total_payload_bytes = 0
+    for entry in artifacts:
+        if not isinstance(entry, dict):
+            continue
+        filename = entry.get("filename")
+        manifest_entry = _artifact_filename_matches(filename, expected)
+        if manifest_entry is None:
+            continue
+        normalized_filename = filename.replace("\\", "/")
+        if "/" in normalized_filename or not _SAFE_FILENAME_RE.fullmatch(filename):
+            raise ServerlessMediaError("Serverless image artifact filename is invalid")
+        if Path(filename).suffix.lower() not in _IMAGE_EXTENSIONS:
+            raise ServerlessMediaError("Serverless image artifact has an unsupported file type")
+        if "artifact_type" in entry and entry["artifact_type"] != "output":
+            raise ServerlessMediaError("Expected serverless image artifact was not an output image")
+        identity = _canonical_artifact_identity(entry, filename)
+        if identity in identities:
+            raise ServerlessMediaError("Duplicate serverless image artifact identity")
+        identities.add(identity)
+        match = re.match(rf"^{re.escape(manifest_entry['prefix'])}_(\d+)(?:_|\.)", filename)
+        if match is None:
+            raise ServerlessMediaError("Expected serverless image filename counter is invalid")
+        artifact_type = entry.get("type")
+        remaining_bytes = MAX_OUTPUT_TOTAL_BYTES - total_payload_bytes
+        if remaining_bytes <= 0:
+            raise ServerlessMediaError("Serverless image artifacts exceed the permitted total size")
+        if artifact_type == "base64":
+            payload = _decode_base64_data(entry.get("data"), remaining_bytes)
+        elif artifact_type in {"gcs_url", "s3_url", "url"}:
+            url = entry.get("data") or entry.get("url")
+            url = _validate_artifact_url(url)
+            try:
+                if download is None:
+                    payload = await download_public_artifact(
+                        url, max_bytes=min(MAX_OUTPUT_IMAGE_BYTES, remaining_bytes)
+                    )
+                else:
+                    payload = await download(url)
+            except ServerlessMediaError:
+                raise
+            except Exception:
+                raise ServerlessMediaError("Serverless image artifact download failed") from None
+            if not isinstance(payload, bytes) or not payload:
+                raise ServerlessMediaError("Serverless image artifact response is invalid")
+            if len(payload) > MAX_OUTPUT_IMAGE_BYTES:
+                raise ServerlessMediaError("Serverless image artifact exceeds the permitted size")
+        else:
+            raise ServerlessMediaError("Serverless image artifact transport type is unsupported")
+        total_payload_bytes += len(payload)
+        if total_payload_bytes > MAX_OUTPUT_TOTAL_BYTES:
+            raise ServerlessMediaError("Serverless image artifacts exceed the permitted total size")
+        selected[manifest_entry["prefix"]].append((int(match.group(1)), filename, identity, payload))
+
+    if any(not selected[item["prefix"]] for item in expected):
+        raise ServerlessMediaError("Serverless job is missing an expected collector image")
+
+    result = {}
+    total_array_bytes = 0
+    for manifest_entry in expected:
+        prefix = manifest_entry["prefix"]
+        rows = sorted(selected[prefix], key=lambda row: (row[0], row[1], row[2]))
+        arrays = []
+        shape = None
+        for _counter, _filename, _identity, payload in rows:
+            array, added_bytes = _decode_image_array(payload, total_array_bytes)
+            total_array_bytes += added_bytes
+            if shape is None:
+                shape = array.shape
+            elif array.shape != shape:
+                raise ServerlessMediaError("Images from one serverless collector have inconsistent dimensions")
+            arrays.append(array)
+        result.setdefault(manifest_entry["job_id"], []).extend(arrays)
+    return result

api/orchestration/serverless_prompt.py

@@ -0,0 +1,391 @@
+"""Compile prepared distributed prompts into worker-side ComfyUI image jobs.
+
+This module is deliberately pure: it only copies and rewrites an API prompt
+mapping and never imports or executes ComfyUI nodes.
+"""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import re
+
+
+_SUPPORTED_DISTRIBUTED = {
+    "DistributedCollector",
+    "DistributedSeed",
+    "DistributedValue",
+    "DistributedModelName",
+}
+_REMOVED_SCALAR_NODES = {
+    "DistributedSeed",
+    "DistributedValue",
+    "DistributedModelName",
+}
+_PRIMITIVE_NODES = {
+    "PrimitiveBoolean": False,
+    "PrimitiveFloat": 0.0,
+    "PrimitiveInt": 0,
+    "PrimitiveString": "",
+}
+_REMOVED_NODE_OUTPUTS = {
+    "DistributedCollector": {0, 1},
+    "DistributedSeed": {0},
+    "DistributedValue": {0},
+    "DistributedModelName": {0},
+}
+_SEED_DEFAULT = 1125899906842
+
+
+def _is_link(value):
+    return isinstance(value, list) and len(value) == 2
+
+
+def _is_unsupported_distributed(class_type):
+    if not isinstance(class_type, str) or class_type in _SUPPORTED_DISTRIBUTED:
+        return False
+    lowered = class_type.lower()
+    return "distributed" in lowered or "batchdivider" in lowered
+
+
+def _safe_component(value):
+    cleaned = re.sub(r"[^A-Za-z0-9_-]+", "_", str(value)).strip("_-")
+    return cleaned[:24] or "empty"
+
+
+def _filename_prefix(job_id, worker_id, collector_id, kind):
+    identity = json.dumps(
+        [job_id, worker_id, collector_id, kind],
+        ensure_ascii=False,
+        separators=(",", ":"),
+    )
+    digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()
+    parts = (
+        "serverless",
+        _safe_component(job_id),
+        _safe_component(worker_id),
+        _safe_component(collector_id),
+        kind,
+        digest,
+    )
+    return "_".join(parts)
+
+
+def _worker_index(worker_id):
+    """Mirror nodes.utilities worker-index parsing; None means fallback."""
+    try:
+        if worker_id.startswith("worker_"):
+            return int(worker_id.split("_")[1])
+        return int(worker_id)
+    except (ValueError, IndexError):
+        return None
+
+
+def _coerce(value, value_type):
+    if value_type == "INT":
+        return int(float(value))
+    if value_type == "FLOAT":
+        return float(value)
+    return value
+
+
+def _coerce_safe(value, value_type):
+    try:
+        return _coerce(value, value_type)
+    except (TypeError, ValueError):
+        return value
+
+
+def _stringify_model_name(value):
+    if isinstance(value, str):
+        return value
+    if isinstance(value, (int, float, bool)):
+        return str(value)
+    try:
+        return json.dumps(value, indent=4)
+    except (TypeError, ValueError):
+        return str(value)
+
+
+def _model_name_result(text):
+    if isinstance(text, list):
+        values = [_stringify_model_name(value) for value in text]
+        return values[0] if len(values) == 1 else values
+    return _stringify_model_name(text)
+
+
+def compile_serverless_prompt(prepared_prompt: dict) -> tuple[dict, list[dict]]:
+    """Compile a prepared worker prompt to an image-only serverless graph.
+
+    Every DistributedCollector's original ``images`` input is connected to a
+    new SaveImage node. Only those savers and their upstream dependencies are
+    retained. DistributedSeed/Value/ModelName links are resolved as scalar
+    literals; unsupported node execution is never attempted.
+    """
+    if not isinstance(prepared_prompt, dict):
+        raise ValueError("prepared prompt must be an API graph mapping")
+
+    source_nodes = {}
+    for raw_id, node in prepared_prompt.items():
+        node_id = str(raw_id)
+        if node_id in source_nodes:
+            raise ValueError("prompt has duplicate node IDs after string normalization")
+        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):
+            raise ValueError(f"prompt node {node_id!r} is missing a class_type")
+        if _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")
+        source_nodes[node_id] = node
+
+    if not source_nodes:
+        raise ValueError("prepared prompt contains no nodes")
+
+    # Validate every API link before any graph rewriting. Slots on the nodes
+    # removed by this compiler have a known schema and are checked exactly.
+    for node_id, node in source_nodes.items():
+        for input_name, value in node.get("inputs", {}).items():
+            if not _is_link(value):
+                continue
+            upstream_id, output_slot = value
+            upstream_id = str(upstream_id)
+            if upstream_id not in source_nodes:
+                raise ValueError(
+                    f"invalid link at node {node_id!r} input {input_name!r}: "
+                    f"source {upstream_id!r} does not exist"
+                )
+            if isinstance(output_slot, bool) or not isinstance(output_slot, int) or output_slot < 0:
+                raise ValueError(
+                    f"invalid link at node {node_id!r} input {input_name!r}: "
+                    "output slot must be a nonnegative integer"
+                )
+            upstream_class = source_nodes[upstream_id]["class_type"]
+            valid_slots = _REMOVED_NODE_OUTPUTS.get(upstream_class)
+            if valid_slots is not None and output_slot not in valid_slots:
+                raise ValueError(
+                    f"invalid output slot {output_slot} for removed node "
+                    f"{upstream_class} {upstream_id!r}"
+                )
+
+    collector_ids = sorted(
+        (node_id for node_id, node in source_nodes.items() if node["class_type"] == "DistributedCollector"),
+        key=str,
+    )
+    if not collector_ids:
+        raise ValueError("prepared prompt contains no DistributedCollector nodes")
+
+    # Only image collectors are supported by the current generic handler.
+    for collector_id in collector_ids:
+        inputs = source_nodes[collector_id].get("inputs", {})
+        if "audio" in inputs:
+            raise ValueError("collector audio input is unsupported by this image-only compiler")
+        if not _is_link(inputs.get("images")):
+            raise ValueError(f"collector {collector_id!r} is missing a linked image input")
+        source_id = str(inputs["images"][0])
+        source_class = source_nodes[source_id]["class_type"]
+        if source_class in _REMOVED_SCALAR_NODES:
+            raise ValueError(f"collector {collector_id!r} image input must link to an image node")
+
+    # A collector cannot be used, directly or through processing, as another
+    # collector's input: this first compiler handles disconnected jobs only.
+    for collector_id in collector_ids:
+        pending = [str(source_nodes[collector_id]["inputs"]["images"][0])]
+        seen = set()
+        while pending:
+            node_id = pending.pop()
+            if node_id in seen:
+                continue
+            seen.add(node_id)
+            if source_nodes[node_id]["class_type"] == "DistributedCollector":
+                raise ValueError("collector-to-collector dependence is unsupported")
+            for value in source_nodes[node_id].get("inputs", {}).values():
+                if _is_link(value):
+                    pending.append(str(value[0]))
+
+    # Resolve only stock primitive nodes. A visited set makes malformed
+    # primitive-to-primitive cycles fail clearly instead of recursing forever.
+    def resolve_primitive(node_id, visiting=None):
+        node_id = str(node_id)
+        node = source_nodes[node_id]
+        class_type = node["class_type"]
+        if class_type not in _PRIMITIVE_NODES:
+            raise ValueError("linked value must be literal or supported PrimitiveInt/Float/String/Boolean")
+        if visiting is None:
+            visiting = set()
+        if node_id in visiting:
+            raise ValueError(f"cycle while resolving primitive value at node {node_id!r}")
+        visiting.add(node_id)
+        value = node.get("inputs", {}).get("value", _PRIMITIVE_NODES[class_type])
+        if _is_link(value):
+            parent_id, output_slot = str(value[0]), value[1]
+            if output_slot != 0:
+                raise ValueError(f"invalid output slot {output_slot} for primitive node {parent_id!r}")
+            if source_nodes[parent_id]["class_type"] not in _PRIMITIVE_NODES:
+                raise ValueError("linked primitive value must come from a stock Primitive node")
+            result = resolve_primitive(parent_id, visiting)
+        else:
+            result = copy.deepcopy(value)
+        visiting.remove(node_id)
+        return result
+
+    def resolve_literal_or_primitive(value, context):
+        if _is_link(value):
+            source_id, output_slot = str(value[0]), value[1]
+            if output_slot != 0 or source_nodes[source_id]["class_type"] not in _PRIMITIVE_NODES:
+                raise ValueError(f"{context} must be a literal or supported Primitive link")
+            try:
+                return resolve_primitive(source_id)
+            except ValueError as exc:
+                if "cycle" in str(exc):
+                    raise
+                raise ValueError(f"{context} must be a literal or supported Primitive link") from None
+        return copy.deepcopy(value)
+
+    def resolve_hidden(node_id, name, default):
+        value = source_nodes[node_id].get("inputs", {}).get(name, default)
+        if _is_link(value):
+            raise ValueError(f"{name} on distributed node {node_id!r} must be a literal")
+        return copy.deepcopy(value)
+
+    resolved_distributed = {}
+
+    def resolve_distributed(node_id, output_slot):
+        node_id = str(node_id)
+        class_type = source_nodes[node_id]["class_type"]
+        if output_slot != 0:
+            raise ValueError(f"invalid output slot {output_slot} for {class_type} {node_id!r}")
+        cache_key = (node_id, output_slot)
+        if cache_key in resolved_distributed:
+            return copy.deepcopy(resolved_distributed[cache_key])
+        inputs = source_nodes[node_id].get("inputs", {})
+        if class_type == "DistributedSeed":
+            seed = resolve_literal_or_primitive(inputs.get("seed", _SEED_DEFAULT), f"seed on {node_id!r}")
+            is_worker = resolve_hidden(node_id, "is_worker", False)
+            worker_id = resolve_hidden(node_id, "worker_id", "")
+            if is_worker:
+                worker_index = _worker_index(worker_id)
+                if worker_index is not None:
+                    seed = seed + worker_index + 1
+            result = seed
+        elif class_type == "DistributedValue":
+            default_value = resolve_literal_or_primitive(
+                inputs.get("default_value", ""), f"default_value on {node_id!r}"
+            )
+            worker_values = resolve_literal_or_primitive(
+                inputs.get("worker_values", "{}"), f"worker_values on {node_id!r}"
+            )
+            is_worker = resolve_hidden(node_id, "is_worker", False)
+            worker_id = resolve_hidden(node_id, "worker_id", "")
+            try:
+                values = json.loads(worker_values) if isinstance(worker_values, str) else worker_values
+            except json.JSONDecodeError:
+                values = {}
+            if not isinstance(values, dict):
+                values = {}
+            value_type = values.get("_type", "STRING")
+            fallback = _coerce_safe(default_value, value_type)
+            result = fallback
+            if is_worker:
+                worker_index = _worker_index(worker_id)
+                if worker_index is not None:
+                    raw_value = values.get(str(worker_index + 1), "")
+                    if raw_value:
+                        try:
+                            result = _coerce(raw_value, value_type)
+                        except (ValueError, IndexError):
+                            result = fallback
+        elif class_type == "DistributedModelName":
+            text = resolve_literal_or_primitive(
+                inputs.get("text", ""), f"DistributedModelName text on {node_id!r}"
+            )
+            result = _model_name_result(text)
+        else:
+            raise ValueError(f"unsupported distributed node: {class_type}")
+        resolved_distributed[cache_key] = copy.deepcopy(result)
+        return result
+
+    # The prepared collector values define the artifact identity. The fallback
+    # for absent hidden values is stable and retains the collector's job scope.
+    collector_specs = []
+    max_numeric_id = 0
+    for node_id in source_nodes:
+        try:
+            max_numeric_id = max(max_numeric_id, int(node_id))
+        except ValueError:
+            pass
+    next_numeric_id = max_numeric_id + 1
+    for collector_id in collector_ids:
+        inputs = source_nodes[collector_id].get("inputs", {})
+        raw_job_id = inputs.get("multi_job_id", collector_id)
+        if _is_link(raw_job_id):
+            raise ValueError(f"collector {collector_id!r} multi_job_id must be a literal")
+        job_id = str(collector_id if raw_job_id in (None, "") else raw_job_id)
+        raw_worker_id = inputs.get("worker_id", "")
+        if _is_link(raw_worker_id):
+            raise ValueError(f"collector {collector_id!r} worker_id must be a literal")
+        worker_id = "" if raw_worker_id is None else str(raw_worker_id)
+        saver_id = str(next_numeric_id)
+        next_numeric_id += 1
+        prefix = _filename_prefix(job_id, worker_id, collector_id, "image")
+        collector_specs.append(
+            {
+                "collector_id": collector_id,
+                "job_id": job_id,
+                "node_id": saver_id,
+                "kind": "image",
+                "prefix": prefix,
+                "source_link": copy.deepcopy(inputs["images"]),
+            }
+        )
+
+    saver_nodes = {}
+    for spec in collector_specs:
+        saver_nodes[spec["node_id"]] = {
+            "class_type": "SaveImage",
+            "inputs": {
+                "images": spec["source_link"],
+                "filename_prefix": spec["prefix"],
+            },
+            "_meta": {"title": f"Serverless output: collector {spec['collector_id']} (image)"},
+        }
+
+    all_nodes = {node_id: copy.deepcopy(node) for node_id, node in source_nodes.items()}
+    all_nodes.update(copy.deepcopy(saver_nodes))
+    needed = set()
+    pending = [spec["node_id"] for spec in collector_specs]
+    while pending:
+        node_id = pending.pop()
+        if node_id in needed:
+            continue
+        needed.add(node_id)
+        node = all_nodes[node_id]
+        if node["class_type"] == "DistributedCollector":
+            raise ValueError("collector-to-collector dependence is unsupported")
+        for input_name, value in list(node.get("inputs", {}).items()):
+            if not _is_link(value):
+                continue
+            upstream_id, output_slot = str(value[0]), value[1]
+            upstream_class = source_nodes[upstream_id]["class_type"]
+            if upstream_class in _REMOVED_SCALAR_NODES:
+                node["inputs"][input_name] = resolve_distributed(upstream_id, output_slot)
+            else:
+                pending.append(upstream_id)
+
+    compiled = {
+        node_id: copy.deepcopy(node)
+        for node_id, node in all_nodes.items()
+        if node_id in needed
+    }
+    if any("distributed" in node["class_type"].lower() for node in compiled.values()):
+        raise ValueError("compiled prompt still contains a distributed dependency")
+
+    manifest = [
+        {key: spec[key] for key in ("collector_id", "job_id", "node_id", "kind", "prefix")}
+        for spec in collector_specs
+    ]
+    return compiled, manifest

api/orchestration/serverless_runtime.py

@@ -0,0 +1,275 @@
+"""Master-owned asynchronous Runpod jobs, independent of worker callbacks."""
+import asyncio
+import time
+import uuid
+from pathlib import Path
+
+from .runpod_client import RunpodClient, RunpodError, AmbiguousSubmission
+from .serverless_ledger import JobLedger, TERMINAL
+
+
+async def decode_results(output, manifest):
+    from .serverless_media import collect_images
+    import torch
+    images = await collect_images(output, manifest)
+    return {job_id: [torch.from_numpy(image).unsqueeze(0) for image in batch] for job_id, batch in images.items()}
+
+
+def state_directory():
+    import folder_paths
+    return Path(folder_paths.get_user_directory()) / "distributed" / "serverless"
+
+
+def get_secret_store():
+    from ...utils.serverless_config import SecretStore
+    return SecretStore(state_directory() / "credentials.json")
+
+
+def get_manager():
+    import server
+    from ...utils.network import get_client_session
+    ps = server.PromptServer.instance
+    manager = getattr(ps, "distributed_serverless", None)
+    if manager is None:
+        manager = ServerlessManager(ps, get_secret_store(), state_directory() / "jobs.json", get_client_session)
+        ps.distributed_serverless = manager
+    return manager
+
+
+class ServerlessManager:
+    def __init__(self, prompt_server, secrets, ledger_path, session_factory):
+        self.server = prompt_server
+        self.secrets = secrets
+        self.ledger = JobLedger(ledger_path)
+        self.session_factory = session_factory
+        self.tasks = {}
+        self.runs = {}
+        self.pool_limits = {}
+        self.health_cache = {}
+        self.health_locks = {}
+        self.poll_seconds = 2.0
+
+    async def _client(self, worker):
+        pool_id = worker.get("pool_id", worker["id"])
+        key = self.secrets.get(pool_id)
+        return RunpodClient(await self.session_factory(), worker["endpoint_id"], key)
+
+    async def health(self, worker, *, fresh=False):
+        pool_id = worker.get("pool_id", worker["id"])
+        cache_key = (pool_id, worker["endpoint_id"])
+        lock = self.health_locks.setdefault(cache_key, asyncio.Lock())
+        async with lock:
+            cached = self.health_cache.get(cache_key)
+            if cached and not fresh and time.monotonic() - cached[0] < 10:
+                return cached[1]
+            client = await self._client(worker)
+            data = await client.health()
+            jobs = data.get("jobs", {})
+            if not isinstance(jobs, dict) or not isinstance(data.get("workers"), dict):
+                raise RunpodError("Runpod returned an invalid endpoint health record")
+            try:
+                pending = max(0, int(jobs.get("inQueue", 0))) + max(0, int(jobs.get("inProgress", 0)))
+            except (ValueError, TypeError):
+                raise RunpodError("Runpod returned invalid queue counts") from None
+            result = {"ok": True, "queueRemaining": pending, "message": "Queued / processing" if pending else "Ready (scales from zero)"}
+            self.health_cache[cache_key] = (time.monotonic(), result)
+            return result
+
+    def _semaphore(self, worker):
+        pool_id = worker.get("pool_id", worker["id"])
+        limit = worker["concurrency"]
+        current = self.pool_limits.get(pool_id)
+        if current is None:
+            current = (limit, asyncio.Semaphore(limit))
+            self.pool_limits[pool_id] = current
+        elif current[0] != limit:
+            if any(not task.done() and self.runs[rid]["worker"].get("pool_id", self.runs[rid]["worker"]["id"]) == pool_id for rid, task in self.tasks.items()):
+                raise ValueError("Wait for this serverless pool's current jobs before changing concurrency")
+            current = (limit, asyncio.Semaphore(limit))
+            self.pool_limits[pool_id] = current
+        return current[1]
+
+    def ensure_capacity(self, count):
+        if sum(not task.done() for task in self.tasks.values()) + count > 64:
+            raise ValueError("Distributed already has too many pending serverless jobs (limit 64)")
+
+    async def start(self, execution_id, prompt_id, prepared_jobs):
+        self.ensure_capacity(len(prepared_jobs))
+        new_runs = []
+        # Journal every local slot before starting any billed submission.
+        for prepared in prepared_jobs:
+            worker = prepared["worker"]
+            semaphore = self._semaphore(worker)
+            record_id = uuid.uuid4().hex
+            now = time.time()
+            self.ledger.update(record_id, execution_id=execution_id, prompt_id=prompt_id,
+                               worker_id=worker.get("pool_id", worker["id"]), participant_id=worker["id"],
+                               endpoint_id=worker["endpoint_id"], job_id=None, status="WAITING", created_at=now, updated_at=now)
+            run = dict(prepared, execution_id=execution_id, prompt_id=prompt_id, record_id=record_id,
+                       provider_job_id=None, client=None, deadline=time.monotonic() + worker["timeout_seconds"])
+            self.runs[record_id] = run
+            new_runs.append((record_id, run, semaphore))
+        for record_id, run, semaphore in new_runs:
+            self.tasks[record_id] = asyncio.create_task(self._run(run, semaphore), name=f"distributed-serverless-{record_id}")
+        # Bound in-memory completed tracking as well as the on-disk journal.
+        done = [rid for rid, task in self.tasks.items() if task.done()]
+        for rid in done[:-200]:
+            self.tasks.pop(rid, None)
+            self.runs.pop(rid, None)
+
+    def is_pending(self, collector_job_id, participant_id):
+        return any(not task.done() and run["worker"]["id"] == participant_id and
+                   any(m["job_id"] == collector_job_id for m in run["manifest"])
+                   for rid, task in self.tasks.items() for run in [self.runs[rid]])
+
+    def owns(self, collector_job_id, participant_id):
+        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:
+            for binding in run["manifest"]:
+                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})
+                else:
+                    batch = images[binding["job_id"]]
+                    for index, tensor in enumerate(batch):
+                        queue.put_nowait({"worker_id": run["worker"]["id"], "image_index": index,
+                                          "tensor": tensor, "is_last": index == len(batch) - 1})
+
+    async def _cancel_provider(self, run):
+        if run["provider_job_id"] and run["client"]:
+            try:
+                result = await run["client"].cancel(run["provider_job_id"])
+                return isinstance(result, dict) and result.get("status") == "CANCELLED"
+            except Exception:
+                return False
+        return False
+
+    def _check_master(self, run):
+        queue = getattr(self.server, "prompt_queue", None)
+        if queue is None:
+            return
+        history = queue.get_history(run["prompt_id"])
+        record = history.get(run["prompt_id"], {})
+        if record.get("status", {}).get("status_str") == "error":
+            raise RunpodError("Master workflow failed; serverless work stopped")
+        if not record and hasattr(queue, "get_current_queue"):
+            running, pending = queue.get_current_queue()
+            if not any(str(item[1]) == run["prompt_id"] for item in [*running, *pending]):
+                raise RunpodError("Master workflow was removed from the queue; serverless work stopped")
+
+    async def _run(self, run, semaphore):
+        rid = run["record_id"]
+        submitted = False
+        provider_terminal = False
+        try:
+            async with asyncio.timeout(max(0, run["deadline"] - time.monotonic())):
+                async with semaphore:
+                    self._check_master(run)
+                    client = await self._client(run["worker"])
+                    run["client"] = client
+                    self.ledger.update(rid, status="SUBMITTING", updated_at=time.time())
+                    submitted = True
+                    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())
+                    while True:
+                        self._check_master(run)
+                        response = await client.status(job_id)
+                        status = response.get("status")
+                        if status not in {"IN_QUEUE", "IN_PROGRESS", "COMPLETED", "FAILED", "CANCELLED", "TIMED_OUT"}:
+                            raise RunpodError("Runpod returned an unknown job state")
+                        provider_terminal = status in TERMINAL
+                        if status == "COMPLETED":
+                            output = response.get("output")
+                            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)
+                            self.ledger.update(rid, status="COMPLETED", updated_at=time.time())
+                            return
+                        self.ledger.update(rid, status=status, updated_at=time.time())
+                        if status in TERMINAL:
+                            raise RunpodError(f"Serverless job ended with {status}")
+                        await asyncio.sleep(self.poll_seconds)
+        except AmbiguousSubmission as exc:
+            self.ledger.update(rid, status="UNKNOWN", error=str(exc), updated_at=time.time())
+            await self._deliver(run, error=str(exc))
+        except (asyncio.TimeoutError, asyncio.CancelledError) as exc:
+            confirmed = await self._cancel_provider(run)
+            if submitted and not run["provider_job_id"]:
+                status = "UNKNOWN"
+                error = "Submission interrupted before its job ID was received; check Runpod before retrying"
+            elif isinstance(exc, asyncio.TimeoutError):
+                status = "TIMED_OUT" if not run["provider_job_id"] or confirmed else "CANCEL_REQUESTED"
+                error = "Serverless job exceeded its queue/execution deadline"
+            else:
+                status = "CANCELLED" if not run["provider_job_id"] or confirmed else "CANCEL_REQUESTED"
+                error = "Serverless job cancelled" if confirmed or not submitted else "Serverless cancellation requested; provider confirmation is pending"
+            self.ledger.update(rid, status=status, error=error, updated_at=time.time())
+            await self._deliver(run, error=error)
+        except Exception as exc:
+            # If polling/decoding failed, stop only our own provider job, never the pool.
+            confirmed = await self._cancel_provider(run) if not provider_terminal else True
+            error = str(exc) if isinstance(exc, RunpodError) else "Serverless processing failed; check the workflow and endpoint contract"
+            status = "CANCEL_REQUESTED" if run["provider_job_id"] and not confirmed else "FAILED"
+            self.ledger.update(rid, status=status, error=error, updated_at=time.time())
+            await self._deliver(run, error=error)
+        finally:
+            # Completed tracking retains IDs/bindings only, not potentially large
+            # base64 inputs, workflow metadata, or a client holding credentials.
+            run.pop("workflow", None)
+            run.pop("images", None)
+            run["client"] = None
+
+    async def wait_for_execution(self, execution_id):
+        tasks = [task for rid, task in self.tasks.items() if self.runs[rid]["execution_id"] == execution_id]
+        await asyncio.gather(*tasks, return_exceptions=False)
+
+    async def cancel(self, worker_id=None, execution_id=None, collector_job_id=None):
+        targets = []
+        for rid, task in self.tasks.items():
+            run = self.runs[rid]
+            if worker_id and run["worker"].get("pool_id", run["worker"]["id"]) != worker_id:
+                continue
+            if execution_id and run["execution_id"] != execution_id:
+                continue
+            if collector_job_id and not any(m["job_id"] == collector_job_id for m in run["manifest"]):
+                continue
+            if not task.done():
+                task.cancel()
+                targets.append(task)
+        if targets:
+            await asyncio.gather(*targets, return_exceptions=True)
+        # A restarted master must still be able to cancel IDs it journalled.
+        # No workflow payloads are persisted, so never resume or resubmit them.
+        recovered = 0
+        for rec in self.ledger.all():
+            task = self.tasks.get(rec["record_id"])
+            if (task is not None and not task.done()) or rec.get("status") in TERMINAL:
+                continue
+            if worker_id and rec.get("worker_id") != worker_id:
+                continue
+            if execution_id and rec.get("execution_id") != execution_id:
+                continue
+            if collector_job_id or not rec.get("job_id"):
+                continue
+            try:
+                client = await self._client({"id": rec["worker_id"], "endpoint_id": rec["endpoint_id"]})
+                result = await client.cancel(rec["job_id"])
+                status = "CANCELLED" if result.get("status") == "CANCELLED" else "CANCEL_REQUESTED"
+            except Exception:
+                status = "CANCEL_REQUESTED"
+            self.ledger.update(rec["record_id"], status=status, updated_at=time.time())
+            recovered += 1
+        return len(targets) + recovered
+
+    def public_jobs(self, worker_id):
+        return [{key: rec.get(key) for key in ("job_id", "status", "worker_id", "execution_id", "error")}
+                for rec in reversed(self.ledger.all()) if rec.get("worker_id") == worker_id][:20]

api/queue_orchestration.py

@@ -74,6 +74,14 @@ def _resolve_enabled_workers(config, requested_ids=None):
         elif not worker.get("enabled", False):
             continue
 
+        if worker.get("type") == "serverless":
+            from ..utils.serverless_config import validate_serverless_worker
+            pool = validate_serverless_worker(worker)
+            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}"))
+            continue
+
         raw_port = worker.get("port", worker.get("listen_port", 8188))
         try:
             port = int(raw_port or 8188)
@@ -154,6 +162,17 @@ async def _prepare_worker_payload(
     """Prepare one worker prompt payload with bounded concurrency and media-sync timeout."""
     async with worker_prep_semaphore:
         worker_prompt = prompt_index.copy_prompt()
+        if worker.get("type") == "serverless":
+            from .orchestration.serverless_prompt import compile_serverless_prompt
+            from .orchestration.serverless_media import prepare_images
+            worker_prompt = apply_participant_overrides(
+                prune_prompt_for_worker(worker_prompt), worker["id"], enabled_ids,
+                job_id_map, "", delegate_master, prompt_index,
+            )
+            workflow, manifest = compile_serverless_prompt(worker_prompt)
+            images = await prepare_images(workflow)
+            return worker, {"worker": worker, "workflow": workflow, "manifest": manifest, "images": images}
+
         worker_master_url = build_master_callback_url(
             worker,
             config=config,
@@ -330,9 +349,6 @@ async def orchestrate_distributed_execution(
             queue_result.get("node_errors", {}),
         )
 
-    for job_id in job_id_map.values():
-        await _ensure_distributed_queue(job_id)
-
     master_prompt = prompt_index.copy_prompt()
     master_prompt = apply_participant_overrides(
         master_prompt,
@@ -387,7 +403,19 @@ async def orchestrate_distributed_execution(
             ]
         )
 
-    if worker_payloads:
+    serverless_jobs = [payload for worker, payload in worker_payloads if worker.get("type") == "serverless"]
+    regular_payloads = [(worker, payload) for worker, payload in worker_payloads if worker.get("type") != "serverless"]
+    manager = None
+    if serverless_jobs:
+        from .orchestration.serverless_runtime import get_manager
+        manager = get_manager()
+        manager.ensure_capacity(len(serverless_jobs))
+
+    # Do not leave collector queues behind when offline compilation/media validation fails.
+    for job_id in job_id_map.values():
+        await _ensure_distributed_queue(job_id)
+
+    if regular_payloads:
         await asyncio.gather(
             *[
                 dispatch_worker_prompt(
@@ -398,17 +426,38 @@ async def orchestrate_distributed_execution(
                     use_websocket=use_websocket,
                     trace_execution_id=execution_trace_id,
                 )
-                for worker, wprompt in worker_payloads
+                for worker, wprompt in regular_payloads
             ]
         )
 
-    queue_result = await queue_prompt_payload(
-        master_prompt,
-        workflow_meta,
-        client_id,
-        include_queue_metadata=True,
-    )
+    try:
+        queue_result = await queue_prompt_payload(
+            master_prompt,
+            workflow_meta,
+            client_id,
+            include_queue_metadata=True,
+        )
+    except Exception:
+        if serverless_jobs:
+            async with prompt_server.distributed_jobs_lock:
+                for job_id in job_id_map.values():
+                    prompt_server.distributed_pending_jobs.pop(job_id, None)
+        raise
     prompt_id = queue_result["prompt_id"]
+    if manager is not None:
+        try:
+            await manager.start(execution_trace_id, prompt_id, serverless_jobs)
+        except Exception:
+            # Wake queued collectors with an explicit error if journal/capacity setup failed.
+            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"])
+                        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"})
+            raise
     prompt_number = queue_result["number"]
     node_errors = queue_result.get("node_errors", {})
     trace_debug(

api/serverless_routes.py

@@ -0,0 +1,58 @@
+"""Master-only Runpod pool monitoring. Health checks never submit inference."""
+from aiohttp import web
+import server
+
+from ..utils.config import load_config
+from .orchestration.serverless_runtime import get_manager
+from .orchestration.runpod_client import RunpodError
+
+
+def _pool(worker_id):
+    return next((w for w in load_config().get("workers", [])
+                 if w.get("id") == worker_id and w.get("type") == "serverless"), None)
+
+
+async def _status(worker_id, fresh=False):
+    worker = _pool(worker_id)
+    if worker is None:
+        return web.json_response({"ok": False, "message": "Serverless worker not found"}, status=404)
+    try:
+        manager = get_manager()
+        result = await manager.health(worker, fresh=fresh)
+        return web.json_response(dict(result, jobs=manager.public_jobs(worker_id)))
+    except Exception as exc:
+        message = str(exc) if isinstance(exc, RunpodError) else "Check the saved endpoint ID and API key"
+        return web.json_response({"ok": False, "queueRemaining": 0, "message": message})
+
+
+@server.PromptServer.instance.routes.post("/distributed/serverless/test")
+async def test_serverless_endpoint(request):
+    try:
+        data = await request.json()
+        worker_id = data.get("worker_id")
+    except Exception:
+        return web.json_response({"ok": False, "message": "Invalid request"}, status=400)
+    return await _status(worker_id, fresh=True)
+
+
+@server.PromptServer.instance.routes.get("/distributed/serverless/status/{worker_id}")
+async def serverless_status_endpoint(request):
+    return await _status(request.match_info["worker_id"])
+
+
+@server.PromptServer.instance.routes.post("/distributed/serverless/cancel")
+async def cancel_serverless_endpoint(request):
+    try:
+        data = await request.json()
+        worker_id = data.get("worker_id")
+    except Exception:
+        return web.json_response({"ok": False, "message": "Invalid request"}, status=400)
+    if _pool(worker_id) is None:
+        return web.json_response({"ok": False, "message": "Serverless worker not found"}, status=404)
+    try:
+        manager = get_manager()
+        count = await manager.cancel(worker_id=worker_id)
+        return web.json_response({"ok": True, "cancelled": count, "jobs": manager.public_jobs(worker_id),
+                                  "message": "Cancellation requested for this master's jobs only"})
+    except Exception:
+        return web.json_response({"ok": False, "message": "Cancellation could not be confirmed; check provider jobs"}, status=502)

api/worker_routes.py

@@ -460,6 +460,9 @@ async def launch_worker_endpoint(request):
         if not worker:
             return await handle_api_error(request, f"Worker {worker_id} not found", 404)
         
+        if worker.get("type") == "serverless":
+            return await handle_api_error(request, "Serverless pools are managed by Runpod, not local processes", 400)
+
         # Ensure consistent string ID
         worker_id_str = worker_id
         
@@ -554,6 +557,9 @@ async def get_local_worker_status_endpoint(request):
         worker_statuses = {}
         
         for worker in config.get("workers", []):
+            # Serverless pools have no host, port, or local process.
+            if worker.get("type") == "serverless":
+                continue
             # Only check local workers
             host = normalize_host(worker.get("host")) or ""
             if not host or host in ["localhost", "127.0.0.1"]: