Include asset id in executed WebSocket messages (#13862)

Amp-Thread-ID: https://ampcode.com/threads/T-019e4ca5-b71a-7168-8f56-58b2325f34c3
Co-authored-by: Amp <amp@ampcode.com>

comfy_execution/asset_enrichment.py

@@ -0,0 +1,73 @@
+"""Enrich executed-node output entries with asset id."""
+import logging
+import os
+
+
+def enrich_output_with_assets(output_ui: dict) -> dict:
+    """Inject asset ``id`` into file-type output entries 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 the WS
+    message from sending.
+    """
+    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
+    from app.assets.database.queries.asset_reference import get_reference_by_file_path
+    from app.database.db import create_session
+
+    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
+
+                # Try DB lookup first (cached node re-send); fall back to registering inline.
+                asset_id = None
+                with create_session() as session:
+                    db_ref = get_reference_by_file_path(session, abs_path)
+                    if db_ref is not None:
+                        asset_id = db_ref.id
+
+                if asset_id is None:
+                    result = register_file_in_place(
+                        abs_path=abs_path,
+                        name=entry["filename"],
+                        tags=[entry["type"]],
+                    )
+                    asset_id = result.ref.id
+
+                entry = dict(entry)
+                entry["id"] = asset_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