Source code for cursus.validation.script_testing.script_dependency_matcher
"""
Two-Phase Script Dependency Resolution System
This module implements intelligent dependency resolution for script testing by directly
reusing pipeline assembler patterns and dependency resolution algorithms from cursus/core.
The two-phase architecture:
1. Phase 1 (Prepare): Automatic dependency analysis using pipeline assembler patterns
2. Phase 2 (User Input): Interactive collection with auto-resolution and user override capability
Key Features:
- Direct reuse of PipelineAssembler._propagate_messages() algorithm
- Direct reuse of UnifiedDependencyResolver._calculate_compatibility()
- Direct reuse of step catalog specification loading
- Direct reuse of config-based extraction functions
- User override capability for auto-resolved paths
- 60-70% reduction in manual path specifications
"""
import logging
from typing import Dict, Any
from pathlib import Path
# DIRECT REUSE: Import existing cursus infrastructure
from ...api.dag.base_dag import PipelineDAG
from ...step_catalog import StepCatalog
from ...core.deps.factory import create_dependency_resolver
from ...steps.configs.utils import load_configs, build_complete_config_classes
from .api import collect_script_inputs
logger = logging.getLogger(__name__)
[docs]
def prepare_script_testing_inputs(
dag: PipelineDAG, # DIRECT REUSE
config_path: str,
step_catalog: StepCatalog, # DIRECT REUSE
) -> Dict[str, Any]:
"""
Phase 1: Automatic dependency analysis using pipeline assembler patterns.
DIRECT REUSE of PipelineAssembler._propagate_messages() algorithm.
Args:
dag: PipelineDAG defining script execution order and dependencies
config_path: Path to configuration file for config-based extraction
step_catalog: For loading specifications and contracts
Returns:
Prepared data structure containing dependency matches and config data
"""
logger.info("Phase 1: Analyzing dependencies and preparing input collection...")
# 1. Load specifications for all DAG nodes (DIRECT REUSE of step catalog patterns)
node_specs = {}
for node_name in dag.nodes:
try:
spec = step_catalog.load_spec_class(
node_name
) # YAML-first via StepCatalog.load_spec_class
if spec:
node_specs[node_name] = spec
logger.debug(
f"Loaded specification for {node_name}: {len(spec.dependencies)} deps, {len(spec.outputs)} outputs"
)
else:
logger.warning(f"No specification found for {node_name}")
except Exception as e:
logger.warning(f"Failed to load specification for {node_name}: {e}")
# 2. Dependency matching (DIRECT REUSE of pipeline assembler algorithm)
dependency_resolver = create_dependency_resolver() # DIRECT REUSE
dependency_matches = {}
logger.info(
f"Analyzing dependencies for {len(node_specs)} nodes with specifications..."
)
for consumer_node in dag.nodes:
if consumer_node not in node_specs:
continue
consumer_spec = node_specs[consumer_node]
matches = {}
# For each dependency in consumer specification
for dep_name, dep_spec in consumer_spec.dependencies.items():
best_match = None
best_score = 0.0
# Check all potential provider nodes
for provider_node in dag.nodes:
if provider_node == consumer_node or provider_node not in node_specs:
continue
provider_spec = node_specs[provider_node]
# Check each output of provider (same as pipeline assembler)
for output_name, output_spec in provider_spec.outputs.items():
try:
# DIRECT REUSE: Same compatibility calculation as pipeline assembler
score = dependency_resolver._calculate_compatibility(
dep_spec, output_spec, provider_spec
)
if (
score > best_score and score > 0.5
): # Same threshold as pipeline
best_match = {
"provider_node": provider_node,
"provider_output": output_name,
"compatibility_score": score,
"match_type": "specification_match",
}
best_score = score
except Exception as e:
logger.debug(
f"Compatibility calculation failed for {consumer_node}.{dep_name} <- {provider_node}.{output_name}: {e}"
)
if best_match:
matches[dep_name] = best_match
logger.info(
f"Matched {consumer_node}.{dep_name} -> {best_match['provider_node']}.{best_match['provider_output']} (score: {best_score:.2f})"
)
dependency_matches[consumer_node] = matches
# 3. Config extraction (DIRECT REUSE of existing functions)
config_data = {}
try:
config_classes = build_complete_config_classes() # DIRECT REUSE
all_configs = load_configs(config_path, config_classes) # DIRECT REUSE
for node_name in dag.nodes:
if node_name in all_configs:
config = all_configs[node_name]
config_data[node_name] = collect_script_inputs(config) # DIRECT REUSE
logger.debug(f"Extracted config data for {node_name}")
else:
logger.warning(f"No config found for {node_name}")
except Exception as e:
logger.error(f"Config extraction failed: {e}")
# Continue with empty config data
# Summary logging
total_matches = sum(len(matches) for matches in dependency_matches.values())
logger.info(
f"Phase 1 complete: Found {total_matches} automatic dependency matches across {len(dependency_matches)} nodes"
)
return {
"node_specs": node_specs,
"dependency_matches": dependency_matches,
"config_data": config_data,
"execution_order": dag.topological_sort(), # DIRECT REUSE
}
[docs]
def collect_user_inputs_with_dependency_resolution(
prepared_data: Dict[str, Any],
) -> Dict[str, Any]:
"""
Phase 2: Interactive input collection with automatic dependency resolution.
Mirrors PipelineAssembler._propagate_messages() + StepBuilder._get_inputs() patterns.
Args:
prepared_data: Output from prepare_script_testing_inputs()
Returns:
Complete user inputs ready for script execution
"""
execution_order = prepared_data["execution_order"]
dependency_matches = prepared_data["dependency_matches"]
node_specs = prepared_data["node_specs"]
config_data = prepared_data["config_data"]
# Track outputs from completed scripts (like pipeline assembler's step_messages)
completed_outputs = {} # {node_name: {logical_name: actual_path}}
all_user_inputs = {}
print(f"\n🔧 Script Testing Input Collection")
print(f" Processing {len(execution_order)} scripts in dependency order...")
for node_name in execution_order:
print(f"\n📝 Script: {node_name}")
# 1. Start with config-based data (job args, env vars, script path)
script_config = config_data.get(node_name, {})
# 2. Auto-resolve input dependencies (like pipeline assembler message passing)
resolved_inputs = {}
unresolved_inputs = []
if node_name in node_specs:
spec = node_specs[node_name]
matches = dependency_matches.get(node_name, {})
for dep_name, dep_spec in spec.dependencies.items():
if dep_name in matches:
# Automatic resolution from previous script (same as pipeline message passing)
match = matches[dep_name]
provider_node = match["provider_node"]
provider_output = match["provider_output"]
if (
provider_node in completed_outputs
and provider_output in completed_outputs[provider_node]
):
actual_path = completed_outputs[provider_node][provider_output]
resolved_inputs[dep_name] = actual_path
print(f" 🔗 Auto-resolved {dep_name} = {actual_path}")
print(
f" Source: {provider_node}.{provider_output} (compatibility: {match['compatibility_score']:.2f})"
)
else:
unresolved_inputs.append(dep_name)
logger.debug(
f"Provider {provider_node} output {provider_output} not yet available for {node_name}.{dep_name}"
)
else:
unresolved_inputs.append(dep_name)
# 3. User input for unresolved dependencies AND allow override of auto-resolved inputs
user_input_paths = {}
# First: Ask for unresolved inputs (required)
if unresolved_inputs:
print(f" 📥 Please provide input paths:")
for dep_name in unresolved_inputs:
while True:
path = input(f" {dep_name}: ").strip()
if path:
# Basic validation - check if path exists
if (
Path(path).exists()
or path.startswith("/")
or path.startswith("./")
):
user_input_paths[dep_name] = path
break
else:
print(
f" ⚠️ Path may not exist: {path}. Continue anyway? (y/n): ",
end="",
)
confirm = input().strip().lower()
if confirm in ["y", "yes"]:
user_input_paths[dep_name] = path
break
else:
print(f" ⚠️ Input path required for {dep_name}")
# Second: Allow user to override auto-resolved inputs (optional)
if resolved_inputs:
print(
f" 🔄 Auto-resolved inputs (press Enter to keep, or provide new path to override):"
)
for dep_name, auto_path in resolved_inputs.items():
override_path = input(f" {dep_name} [{auto_path}]: ").strip()
if override_path:
user_input_paths[dep_name] = override_path
print(f" ✏️ Overridden: {dep_name} = {override_path}")
# 4. Combine resolved and user-provided inputs (user inputs take precedence)
final_input_paths = {**resolved_inputs, **user_input_paths}
# 5. User input for output paths (always required)
output_paths = {}
if node_name in node_specs:
spec = node_specs[node_name]
if spec.outputs:
print(f" 📤 Please provide output paths:")
for output_name, output_spec in spec.outputs.items():
while True:
path = input(f" {output_name}: ").strip()
if path:
output_paths[output_name] = path
break
else:
print(f" ⚠️ Output path required for {output_name}")
# 6. Store complete input configuration
all_user_inputs[node_name] = {
"input_paths": final_input_paths,
"output_paths": output_paths,
"environment_variables": script_config.get("environment_variables", {}),
"job_arguments": script_config.get("job_arguments", {}),
"script_path": script_config.get("script_path"),
}
# 7. Register outputs for next scripts (like pipeline assembler's step_messages)
completed_outputs[node_name] = output_paths
print(
f" ✅ Configured {node_name} with {len(final_input_paths)} inputs, {len(output_paths)} outputs"
)
return all_user_inputs
[docs]
def resolve_script_dependencies(
dag: PipelineDAG, config_path: str, step_catalog: StepCatalog, registry=None
) -> Dict[str, Any]:
"""
SIMPLIFIED: Two-phase dependency resolution with optional registry integration.
When registry is provided, uses registry coordination for state management.
When registry is None, uses legacy standalone mode for backward compatibility.
Args:
dag: PipelineDAG defining script execution order and dependencies
config_path: Path to configuration file for config-based extraction
step_catalog: For loading specifications and contracts
registry: Optional ScriptExecutionRegistry for state coordination
Returns:
Complete user inputs ready for script execution
"""
try:
# Phase 1: Prepare (automatic dependency analysis)
print("🔄 Phase 1: Analyzing dependencies and preparing input collection...")
prepared_data = prepare_script_testing_inputs(dag, config_path, step_catalog)
# Initialize registry if provided (INTEGRATION POINT 1)
if registry:
registry.initialize_from_dependency_matcher(prepared_data)
total_matches = sum(
len(matches) for matches in prepared_data["dependency_matches"].values()
)
print(f" Found {total_matches} automatic dependency matches")
# Phase 2: User input collection (registry-aware or legacy)
print(
"🔄 Phase 2: Collecting user inputs with automatic dependency resolution..."
)
if registry:
user_inputs = collect_user_inputs_with_registry_coordination(
prepared_data, registry
)
else:
user_inputs = collect_user_inputs_with_dependency_resolution(prepared_data)
print(f"\n✅ Input collection complete! Configured {len(user_inputs)} scripts.")
# Summary of automation benefits
total_inputs = sum(
len(inputs["input_paths"]) for inputs in user_inputs.values()
)
auto_resolved = sum(
len(matches) for matches in prepared_data["dependency_matches"].values()
)
if total_inputs > 0:
automation_percentage = (auto_resolved / total_inputs) * 100
print(
f"📊 Automation Summary: {auto_resolved}/{total_inputs} inputs auto-resolved ({automation_percentage:.1f}%)"
)
return user_inputs
except Exception as e:
logger.error(f"Dependency resolution failed: {e}")
raise RuntimeError(f"Failed to resolve script dependencies: {e}") from e
[docs]
def resolve_script_dependencies_with_registry(
dag: PipelineDAG, config_path: str, step_catalog: StepCatalog, registry
) -> Dict[str, Any]:
"""
SIMPLIFIED: Registry-coordinated dependency resolution (delegates to main function).
This is now just a convenience wrapper that calls the main function with registry.
Args:
dag: PipelineDAG defining script execution order and dependencies
config_path: Path to configuration file for config-based extraction
step_catalog: For loading specifications and contracts
registry: ScriptExecutionRegistry instance for state coordination
Returns:
Complete user inputs ready for script execution
"""
return resolve_script_dependencies(dag, config_path, step_catalog, registry)
[docs]
def validate_dependency_resolution_result(user_inputs: Dict[str, Any]) -> bool:
"""
Validate the result of dependency resolution.
Args:
user_inputs: Result from resolve_script_dependencies()
Returns:
True if validation passes, False otherwise
"""
try:
for node_name, node_inputs in user_inputs.items():
# Check required fields
required_fields = [
"input_paths",
"output_paths",
"environment_variables",
"job_arguments",
]
for field in required_fields:
if field not in node_inputs:
logger.error(
f"Missing required field '{field}' for node {node_name}"
)
return False
# Check script path if available
script_path = node_inputs.get("script_path")
if script_path and not Path(script_path).exists():
logger.warning(
f"Script path does not exist for {node_name}: {script_path}"
)
logger.info(
f"Dependency resolution validation passed for {len(user_inputs)} nodes"
)
return True
except Exception as e:
logger.error(f"Dependency resolution validation failed: {e}")
return False
[docs]
def get_dependency_resolution_summary(
prepared_data: Dict[str, Any], user_inputs: Dict[str, Any]
) -> Dict[str, Any]:
"""
Generate a summary of the dependency resolution process.
Args:
prepared_data: Output from prepare_script_testing_inputs()
user_inputs: Output from collect_user_inputs_with_dependency_resolution()
Returns:
Summary statistics and information
"""
dependency_matches = prepared_data["dependency_matches"]
total_nodes = len(user_inputs)
total_dependencies = sum(
len(inputs["input_paths"]) for inputs in user_inputs.values()
)
auto_resolved_dependencies = sum(
len(matches) for matches in dependency_matches.values()
)
manual_dependencies = total_dependencies - auto_resolved_dependencies
automation_rate = (
(auto_resolved_dependencies / total_dependencies * 100)
if total_dependencies > 0
else 0
)
return {
"total_nodes": total_nodes,
"total_dependencies": total_dependencies,
"auto_resolved_dependencies": auto_resolved_dependencies,
"manual_dependencies": manual_dependencies,
"automation_rate_percentage": automation_rate,
"nodes_with_auto_resolution": len(
[node for node, matches in dependency_matches.items() if matches]
),
"dependency_matches": dependency_matches,
}
[docs]
def collect_user_inputs_with_registry_coordination(
prepared_data: Dict[str, Any], registry
) -> Dict[str, Any]:
"""
SIMPLIFIED: Registry-coordinated Phase 2 input collection.
This function coordinates with the ScriptExecutionRegistry to enable
message passing and state management during input collection.
Args:
prepared_data: Output from prepare_script_testing_inputs()
registry: ScriptExecutionRegistry instance for coordination
Returns:
Complete user inputs ready for script execution
"""
execution_order = prepared_data["execution_order"]
dependency_matches = prepared_data["dependency_matches"]
node_specs = prepared_data["node_specs"]
config_data = prepared_data["config_data"]
all_user_inputs = {}
print(f"\n🔧 Registry-Coordinated Script Testing Input Collection")
print(f" Processing {len(execution_order)} scripts in dependency order...")
for node_name in execution_order:
print(f"\n📝 Script: {node_name}")
# INTEGRATION POINT 2: Get dependency outputs from registry
dependency_outputs = registry.get_dependency_outputs_for_node(node_name)
# 1. Start with config-based data (job args, env vars, script path)
script_config = config_data.get(node_name, {})
# 2. SIMPLIFIED: Auto-resolve using registry message passing
resolved_inputs = {}
unresolved_inputs = []
if node_name in node_specs:
spec = node_specs[node_name]
matches = dependency_matches.get(node_name, {})
for dep_name, dep_spec in spec.dependencies.items():
if dep_name in matches:
# Check if dependency output is available from registry
match = matches[dep_name]
provider_node = match["provider_node"]
provider_output = match["provider_output"]
# Try direct mapping first
if provider_output in dependency_outputs:
actual_path = dependency_outputs[provider_output]
resolved_inputs[dep_name] = actual_path
print(f" 🔗 Auto-resolved {dep_name} = {actual_path}")
print(
f" Source: {provider_node}.{provider_output} (compatibility: {match['compatibility_score']:.2f})"
)
# Try prefixed mapping
elif f"{provider_node}_{provider_output}" in dependency_outputs:
actual_path = dependency_outputs[
f"{provider_node}_{provider_output}"
]
resolved_inputs[dep_name] = actual_path
print(f" 🔗 Auto-resolved {dep_name} = {actual_path}")
print(
f" Source: {provider_node}.{provider_output} (prefixed, compatibility: {match['compatibility_score']:.2f})"
)
else:
unresolved_inputs.append(dep_name)
logger.debug(
f"Provider {provider_node} output {provider_output} not yet available for {node_name}.{dep_name}"
)
else:
unresolved_inputs.append(dep_name)
# 3. User input for unresolved dependencies AND allow override of auto-resolved inputs
user_input_paths = {}
# First: Ask for unresolved inputs (required)
if unresolved_inputs:
print(f" 📥 Please provide input paths:")
for dep_name in unresolved_inputs:
while True:
path = input(f" {dep_name}: ").strip()
if path:
# Basic validation - check if path exists
if (
Path(path).exists()
or path.startswith("/")
or path.startswith("./")
):
user_input_paths[dep_name] = path
break
else:
print(
f" ⚠️ Path may not exist: {path}. Continue anyway? (y/n): ",
end="",
)
confirm = input().strip().lower()
if confirm in ["y", "yes"]:
user_input_paths[dep_name] = path
break
else:
print(f" ⚠️ Input path required for {dep_name}")
# Second: Allow user to override auto-resolved inputs (optional)
if resolved_inputs:
print(
f" 🔄 Auto-resolved inputs (press Enter to keep, or provide new path to override):"
)
for dep_name, auto_path in resolved_inputs.items():
override_path = input(f" {dep_name} [{auto_path}]: ").strip()
if override_path:
user_input_paths[dep_name] = override_path
print(f" ✏️ Overridden: {dep_name} = {override_path}")
# 4. Combine resolved and user-provided inputs (user inputs take precedence)
final_input_paths = {**resolved_inputs, **user_input_paths}
# 5. User input for output paths (always required)
output_paths = {}
if node_name in node_specs:
spec = node_specs[node_name]
if spec.outputs:
print(f" 📤 Please provide output paths:")
for output_name, output_spec in spec.outputs.items():
while True:
path = input(f" {output_name}: ").strip()
if path:
output_paths[output_name] = path
break
else:
print(f" ⚠️ Output path required for {output_name}")
# 6. Store complete input configuration
node_inputs = {
"input_paths": final_input_paths,
"output_paths": output_paths,
"environment_variables": script_config.get("environment_variables", {}),
"job_arguments": script_config.get("job_arguments", {}),
"script_path": script_config.get("script_path"),
}
all_user_inputs[node_name] = node_inputs
# INTEGRATION POINT 4: Store resolved inputs in registry
registry.store_resolved_inputs(node_name, node_inputs)
print(
f" ✅ Configured {node_name} with {len(final_input_paths)} inputs, {len(output_paths)} outputs"
)
return all_user_inputs
[docs]
def collect_manual_inputs_with_registry(
dag: PipelineDAG, config_path: str, step_catalog: StepCatalog, registry
) -> Dict[str, Any]:
"""
REGISTRY-ONLY: Manual input collection through registry pattern.
This function provides the same functionality as ScriptTestingInputCollector
but works entirely through the registry pattern, eliminating the need for
dynamic imports.
Args:
dag: PipelineDAG defining script execution order and dependencies
config_path: Path to configuration file for config-based extraction
step_catalog: For loading specifications and contracts
registry: ScriptExecutionRegistry instance for state coordination
Returns:
Complete user inputs ready for script execution
"""
try:
logger.info(
"Using registry-coordinated manual input collection (no dependency resolution)"
)
# Phase 1: Prepare with empty dependency matches (manual mode)
prepared_data = {
"node_specs": {},
"dependency_matches": {}, # Empty - no automatic dependency resolution
"config_data": {},
"execution_order": dag.topological_sort(),
}
# Load config data for all nodes
try:
config_classes = build_complete_config_classes()
all_configs = load_configs(config_path, config_classes)
for node_name in dag.nodes:
if node_name in all_configs:
config = all_configs[node_name]
prepared_data["config_data"][node_name] = collect_script_inputs(
config
)
logger.debug(f"Extracted config data for {node_name}")
except Exception as e:
logger.error(f"Config extraction failed: {e}")
# Initialize registry with prepared data
registry.initialize_from_dependency_matcher(prepared_data)
# Phase 2: Manual input collection (no automatic dependency resolution)
execution_order = prepared_data["execution_order"]
config_data = prepared_data["config_data"]
all_user_inputs = {}
print(f"\n🔧 Registry-Coordinated Manual Input Collection")
print(
f" Processing {len(execution_order)} scripts (no automatic dependency resolution)..."
)
for node_name in execution_order:
print(f"\n📝 Script: {node_name}")
# 1. Start with config-based data (job args, env vars, script path)
script_config = config_data.get(node_name, {})
# 2. Manual input collection - user provides all input paths
input_paths = {}
print(f" 📥 Please provide input paths:")
# For manual mode, we ask for common input types
common_inputs = ["data_input", "model_input", "config_input"]
for input_name in common_inputs:
path = input(f" {input_name} (optional): ").strip()
if path:
input_paths[input_name] = path
# Allow additional custom inputs
while True:
custom_input = input(
f" Additional input name (or press Enter to continue): "
).strip()
if not custom_input:
break
path = input(f" {custom_input}: ").strip()
if path:
input_paths[custom_input] = path
# 3. Manual output collection - user provides all output paths
output_paths = {}
print(f" 📤 Please provide output paths:")
# For manual mode, we ask for common output types
common_outputs = ["data_output", "model_output", "metrics_output"]
for output_name in common_outputs:
path = input(f" {output_name} (optional): ").strip()
if path:
output_paths[output_name] = path
# Allow additional custom outputs
while True:
custom_output = input(
f" Additional output name (or press Enter to continue): "
).strip()
if not custom_output:
break
path = input(f" {custom_output}: ").strip()
if path:
output_paths[custom_output] = path
# 4. Store complete input configuration
node_inputs = {
"input_paths": input_paths,
"output_paths": output_paths,
"environment_variables": script_config.get("environment_variables", {}),
"job_arguments": script_config.get("job_arguments", {}),
"script_path": script_config.get("script_path"),
}
all_user_inputs[node_name] = node_inputs
# INTEGRATION POINT 4: Store resolved inputs in registry
registry.store_resolved_inputs(node_name, node_inputs)
print(
f" ✅ Configured {node_name} with {len(input_paths)} inputs, {len(output_paths)} outputs"
)
logger.info(
f"Manual input collection complete! Configured {len(all_user_inputs)} scripts."
)
return all_user_inputs
except Exception as e:
logger.error(f"Manual input collection failed: {e}")
raise RuntimeError(f"Failed to collect manual inputs: {e}") from e