Add Comfy-Usage-Source pass-through for API node requests

Capture the Comfy-Usage-Source header (or extra_data.comfy_usage_source)
on POST /prompt and forward it on API nodes' outbound requests to
api.comfy.org, defaulting to comfyui-server when absent.

README.md

@@ -462,6 +462,16 @@ To use the most up-to-date frontend version:
 
 This approach allows you to easily switch between the stable fortnightly release and the cutting-edge daily updates, or even specific versions for testing purposes.
 
+### Accessing the Legacy Frontend
+
+If you need to use the legacy frontend for any reason, you can access it using the following command line argument:
+
+```
+--front-end-version Comfy-Org/ComfyUI_legacy_frontend@latest
+```
+
+This will use a snapshot of the legacy frontend preserved in the [ComfyUI Legacy Frontend repository](https://github.com/Comfy-Org/ComfyUI_legacy_frontend).
+
 # QA
 
 ### Which GPU should I buy for this?

comfy_api/latest/_io.py

@@ -1400,7 +1400,8 @@ class V3Data(TypedDict):
 class HiddenHolder:
     def __init__(self, unique_id: str, prompt: Any,
                  extra_pnginfo: Any, dynprompt: Any,
-                 auth_token_comfy_org: str, api_key_comfy_org: str, **kwargs):
+                 auth_token_comfy_org: str, api_key_comfy_org: str,
+                 comfy_usage_source: str = None, **kwargs):
         self.unique_id = unique_id
         """UNIQUE_ID is the unique identifier of the node, and matches the id property of the node on the client side. It is commonly used in client-server communications (see messages)."""
         self.prompt = prompt
@@ -1413,6 +1414,8 @@ class HiddenHolder:
         """AUTH_TOKEN_COMFY_ORG is a token acquired from signing into a ComfyOrg account on frontend."""
         self.api_key_comfy_org = api_key_comfy_org
         """API_KEY_COMFY_ORG is an API Key generated by ComfyOrg that allows skipping signing into a ComfyOrg account on frontend."""
+        self.comfy_usage_source = comfy_usage_source
+        """COMFY_USAGE_SOURCE identifies the client that submitted the prompt (e.g. comfyui-frontend, comfy-cli, comfyui-mcp); forwarded to API nodes' upstream requests via the Comfy-Usage-Source header."""
 
     def __getattr__(self, key: str):
         '''If hidden variable not found, return None.'''
@@ -1429,6 +1432,7 @@ class HiddenHolder:
             dynprompt=d.get(Hidden.dynprompt, None),
             auth_token_comfy_org=d.get(Hidden.auth_token_comfy_org, None),
             api_key_comfy_org=d.get(Hidden.api_key_comfy_org, None),
+            comfy_usage_source=d.get(Hidden.comfy_usage_source, None),
         )
 
     @classmethod
@@ -1451,6 +1455,8 @@ class Hidden(str, Enum):
     """AUTH_TOKEN_COMFY_ORG is a token acquired from signing into a ComfyOrg account on frontend."""
     api_key_comfy_org = "API_KEY_COMFY_ORG"
     """API_KEY_COMFY_ORG is an API Key generated by ComfyOrg that allows skipping signing into a ComfyOrg account on frontend."""
+    comfy_usage_source = "COMFY_USAGE_SOURCE"
+    """COMFY_USAGE_SOURCE identifies the client that submitted the prompt (e.g. comfyui-frontend, comfy-cli, comfyui-mcp); forwarded to API nodes' upstream requests via the Comfy-Usage-Source header."""
 
 
 @dataclass
@@ -1654,6 +1660,8 @@ class Schema:
                 self.hidden.append(Hidden.auth_token_comfy_org)
             if Hidden.api_key_comfy_org not in self.hidden:
                 self.hidden.append(Hidden.api_key_comfy_org)
+            if Hidden.comfy_usage_source not in self.hidden:
+                self.hidden.append(Hidden.comfy_usage_source)
         # if is an output_node, will need prompt and extra_pnginfo
         if self.is_output_node:
             if Hidden.prompt not in self.hidden:

comfy_api_nodes/nodes_sonilo.py

@@ -18,6 +18,7 @@ from comfy_api_nodes.util._helpers import (
     default_base_url,
     get_auth_header,
     get_node_id,
+    get_usage_source,
     is_processing_interrupted,
 )
 from comfy_api_nodes.util.common_exceptions import ProcessingInterrupted
