cursus.validation.script_testing.script_execution_registry

Script Execution Registry

Central state coordinator for DAG execution with sequential message passing. This module implements the Script Execution Registry as designed in the comprehensive architecture document, providing integration between script_dependency_matcher and script_input_resolver layers.

Key Features: - Sequential state updates via topological ordering - Message passing between script executions - Runtime data tracking (inputs, outputs, status) - Integration coordination between layers - Six clear integration points for layer coordination

class ScriptExecutionRegistry(dag, step_catalog=None)

Bases: object

Central state coordinator that integrates both layers:

Layer 1 (script_dependency_matcher): DAG-level orchestration Layer 2 (script_input_resolver): Script-level resolution

Registry Role: - Maintains DAG execution state - Coordinates message passing between layers - Provides state interface for both layers - Ensures sequential consistency

initialize_from_dependency_matcher(prepared_data)

Integration Point 1: Receive prepared data from script_dependency_matcher.

script_dependency_matcher calls this to initialize registry state.

Parameters:

prepared_data (Dict[str, Any]) – Output from prepare_script_testing_inputs() containing: - node_specs: Loaded step specifications - dependency_matches: Automatic dependency matches - config_data: Extracted configuration data - execution_order: Topological sort order

get_dependency_outputs_for_node(node_name)

Integration Point 2: Provide dependency outputs for message passing.

script_dependency_matcher calls this to get outputs from completed dependencies.

Parameters:

node_name (str) – Name of the node requesting dependency outputs

Returns:

Dictionary mapping output names to actual paths from completed dependencies

Return type:

Dict[str, str]

get_node_config_for_resolver(node_name)

Integration Point 3: Provide node config to script_input_resolver.

script_input_resolver calls this to get base configuration for a node.

Parameters:

node_name (str) – Name of the node requesting configuration

Returns:

Dictionary containing node configuration data

Return type:

Dict[str, Any]

store_resolved_inputs(node_name, resolved_inputs)

Integration Point 4: Store resolved inputs from script_input_resolver.

script_input_resolver calls this to store its resolution results.

Parameters:

• node_name (str) – Name of the node

• resolved_inputs (Dict[str, Any]) – Dictionary containing resolved input configuration

get_ready_node_inputs(node_name)

Integration Point 5: Provide complete inputs for script execution.

API layer calls this to get final inputs for script execution.

Parameters:

node_name (str) – Name of the node requesting inputs

Returns:

Dictionary containing complete input configuration for script execution

Return type:

Dict[str, Any]

commit_execution_results(node_name, execution_result)

Integration Point 6: Store execution results for message passing.

API layer calls this after script execution to update state.

Parameters:

• node_name (str) – Name of the executed node

• execution_result (ScriptTestResult) – Result of script execution

sequential_state_update()

Generator that yields nodes in topological order with updated state.

This ensures: - Dependencies are processed before dependents - State updates are sequential and consistent - Message passing happens in correct order

Yields:

Tuple of (node_name, updated_node_state) for each node in execution order

get_execution_summary()

Get summary of execution state and message passing.

Returns:

Dictionary with execution summary information

Return type:

Dict[str, Any]

get_message_passing_history()

Get complete message passing history for analysis.

Returns:

List of message passing events with timestamps

Return type:

List[Dict[str, Any]]

get_node_status(node_name)

Get current status of a specific node.

Parameters:

node_name (str) – Name of the node

Returns:

Current status (‘pending’, ‘ready’, ‘completed’, ‘failed’)

Return type:

str

get_node_outputs(node_name)

Get outputs produced by a specific node.

Parameters:

node_name (str) – Name of the node

Returns:

Dictionary of outputs produced by the node

Return type:

Dict[str, str]

clear_registry()

Clear registry for testing or new execution.

Resets all state while preserving DAG and step catalog references.

class DAGStateConsistency

Bases: object

Ensures state consistency during sequential message passing.

Guarantees: 1. Dependencies are always processed before dependents (topological order) 2. Node state is only updated when all dependencies are completed 3. Message passing only uses outputs from completed nodes 4. State updates are atomic and consistent

static validate_execution_order(dag, execution_order)

Validate that execution order respects dependency constraints.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance

• execution_order (List[str]) – List of node names in execution order

Raises:

ValueError – If execution order violates dependency constraints

static ensure_state_consistency(registry, node_name)

Ensure node state is consistent before execution.

Parameters:

• registry (ScriptExecutionRegistry) – ScriptExecutionRegistry instance

• node_name (str) – Name of the node to validate

Raises:

RuntimeError – If state is inconsistent

create_script_execution_registry(dag, step_catalog=None)

Factory function to create a ScriptExecutionRegistry instance.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance defining the pipeline structure

• step_catalog (StepCatalog | None) – Optional StepCatalog instance (will create if not provided)

Returns:

Configured ScriptExecutionRegistry instance

Return type:

ScriptExecutionRegistry