diff --git a/core/resolver.py b/core/resolver.py index 970769f..7467940 100644 --- a/core/resolver.py +++ b/core/resolver.py @@ -883,6 +883,7 @@ def analyze_and_find_matches( max_matches_per_model: int = 10, progress_callback: Optional[Callable[[Dict[str, Any]], None]] = None, force_rescan: bool = False, + analysis_id: Optional[str] = None, ) -> Dict[str, Any]: """ Main entry point: analyze workflow and find matches for missing models. @@ -892,6 +893,7 @@ def analyze_and_find_matches( similarity_threshold: Minimum similarity score (0.0 to 1.0) for matches max_matches_per_model: Maximum number of matches to return per missing model force_rescan: If True, bypass the short-lived local model scan cache + analysis_id: Optional identifier used to correlate logs for one request Returns: Dictionary with analysis results: @@ -953,10 +955,16 @@ def analyze_and_find_matches( } # Extract URLs from workflow (node.properties.models + regex) + analysis_context = f" (analysis_id={analysis_id})" if analysis_id else "" workflow_urls = extract_workflow_urls(workflow_json) - log.debug(f"Extracted {len(workflow_urls)} URLs from workflow") + log.debug( + f"Extracted {len(workflow_urls)} URLs from workflow{analysis_context}" + ) workflow_hashes = extract_workflow_hash_metadata(workflow_json) - log.debug(f"Extracted {len(workflow_hashes)} workflow hash metadata keys") + log.debug( + f"Extracted {len(workflow_hashes)} workflow hash metadata keys" + f"{analysis_context}" + ) if progress_callback: progress_callback( @@ -972,6 +980,7 @@ def analyze_and_find_matches( workflow_json, force_rescan=force_rescan, progress_callback=progress_callback, + analysis_id=analysis_id, ) available_models = inventory["available_models"] all_model_refs = inventory["model_refs"] diff --git a/core/services/workflow_service.py b/core/services/workflow_service.py index ffae90f..5465427 100644 --- a/core/services/workflow_service.py +++ b/core/services/workflow_service.py @@ -80,6 +80,7 @@ class WorkflowService: 10, update_analysis_progress if analysis_id else None, force_rescan=force_rescan, + analysis_id=analysis_id, ) missing_models = result.get("missing_models", []) diff --git a/core/workflow/analysis.py b/core/workflow/analysis.py index 7b26142..c25c997 100644 --- a/core/workflow/analysis.py +++ b/core/workflow/analysis.py @@ -219,6 +219,7 @@ def analyze_workflow_models( Dict[tuple[str, str, str], Dict[str, Any]] ] = None, analysis_stats: Optional[Dict[str, int]] = None, + analysis_context: Optional[str] = None, ) -> List[Dict[str, Any]]: """ Extract all model references from a workflow, including nested subgraphs. @@ -235,6 +236,7 @@ def analyze_workflow_models( all_model_refs = [] reused_nodes = 0 analyzed_nodes = 0 + log_context = f" ({analysis_context})" if analysis_context else "" # Get subgraph definitions first to check if node types are subgraph UUIDs definitions = workflow_json.get("definitions", {}) @@ -392,6 +394,7 @@ def analyze_workflow_models( ) analyzed_subgraph_nodes = 0 + reused_subgraph_nodes = 0 for node in subgraph_nodes: try: @@ -413,6 +416,7 @@ def analyze_workflow_models( if reused: base_model_refs = cached_node.get("refs", []) reused_nodes += 1 + reused_subgraph_nodes += 1 else: base_model_refs = references.get_node_model_info( node, @@ -465,10 +469,11 @@ def analyze_workflow_models( ) continue - if analyzed_subgraph_nodes: + if subgraph_nodes: log.debug( f"Analyzing subgraph: {subgraph_name} (ID: {subgraph_id}) " - f"with {analyzed_subgraph_nodes} changed nodes" + f"with {analyzed_subgraph_nodes} changed nodes, " + f"reused {reused_subgraph_nodes} nodes{log_context}" ) if analysis_stats is not None: @@ -483,7 +488,7 @@ def analyze_workflow_models( if previous_node_cache is not None: log.debug( f"Incremental workflow analysis: analyzed {analyzed_nodes} nodes, " - f"reused {reused_nodes} nodes" + f"reused {reused_nodes} nodes{log_context}" ) return all_model_refs diff --git a/core/workflow/inventory.py b/core/workflow/inventory.py index 8395fb7..0f7c8e1 100644 --- a/core/workflow/inventory.py +++ b/core/workflow/inventory.py @@ -57,6 +57,16 @@ def _get_workflow_model_inventory_cache_key( return hashlib.sha256(serialized_workflow.encode("utf-8")).hexdigest() +def _get_analysis_log_context( + cache_key: str, + analysis_id: Optional[str], +) -> str: + normalized_analysis_id = str(analysis_id or "").strip() + if normalized_analysis_id: + return f"analysis_id={normalized_analysis_id}" + return f"workflow_signature={cache_key[:12]}" + + def invalidate_workflow_model_inventory_cache() -> None: """Clear shared workflow model analysis snapshots.""" with _WORKFLOW_MODEL_INVENTORY_CACHE_LOCK: @@ -68,6 +78,7 @@ def get_workflow_model_inventory( *, force_rescan: bool = False, progress_callback: Optional[Callable[[Dict[str, Any]], None]] = None, + analysis_id: Optional[str] = None, ) -> Dict[str, List[Dict[str, Any]]]: """ Return the shared base model inventory for a workflow. @@ -79,6 +90,7 @@ def get_workflow_model_inventory( from ..scanner import get_model_files cache_key = _get_workflow_model_inventory_cache_key(workflow_json) + analysis_context = _get_analysis_log_context(cache_key, analysis_id) promoted_context_key = analysis._get_promoted_widget_context_cache_key(workflow_json) current_node_fingerprints = analysis._get_workflow_node_fingerprints(workflow_json) now = time.monotonic() @@ -131,7 +143,9 @@ def get_workflow_model_inventory( "cached": True, } ) - log.debug("Reusing shared workflow model analysis") + log.debug( + f"Reusing shared workflow model analysis ({analysis_context})" + ) return { "available_models": cached["available_models"], "model_refs": model_refs, @@ -148,6 +162,7 @@ def get_workflow_model_inventory( previous_node_cache=previous_inventory.get("node_cache", {}), node_cache_out=node_cache, analysis_stats=analysis_stats, + analysis_context=analysis_context, ) else: available_models = get_model_files(force_rescan=force_rescan) @@ -155,6 +170,7 @@ def get_workflow_model_inventory( workflow_json, available_models=available_models, progress_callback=progress_callback, + analysis_context=analysis_context, ) node_cache = _build_workflow_node_cache(workflow_json, model_refs) analysis_stats = { diff --git a/tests/test_workflow_analyzer.py b/tests/test_workflow_analyzer.py index 4a28a25..a5659b0 100644 --- a/tests/test_workflow_analyzer.py +++ b/tests/test_workflow_analyzer.py @@ -2,7 +2,7 @@ import os import tempfile import unittest from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import ANY, patch from core import resolver as resolver_core from core.scanner import invalidate_model_files_cache, scan_directory @@ -1370,6 +1370,12 @@ class WorkflowModelInventoryCacheTests(unittest.TestCase): workflow, available_models=available_models, progress_callback=None, + analysis_context=ANY, + ) + self.assertTrue( + analyze_models.call_args.kwargs["analysis_context"].startswith( + "workflow_signature=" + ) ) def test_force_rescan_rebuilds_shared_inventory(self): @@ -1672,11 +1678,31 @@ class WorkflowModelInventoryCacheTests(unittest.TestCase): "get_node_model_info", wraps=references.get_node_model_info, ) as get_node_info, + patch.object(analysis.log, "debug") as debug_log, ): - get_workflow_model_inventory(build_workflow("first.safetensors")) - get_workflow_model_inventory(build_workflow("second.safetensors")) + get_workflow_model_inventory( + build_workflow("first.safetensors"), + analysis_id="analysis-1", + ) + get_workflow_model_inventory( + build_workflow("second.safetensors"), + analysis_id="analysis-2", + ) self.assertEqual(4, get_node_info.call_count) + debug_messages = [ + call.args[0] + for call in debug_log.call_args_list + if call.args + ] + self.assertTrue( + any( + "Analyzing subgraph: Model subgraph (ID: subgraph-1) " + "with 1 changed nodes, reused 1 nodes (analysis_id=analysis-2)" + in message + for message in debug_messages + ) + ) def test_scanner_invalidation_clears_shared_inventory(self): workflow = _workflow_with_model("invalidated.safetensors") diff --git a/tests/test_workflow_service.py b/tests/test_workflow_service.py index cf9971f..e71411a 100644 --- a/tests/test_workflow_service.py +++ b/tests/test_workflow_service.py @@ -194,6 +194,9 @@ async def test_analyze_route_filters_references_and_uses_local_download_sources( assert sources["popular.safetensors"]["size"] == 200 assert sources["listed.safetensors"]["source"] == "model_list" values["invalidate_local_hash_match_cache"].assert_called_once_with() + assert values["analyze_and_find_matches"].call_args.kwargs["analysis_id"] == ( + "analysis-1" + ) values["self"]._update_analysis_progress.assert_any_call( "analysis-1", {"stage": "matching", "current": 1, "total": 5},