@@ -176,6 +177,7 @@ async def _stream_sonilo_music(
 
     headers: dict[str, str] = {}
     headers.update(get_auth_header(cls))
+    headers["Comfy-Usage-Source"] = get_usage_source(cls)
     headers.update(endpoint.headers)
 
     node_id = get_node_id(cls)

comfy_api_nodes/util/_helpers.py

@@ -35,6 +35,11 @@ def get_auth_header(node_cls: type[IO.ComfyNode]) -> dict[str, str]:
     return {}
 
 
+def get_usage_source(node_cls: type[IO.ComfyNode]) -> str:
+    """Source of the prompt that triggered this API node, defaulting to this server itself."""
+    return node_cls.hidden.comfy_usage_source or "comfyui-server"
+
+
 def default_base_url() -> str:
     return getattr(args, "comfy_api_base", "https://api.comfy.org")
 

comfy_api_nodes/util/client.py

@@ -26,6 +26,7 @@ from ._helpers import (
     default_base_url,
     get_auth_header,
     get_node_id,
+    get_usage_source,
     is_processing_interrupted,
     sleep_with_interrupt,
 )
@@ -647,6 +648,7 @@ async def _request_base(cfg: _RequestConfig, expect_binary: bool):
         if not parsed_url.scheme and not parsed_url.netloc:  # is URL relative?
             payload_headers.update(get_auth_header(cfg.node_cls))
             payload_headers["Comfy-Env"] = get_deploy_environment()
+            payload_headers["Comfy-Usage-Source"] = get_usage_source(cfg.node_cls)
         if cfg.endpoint.headers:
             payload_headers.update(cfg.endpoint.headers)
 

comfy_api_nodes/util/download_helpers.py

@@ -18,6 +18,7 @@ from . import request_logger
 from ._helpers import (
     default_base_url,
     get_auth_header,
+    get_usage_source,
     is_processing_interrupted,
     sleep_with_interrupt,
     to_aiohttp_url,
@@ -65,6 +66,7 @@ async def download_url_to_bytesio(
             raise ValueError("For relative 'cloud' paths, the `cls` parameter is required.")
         url = urljoin(default_base_url().rstrip("/") + "/", url.lstrip("/"))
         headers = get_auth_header(cls)
+        headers["Comfy-Usage-Source"] = get_usage_source(cls)
 
     while True:
         attempt += 1

comfy_execution/asset_enrichment.py

@@ -1,66 +0,0 @@
-"""Enrich executed-node output entries with asset id."""
-import logging
-import os
-
-
-def enrich_output_with_assets(output_ui: dict) -> dict:
-    """Register file-type output entries as assets and inject their ``id``.
-
-    Runs at output-processing time, once per produced output, when
-    --enable-assets is set. Returns a new dict; entries without a resolvable
-    on-disk file path are left unchanged. Errors are caught per-entry so a
-    failure never blocks execution or the other entries.
-    """
-    from comfy.cli_args import args
-    if not args.enable_assets:
-        return output_ui
-
-    import folder_paths
-    from app.assets.services.ingest import register_file_in_place, DependencyMissingError
-
-    enriched = {}
-    for key, entries in output_ui.items():
-        if not isinstance(entries, list):
-            enriched[key] = entries
-            continue
-        new_entries = []
-        for entry in entries:
-            if not isinstance(entry, dict) or "filename" not in entry or "type" not in entry:
-                new_entries.append(entry)
-                continue
-            try:
-                base = folder_paths.get_directory_by_type(entry["type"])
-                if base is None:
-                    new_entries.append(entry)
-                    continue
-                base_abs = os.path.abspath(base)
-                abs_path = os.path.abspath(os.path.join(base_abs, entry.get("subfolder") or "", entry["filename"]))
-                try:
-                    if os.path.commonpath([base_abs, abs_path]) != base_abs:
-                        raise ValueError("escapes base")
-                except ValueError:
-                    logging.warning("Asset enrichment skipped (path escapes base): %s", entry.get("filename"))
-                    new_entries.append(entry)
-                    continue
-                if not os.path.isfile(abs_path):
-                    new_entries.append(entry)
-                    continue
-
-                # Register unconditionally: the file was just produced, and
-                # register_file_in_place re-hashes so an overwritten path can
-                # never carry a stale id.
-                result = register_file_in_place(
-                    abs_path=abs_path,
-                    name=entry["filename"],
-                    tags=[entry["type"]],
-                )
-
-                entry = dict(entry)
-                entry["id"] = result.ref.id
-            except DependencyMissingError:
-                logging.warning("Asset enrichment skipped (blake3 not available): %s", entry.get("filename"))
-            except Exception:
-                logging.warning("Failed to enrich output entry with asset id: %s", entry.get("filename"), exc_info=True)
-            new_entries.append(entry)
-        enriched[key] = new_entries
-    return enriched

comfy_execution/jobs.py

@@ -3,7 +3,6 @@ Job utilities for the /api/jobs endpoint.
 Provides normalization and helper functions for job status tracking.
 """
 
-import uuid
 from typing import Optional
 
 from comfy_api.internal import prune_dict
@@ -20,25 +19,6 @@ class JobStatus:
     ALL = [PENDING, IN_PROGRESS, COMPLETED, FAILED, CANCELLED]
 
 
-def validate_job_id(value) -> str:
-    """Validate a client-supplied job (prompt) id.
-
-    Job ids must be UUIDs in the canonical lowercase hyphenated form. The id
-    is stored and compared verbatim everywhere downstream — history keys,
-    websocket events, and /interrupt matching — so accepting another spelling
-    would silently rewrite the client's id and then miss every exact-match
-    lookup. Rejecting loudly beats that.
-
-    Returns the id unchanged. Raises ValueError when the value is not a
-    string in canonical UUID form.
-    """
-    if not isinstance(value, str):
-        raise ValueError(f"job id must be a string, got {type(value).__name__}")
-    if str(uuid.UUID(value)) != value:
-        raise ValueError("job id must be a UUID in canonical lowercase hyphenated form")
-    return value
-
-
 # Media types that can be previewed in the frontend
 PREVIEWABLE_MEDIA_TYPES = frozenset({'images', 'video', 'audio', '3d', 'text'})
 

execution.py

@@ -40,7 +40,6 @@ from comfy_execution.graph_utils import GraphBuilder, is_link
 from comfy_execution.validation import validate_node_input
 from comfy_execution.progress import get_progress_state, reset_progress_state, add_progress_handler, WebUIProgressHandler
 from comfy_execution.utils import CurrentNodeContext
-from comfy_execution.asset_enrichment import enrich_output_with_assets
 from comfy_api.internal import _ComfyNodeInternal, _NodeOutputInternal, first_real_override, is_class, make_locked_method_func
 from comfy_api.latest import io, _io
 from comfy_execution.cache_provider import _has_cache_providers, _get_cache_providers, _logger as _cache_logger
@@ -200,6 +199,8 @@ def get_input_data(inputs, class_def, unique_id, execution_list=None, dynprompt=
                 hidden_inputs_v3[io.Hidden.auth_token_comfy_org] = extra_data.get("auth_token_comfy_org", None)
             if io.Hidden.api_key_comfy_org.name in hidden:
                 hidden_inputs_v3[io.Hidden.api_key_comfy_org] = extra_data.get("api_key_comfy_org", None)
+            if io.Hidden.comfy_usage_source.name in hidden:
+                hidden_inputs_v3[io.Hidden.comfy_usage_source] = extra_data.get("comfy_usage_source", None)
     else:
         if "hidden" in valid_inputs:
             h = valid_inputs["hidden"]
@@ -216,6 +217,8 @@ def get_input_data(inputs, class_def, unique_id, execution_list=None, dynprompt=
                     input_data_all[x] = [extra_data.get("auth_token_comfy_org", None)]
                 if h[x] == "API_KEY_COMFY_ORG":
                     input_data_all[x] = [extra_data.get("api_key_comfy_org", None)]
+                if h[x] == "COMFY_USAGE_SOURCE":
+                    input_data_all[x] = [extra_data.get("comfy_usage_source", None)]
     v3_data["hidden_inputs"] = hidden_inputs_v3
     return input_data_all, missing_keys, v3_data
 
@@ -419,7 +422,6 @@ def _is_intermediate_output(dynprompt, node_id):
     class_def = nodes.NODE_CLASS_MAPPINGS[class_type]
     return getattr(class_def, 'HAS_INTERMEDIATE_OUTPUT', False)
 
-
 def _send_cached_ui(server, node_id, display_node_id, cached, prompt_id, ui_outputs):
     if server.client_id is None:
         return
@@ -554,10 +556,6 @@ async def execute(server, dynprompt, caches, current_item, extra_data, executed,
                 asyncio.create_task(await_completion())
                 return (ExecutionResult.PENDING, None, None)
         if len(output_ui) > 0:
-            # Enrich at output-processing time (not in the send path) so assets
-            # are registered even when no client is connected, and the asset id
-            # flows into ui_outputs and the cache alongside the raw entries.
-            output_ui = enrich_output_with_assets(output_ui)
             ui_outputs[unique_id] = {
                 "meta": {
                     "node_id": unique_id,

openapi.yaml

@@ -896,11 +896,6 @@ components:
                     additionalProperties: true
                     description: The workflow graph to execute
                     type: object
-                prompt_id:
-                    description: Optional client-supplied job id. Must be a UUID in canonical lowercase hyphenated form; it is echoed back in the response. Omitted or null means the server generates one.
-                    format: uuid
-                    nullable: true
-                    type: string
                 workflow_id:
                     description: UUID identifying the cloud workflow entity to associate with this job
                     type: string
@@ -1067,9 +1062,6 @@ components:
                         comfyui_version:
                             description: ComfyUI version
                             type: string
-                        deploy_environment:
-                            description: How this ComfyUI instance is deployed (e.g. cloud, local-git, local-portable, local-desktop)
-                            type: string
                         embedded_python:
                             description: Whether using embedded Python
                             type: boolean

server.py

@@ -8,7 +8,7 @@ import time
 import nodes
 import folder_paths
 import execution
-from comfy_execution.jobs import JobStatus, get_job, get_all_jobs, validate_job_id
+from comfy_execution.jobs import JobStatus, get_job, get_all_jobs
 import uuid
 import urllib
 import json
@@ -942,21 +942,7 @@ class PromptServer():
 
             if "prompt" in json_data:
                 prompt = json_data["prompt"]
-                client_prompt_id = json_data.get("prompt_id")
-                if client_prompt_id is None:
-                    # Absent or explicit null: the server mints the id.
-                    prompt_id = str(uuid.uuid4())
-                else:
-                    try:
-                        prompt_id = validate_job_id(client_prompt_id)
-                    except ValueError:
-                        error = {
-                            "type": "invalid_prompt_id",
-                            "message": "prompt_id must be a valid UUID",
-                            "details": "prompt_id must be a UUID string in canonical lowercase hyphenated form; omit it to let the server generate one",
-                            "extra_info": {}
-                        }
-                        return web.json_response({"error": error, "node_errors": {}}, status=400)
+                prompt_id = str(json_data.get("prompt_id", uuid.uuid4()))
 
                 partial_execution_targets = None
                 if "partial_execution_targets" in json_data:
@@ -971,6 +957,11 @@ class PromptServer():
 
                 if "client_id" in json_data:
                     extra_data["client_id"] = json_data["client_id"]
+
+                if "comfy_usage_source" not in extra_data:
+                    usage_source = request.headers.get("Comfy-Usage-Source")
+                    if usage_source:
+                        extra_data["comfy_usage_source"] = usage_source
                 if valid[0]:
                     outputs_to_execute = valid[2]
                     sensitive = {}

tests-unit/assets_test/test_prompt_id_enforcement.py

@@ -1,69 +0,0 @@
-"""POST /prompt enforces canonical-UUID job ids at creation time.
-
-Lives in assets_test because it uses this suite's booted-server fixture. The
-invariant itself is pipeline-wide: a job id is stored and compared verbatim
-downstream — history keys, websocket correlation, and /interrupt matching —
-so a job minted with a non-canonical id would miss every exact-match lookup.
-
-The prompt bodies here are intentionally invalid workflows — prompt_id
-validation happens before workflow validation, so a rejected id returns
-``invalid_prompt_id`` while an accepted id falls through to the ordinary
-workflow-validation error (proving it cleared the id check).
-"""
-import requests
-
-
-def _post_prompt(http: requests.Session, api_base: str, body: dict) -> requests.Response:
-    return http.post(api_base + "/prompt", json=body, timeout=30)
-
-
-def _error_type(r: requests.Response) -> str:
-    return r.json()["error"]["type"]
-
-
-def test_non_uuid_prompt_id_rejected(http: requests.Session, api_base: str):
-    r = _post_prompt(http, api_base, {"prompt": {}, "prompt_id": "not-a-uuid"})
-    assert r.status_code == 400, r.text
-    assert _error_type(r) == "invalid_prompt_id"
-
-
-def test_non_string_prompt_id_rejected(http: requests.Session, api_base: str):
-    # Previously str()-coerced (123 became the job id "123"); must now be a 400,
-    # not a 500 from uuid.UUID choking on a non-string.
-    r = _post_prompt(http, api_base, {"prompt": {}, "prompt_id": 123})
-    assert r.status_code == 400, r.text
-    assert _error_type(r) == "invalid_prompt_id"
-
-
-def test_non_canonical_uuid_rejected(http: requests.Session, api_base: str):
-    # Parseable as a UUID, but not the canonical lowercase form: rejected
-    # loudly rather than silently rewritten (downstream lookups match the
-    # stored id exactly).
-    r = _post_prompt(
-        http,
-        api_base,
-        {"prompt": {}, "prompt_id": "AAAAAAAA-BBBB-4CCC-8DDD-EEEEEEEEEEEE"},
-    )
-    assert r.status_code == 400, r.text
-    assert _error_type(r) == "invalid_prompt_id"
-
-
-def test_canonical_uuid_accepted(http: requests.Session, api_base: str):
-    # The id clears validation; the empty workflow then fails ordinary prompt
-    # validation, proving the request got past the id check.
-    r = _post_prompt(
-        http,
-        api_base,
-        {"prompt": {}, "prompt_id": "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee"},
-    )
-    assert r.status_code == 400, r.text
-    assert _error_type(r) != "invalid_prompt_id"
-
-
-def test_null_prompt_id_not_rejected(http: requests.Session, api_base: str):
-    # Explicit null means "server generates" and must not be rejected as an
-    # invalid id. (The minted id itself is not observable here because the
-    # workflow is invalid; unit tests cover validate_job_id directly.)
-    r = _post_prompt(http, api_base, {"prompt": {}, "prompt_id": None})
-    assert r.status_code == 400, r.text
-    assert _error_type(r) != "invalid_prompt_id"

tests-unit/execution_test/test_enrich_output.py

@@ -1,205 +0,0 @@
-"""Tests for enrich_output_with_assets in comfy_execution/asset_enrichment.py."""
-import os
-import types
-import unittest
-from unittest.mock import MagicMock, patch
-
-
-def _make_args(enable_assets: bool):
-    a = types.SimpleNamespace()
-    a.enable_assets = enable_assets
-    return a
-
-
-def _make_register_result(ref_id="ref-id-2"):
-    result = MagicMock()
-    result.ref.id = ref_id
-    return result
-
-
-# Platform-appropriate absolute base. tempfile.gettempdir() returns C:\... on
-# Windows and /tmp on POSIX, so containment via commonpath behaves naturally.
-_DEFAULT_BASE = os.path.join(__import__("tempfile").gettempdir(), "asset-enrichment-test-base")
-
-
-def _mocked_modules(*, enable_assets=True, register_file_in_place=None, directory=_DEFAULT_BASE):
-    return {
-        "comfy.cli_args": MagicMock(args=_make_args(enable_assets)),
-        "folder_paths": MagicMock(get_directory_by_type=MagicMock(return_value=directory)),
-        "app.assets.services.ingest": MagicMock(
-            register_file_in_place=register_file_in_place or MagicMock(return_value=_make_register_result()),
-            DependencyMissingError=type("DependencyMissingError", (Exception,), {}),
-        ),
-    }
-
-
-def _call(output_ui, *, enable_assets=True, file_exists=True, register_result=None, directory=_DEFAULT_BASE):
-    register_mock = MagicMock(return_value=register_result or _make_register_result())
-    mocked = _mocked_modules(
-        enable_assets=enable_assets,
-        register_file_in_place=register_mock,
-        directory=directory,
-    )
-
-    # Only os.path.isfile is patched — abspath/join must run natively so the
-    # containment check sees real platform paths.
-    with patch.dict("sys.modules", mocked), \
-         patch("os.path.isfile", return_value=file_exists):
-        import importlib
-        import comfy_execution.asset_enrichment as mod
-        importlib.reload(mod)
-        return mod.enrich_output_with_assets(output_ui)
-
-
-class TestEnrichOutputWithAssets(unittest.TestCase):
-
-    def test_disabled_returns_unchanged(self):
-        output = {"images": [{"filename": "a.png", "subfolder": "", "type": "output"}]}
-        result = _call(output, enable_assets=False)
-        self.assertNotIn("id", result["images"][0])
-
-    def test_non_list_value_passed_through(self):
-        output = {"text": "hello"}
-        result = _call(output)
-        self.assertEqual(result["text"], "hello")
-
-    def test_entry_without_filename_unchanged(self):
-        output = {"latent": [{"subfolder": "", "type": "output"}]}
-        result = _call(output)
-        self.assertNotIn("id", result["latent"][0])
-
-    def test_entry_without_type_unchanged(self):
-        output = {"data": [{"filename": "a.png", "subfolder": ""}]}
-        result = _call(output)
-        self.assertNotIn("id", result["data"][0])
-
-    def test_file_not_on_disk_unchanged(self):
-        output = {"images": [{"filename": "missing.png", "subfolder": "", "type": "output"}]}
-        result = _call(output, file_exists=False)
-        self.assertNotIn("id", result["images"][0])
-
-    def test_unknown_type_returns_none_directory_unchanged(self):
-        output = {"images": [{"filename": "a.png", "subfolder": "", "type": "unknown"}]}
-        result = _call(output, directory=None)
-        self.assertNotIn("id", result["images"][0])
-
-    def test_register_injects_only_id(self):
-        reg = _make_register_result(ref_id="inline-ref")
-        output = {"images": [{"filename": "new.png", "subfolder": "", "type": "output"}]}
-        result = _call(output, register_result=reg)
-        img = result["images"][0]
-        self.assertEqual(img["id"], "inline-ref")
-        # Only id is injected — no asset_hash, name, preview_url, size
-        self.assertNotIn("asset_hash", img)
-        self.assertNotIn("name", img)
-        self.assertNotIn("preview_url", img)
-        self.assertNotIn("size", img)
-
-    def test_register_called_per_entry(self):
-        register_mock = MagicMock(return_value=_make_register_result())
-        mocked = _mocked_modules(register_file_in_place=register_mock)
-        output = {
-            "images": [
-                {"filename": "a.png", "subfolder": "", "type": "output"},
-                {"filename": "b.png", "subfolder": "", "type": "output"},
-            ]
-        }
-
-        with patch.dict("sys.modules", mocked), \
-             patch("os.path.isfile", return_value=True):
-            import importlib
-            import comfy_execution.asset_enrichment as mod
-            importlib.reload(mod)
-            mod.enrich_output_with_assets(output)
-
-        self.assertEqual(register_mock.call_count, 2)
-
-    def test_original_entry_not_mutated(self):
-        orig = {"filename": "a.png", "subfolder": "", "type": "output"}
-        output = {"images": [orig]}
-        _call(output)
-        self.assertNotIn("id", orig)
-
-    def test_enrichment_error_does_not_block_sibling_entries(self):
-        call_count = [0]
-        good_reg = _make_register_result(ref_id="good-ref")
-
-        def register_side_effect(abs_path, name, tags):
-            call_count[0] += 1
-            if call_count[0] == 1:
-                raise RuntimeError("boom")
-            return good_reg
-
-        mocked = _mocked_modules(register_file_in_place=register_side_effect)
-
-        output = {
-            "images": [
-                {"filename": "bad.png", "subfolder": "", "type": "output"},
-                {"filename": "good.png", "subfolder": "", "type": "output"},
-            ]
-        }
-
-        with patch.dict("sys.modules", mocked), \
-             patch("os.path.isfile", return_value=True):
-            import importlib
-            import comfy_execution.asset_enrichment as mod
-            importlib.reload(mod)
-            result = mod.enrich_output_with_assets(output)
-
-        imgs = result["images"]
-        self.assertNotIn("id", imgs[0])
-        self.assertEqual(imgs[1]["id"], "good-ref")
-
-    def test_multiple_output_keys_all_enriched(self):
-        output = {
-            "images": [{"filename": "a.png", "subfolder": "", "type": "output"}],
-            "videos": [{"filename": "b.mp4", "subfolder": "", "type": "output"}],
-        }
-        result = _call(output)
-        self.assertIn("id", result["images"][0])
-        self.assertIn("id", result["videos"][0])
-
-    def test_none_entry_in_list_unchanged(self):
-        output = {"images": [None, {"filename": "a.png", "subfolder": "", "type": "output"}]}
-        result = _call(output)
-        self.assertIsNone(result["images"][0])
-        self.assertIn("id", result["images"][1])
-
-    def test_path_traversal_subfolder_skipped(self):
-        register_mock = MagicMock(return_value=_make_register_result())
-        mocked = _mocked_modules(register_file_in_place=register_mock)
-
-        output = {"images": [{"filename": "passwd", "subfolder": "../../etc", "type": "output"}]}
-
-        # Do NOT patch os.path.abspath — real resolution is required for the containment check.
-        with patch.dict("sys.modules", mocked), \
-             patch("os.path.isfile", return_value=True):
-            import importlib
-            import comfy_execution.asset_enrichment as mod
-            importlib.reload(mod)
-            result = mod.enrich_output_with_assets(output)
-
-        self.assertNotIn("id", result["images"][0])
-        register_mock.assert_not_called()
-
-    def test_absolute_filename_skipped(self):
-        register_mock = MagicMock(return_value=_make_register_result())
-        mocked = _mocked_modules(register_file_in_place=register_mock)
-
-        # Absolute filename — os.path.join discards earlier components when a later one is absolute.
-        absolute_filename = os.path.abspath(os.sep + "etc" + os.sep + "passwd")
-        output = {"images": [{"filename": absolute_filename, "subfolder": "", "type": "output"}]}
-
-        with patch.dict("sys.modules", mocked), \
-             patch("os.path.isfile", return_value=True):
-            import importlib
-            import comfy_execution.asset_enrichment as mod
-            importlib.reload(mod)
-            result = mod.enrich_output_with_assets(output)
-
-        self.assertNotIn("id", result["images"][0])
-        register_mock.assert_not_called()
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/execution/test_jobs.py

@@ -1,7 +1,5 @@
 """Unit tests for comfy_execution/jobs.py"""
 
-import pytest
-
 from comfy_execution.jobs import (
     JobStatus,
     is_previewable,
@@ -12,50 +10,9 @@ from comfy_execution.jobs import (
     get_outputs_summary,
     apply_sorting,
     has_3d_extension,
-    validate_job_id,
 )
 
 
-class TestValidateJobId:
-    """validate_job_id guards job creation: POST /prompt rejects ids it raises on."""
-
-    def test_canonical_form_passes_through(self):
-        cid = "a1b2c3d4-e5f6-7a89-b0c1-d2e3f4a5b6c7"
-        assert validate_job_id(cid) == cid
-
-    @pytest.mark.parametrize(
-        "variant",
-        [
-            "A1B2C3D4-E5F6-7A89-B0C1-D2E3F4A5B6C7",          # uppercase
-            "{a1b2c3d4-e5f6-7a89-b0c1-d2e3f4a5b6c7}",        # braced
-            "urn:uuid:a1b2c3d4-e5f6-7a89-b0c1-d2e3f4a5b6c7",  # URN
-            "a1b2c3d4e5f67a89b0c1d2e3f4a5b6c7",              # bare hex
-            " a1b2c3d4-e5f6-7a89-b0c1-d2e3f4a5b6c7 ",        # padded
-        ],
-    )
-    def test_non_canonical_spellings_rejected(self, variant):
-        # uuid.UUID parses all of these, but accepting them would silently
-        # rewrite the client's id (history keys, websocket events, and
-        # /interrupt matching all match the stored form exactly).
-        with pytest.raises(ValueError):
-            validate_job_id(variant)
-
-    @pytest.mark.parametrize(
-        "bad",
-        ["", "not-a-uuid", "prompt-123", "a1b2c3d4-e5f6-7a89-b0c1", "None"],
-    )
-    def test_non_uuid_strings_rejected(self, bad):
-        with pytest.raises(ValueError):
-            validate_job_id(bad)
-
-    @pytest.mark.parametrize("bad", [123, 1.5, True, None, ["a"], {"id": "x"}])
-    def test_non_strings_rejected(self, bad):
-        # uuid.UUID raises AttributeError/TypeError on non-strings; the helper
-        # must normalize those to ValueError so callers need one except clause.
-        with pytest.raises(ValueError):
-            validate_job_id(bad)
-
-
 class TestJobStatus:
     """Test JobStatus constants."""