refactor: share workflow cache helpers

core/workflow/analysis.py

@@ -37,24 +37,41 @@ def _normalize_workflow_analysis_value(value: Any) -> Any:
     return value
 
 
+def _hash_normalized_workflow_value(value: Any) -> str:
+    serialized_value = json.dumps(
+        _normalize_workflow_analysis_value(value),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(serialized_value.encode("utf-8")).hexdigest()
+
+
+def _get_workflow_scope_key(
+    node_id: Any,
+    *,
+    subgraph_id: Any = "",
+    is_top_level: bool,
+) -> tuple[str, str, str]:
+    scope = "top" if is_top_level else "subgraph"
+    return (scope, "" if is_top_level else str(subgraph_id or ""), str(node_id))
+
+
 def _get_workflow_node_cache_key(
     node: Dict[str, Any],
     *,
     subgraph_id: Any = "",
     is_top_level: bool,
 ) -> tuple[str, str, str]:
-    scope = "top" if is_top_level else "subgraph"
-    return (scope, "" if is_top_level else str(subgraph_id or ""), str(node.get("id")))
+    return _get_workflow_scope_key(
+        node.get("id"),
+        subgraph_id=subgraph_id,
+        is_top_level=is_top_level,
+    )
 
 
 def _get_workflow_node_fingerprint(node: Dict[str, Any]) -> str:
-    serialized_node = json.dumps(
-        _normalize_workflow_analysis_value(node),
-        ensure_ascii=False,
-        sort_keys=True,
-        separators=(",", ":"),
-    )
-    return hashlib.sha256(serialized_node.encode("utf-8")).hexdigest()
+    return _hash_normalized_workflow_value(node)
 
 
 def _get_workflow_node_fingerprints(
@@ -176,16 +193,12 @@ def _get_promoted_widget_context_cache_key(
                 _normalize_workflow_analysis_value(node)
             )
 
-    serialized_context = json.dumps(
+    return _hash_normalized_workflow_value(
         {
             "subgraphs": context_subgraphs,
             "instances": context_instances,
         },
-        ensure_ascii=False,
-        sort_keys=True,
-        separators=(",", ":"),
     )
-    return hashlib.sha256(serialized_context.encode("utf-8")).hexdigest()
 
 
 def analyze_workflow_models(

core/workflow/inventory.py

@@ -1,7 +1,5 @@
 """Shared workflow model inventory and incremental cache management."""
 
-import hashlib
-import json
 import threading
 import time
 from collections import OrderedDict
@@ -28,10 +26,10 @@ def _build_workflow_node_cache(
 
     for ref in model_refs:
         is_top_level = ref.get("is_top_level") is not False
-        key = (
-            "top" if is_top_level else "subgraph",
-            "" if is_top_level else str(ref.get("subgraph_id") or ""),
-            str(ref.get("node_id")),
+        key = analysis._get_workflow_scope_key(
+            ref.get("node_id"),
+            subgraph_id=ref.get("subgraph_id"),
+            is_top_level=is_top_level,
         )
         if key in refs_by_node:
             refs_by_node[key].append(ref)
@@ -48,13 +46,7 @@ def _build_workflow_node_cache(
 def _get_workflow_model_inventory_cache_key(
     workflow_json: Dict[str, Any],
 ) -> str:
-    serialized_workflow = json.dumps(
-        analysis._normalize_workflow_analysis_value(workflow_json),
-        ensure_ascii=False,
-        sort_keys=True,
-        separators=(",", ":"),
-    )
-    return hashlib.sha256(serialized_workflow.encode("utf-8")).hexdigest()
+    return analysis._hash_normalized_workflow_value(workflow_json)
 
 
 def _get_analysis_log_context(

tests/test_workflow_components.py

@@ -57,6 +57,30 @@ def test_workflow_node_traversal_preserves_node_order_and_scope():
     assert contexts[3].subgraph_id == "group-1"
 
 
+def test_workflow_cache_helpers_share_scope_keys_and_normalized_hashing():
+    node = {"id": 2, "type": "Inner", "pos": [10, 20]}
+
+    assert analysis._get_workflow_node_cache_key(
+        node,
+        subgraph_id="group-1",
+        is_top_level=False,
+    ) == analysis._get_workflow_scope_key(
+        2,
+        subgraph_id="group-1",
+        is_top_level=False,
+    )
+    assert analysis._get_workflow_node_cache_key(
+        node,
+        is_top_level=True,
+    ) == ("top", "", "2")
+    assert analysis._get_workflow_node_fingerprint(node) == (
+        analysis._hash_normalized_workflow_value(node)
+    )
+    assert analysis._get_workflow_node_fingerprint(
+        {"type": "Inner", "id": 2, "pos": [999, 999]}
+    ) == analysis._get_workflow_node_fingerprint(node)
+
+
 def test_active_workflow_node_traversal_follows_nested_referenced_subgraphs_only():
     workflow = {
         "nodes": [