ryo_executor/executor/

orchestrator.rs

//! ExecutionOrchestrator: Coordinate multiple Executions for large-scale mutations
//!
//! L3 Orchestration layer that manages multiple L2 Executions, handling:
//! - Speculative parallel execution with result composition
//! - Conflict detection and resolution at ItemRef granularity
//! - Resume points for failure recovery
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────────┐
//! │  L3 Orchestration                                                │
//! │  ┌─────────────────────────────────────────────────────────────┐│
//! │  │  ExecutionOrchestrator                                       ││
//! │  │    ├── strategy: OrchestrationStrategy                       ││
//! │  │    ├── run() → OrchestratedResult                            ││
//! │  │    └── compose_results() → Merge at ItemRef level            ││
//! │  └─────────────────────────────────────────────────────────────┘│
//! │                              │                                   │
//! │                              ▼ spawn executions                  │
//! ├─────────────────────────────────────────────────────────────────┤
//! │  L2 Executions                                                   │
//! │  ┌───────────┐  ┌───────────┐  ┌───────────┐                    │
//! │  │ Execution │  │ Execution │  │ Execution │  ...               │
//! │  │  Group A  │  │  Group B  │  │  Group C  │                    │
//! │  └───────────┘  └───────────┘  └───────────┘                    │
//! └─────────────────────────────────────────────────────────────────┘
//! ```

use super::blueprint::ParallelBlueprint;
use super::blueprint_executor::{BlueprintExecutor, ExecutionStrategy};
use super::conflict;
use super::spec::MutationSpec;
use im::HashMap as ImHashMap;
use rayon::prelude::*;
use ryo_analysis::AnalysisContext;
use ryo_source::pure::{PureFile, ToSynError};
use ryo_symbol::{WorkspaceFilePath, WorkspacePathResolver};
use ryo_verification::{FileChange, PipelineResult, VerificationInput, VerificationPipeline};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

/// Orchestration strategy for coordinating multiple executions
#[derive(Debug, Clone, PartialEq, Default)]
pub enum OrchestrationStrategy {
    /// Execute all specs sequentially in a single execution (safe, debuggable)
    #[default]
    Sequential,

    /// Execute spec groups in parallel, then compose results
    Speculative,

    /// Agent-like execution with tick-based coordination (future)
    Murmuration { tick_budget_ms: u64 },
}

/// Result of orchestrated execution
#[derive(Debug, Clone)]
pub enum OrchestratedResult {
    /// All specs applied successfully
    Success {
        applied: Vec<usize>,
        modified_files: Vec<WorkspaceFilePath>,
        total_changes: usize,
    },

    /// Some specs applied, others had conflicts requiring resolution
    PartialSuccess {
        applied: Vec<usize>,
        conflicts: Vec<ConflictInfo>,
        modified_files: Vec<WorkspaceFilePath>,
        total_changes: usize,
    },

    /// Execution failed
    Error(OrchestratorError),
}

impl OrchestratedResult {
    pub fn is_success(&self) -> bool {
        matches!(self, Self::Success { .. })
    }

    pub fn applied_count(&self) -> usize {
        match self {
            Self::Success { applied, .. } => applied.len(),
            Self::PartialSuccess { applied, .. } => applied.len(),
            Self::Error(_) => 0,
        }
    }
}

/// Error during orchestration
#[derive(Debug, Clone)]
pub struct OrchestratorError {
    pub kind: OrchestratorErrorKind,
    pub message: String,
}

#[derive(Debug, Clone)]
pub enum OrchestratorErrorKind {
    /// Blueprint has unresolved conflicts
    BlueprintConflict,
    /// Execution failed
    ExecutionFailed,
    /// Compose failed due to conflicts
    ComposeFailed,
}

/// Information about a conflict between executions
#[derive(Debug, Clone)]
pub struct ConflictInfo {
    /// File affected by the conflict
    pub file: Option<WorkspaceFilePath>,
    /// Indices of specs that conflicted
    pub spec_indices: Vec<usize>,
    /// Description of the conflict
    pub description: String,
}

/// Result of verified orchestrated execution
///
/// Includes both the orchestration result and the verification pipeline result.
#[derive(Debug)]
pub struct VerifiedOrchestratedResult {
    /// The underlying orchestration result
    pub orchestration: OrchestratedResult,
    /// Verification pipeline result (pre-check + post-check)
    pub verification: Option<PipelineResult>,
    /// Whether the result is fully verified
    pub verified: bool,
}

impl VerifiedOrchestratedResult {
    /// Create a new verified result
    pub fn new(orchestration: OrchestratedResult, verification: PipelineResult) -> Self {
        let verified = verification.is_success();
        Self {
            orchestration,
            verification: Some(verification),
            verified,
        }
    }

    /// Create from orchestration failure (no verification attempted)
    pub fn from_orchestration_failure(orchestration: OrchestratedResult) -> Self {
        Self {
            orchestration,
            verification: None,
            verified: false,
        }
    }

    /// Check if both orchestration and verification succeeded
    pub fn is_success(&self) -> bool {
        self.orchestration.is_success() && self.verified
    }

    /// Get the number of applied specs
    pub fn applied_count(&self) -> usize {
        self.orchestration.applied_count()
    }
}

/// A group of specs that can be executed together
#[derive(Debug, Clone)]
struct SpecGroup {
    /// Indices of specs in this group
    indices: Vec<usize>,
}

/// Result of executing a single group
#[derive(Debug)]
struct GroupResult {
    /// Modified files from this group's execution
    files: ImHashMap<WorkspaceFilePath, Arc<PureFile>>,
    /// Indices of specs that were applied
    applied_indices: Vec<usize>,
    /// Files that were modified
    modified_files: Vec<WorkspaceFilePath>,
    /// Number of changes made
    changes: usize,
}

/// Orchestrator for coordinating multiple executions
pub struct ExecutionOrchestrator {
    specs: Vec<MutationSpec>,
    strategy: OrchestrationStrategy,
}

impl ExecutionOrchestrator {
    pub fn new(specs: Vec<MutationSpec>) -> Self {
        Self {
            specs,
            strategy: OrchestrationStrategy::default(),
        }
    }

    pub fn with_strategy(mut self, strategy: OrchestrationStrategy) -> Self {
        self.strategy = strategy;
        self
    }

    /// Run the orchestrator with the configured strategy
    pub fn run(&self, ctx: &mut AnalysisContext) -> OrchestratedResult {
        if self.specs.is_empty() {
            return OrchestratedResult::Success {
                applied: vec![],
                modified_files: vec![],
                total_changes: 0,
            };
        }

        match &self.strategy {
            OrchestrationStrategy::Sequential => self.run_sequential(ctx),
            OrchestrationStrategy::Speculative => self.run_speculative(ctx),
            OrchestrationStrategy::Murmuration { .. } => {
                // Murmuration falls back to Speculative for now
                self.run_speculative(ctx)
            }
        }
    }

    /// Run speculative execution with verification pipeline.
    ///
    /// This method:
    /// 1. Saves the original file state
    /// 2. Runs the orchestrated execution
    /// 3. If successful, runs verification pipeline (pre-check + post-check)
    /// 4. Returns combined result with verification status
    ///
    /// # Arguments
    /// - `ctx`: The analysis context (will be modified on success)
    /// - `pipeline`: The verification pipeline to use
    ///
    /// # Returns
    /// A `VerifiedOrchestratedResult` containing both orchestration and verification results.
    pub fn run_speculative_verified(
        &self,
        ctx: &mut AnalysisContext,
        pipeline: &VerificationPipeline,
    ) -> Result<VerifiedOrchestratedResult, ToSynError> {
        use ryo_verification::Status;

        // Save original file state for diff calculation
        let original_files = ctx.files.clone();
        let original_sources = self.collect_sources(&original_files)?;

        // Run orchestrated execution
        let orchestration_result = self.run(ctx);

        // If orchestration failed, return without verification
        if !orchestration_result.is_success() {
            return Ok(VerifiedOrchestratedResult::from_orchestration_failure(
                orchestration_result,
            ));
        }

        // Calculate file changes for verification
        let changes =
            FileChange::from_execution_diff(&original_files, &ctx.files, &original_sources)?;

        if changes.is_empty() {
            // No changes to verify
            return Ok(VerifiedOrchestratedResult {
                orchestration: orchestration_result,
                verification: None,
                verified: true,
            });
        }

        // Create verification input
        let resolver = WorkspacePathResolver::new(ctx.workspace_root.to_path_buf());
        let input = VerificationInput::new(changes, resolver);

        // Run pre-check (in-memory graph verification)
        let precheck_result = pipeline.run_precheck(&input, ctx);
        if !precheck_result.is_success() {
            return Ok(VerifiedOrchestratedResult {
                orchestration: orchestration_result,
                verification: Some(precheck_result),
                verified: false,
            });
        }

        // Run post-check (filesystem verification with cargo check)
        let postcheck_result = pipeline.run_postcheck(&input);
        let pipeline_result = match postcheck_result {
            Ok(result) => {
                // Check success before moving results
                let postcheck_success = result.pipeline_result.is_success();

                // Combine precheck and postcheck results
                let mut combined_results = precheck_result.results;
                combined_results.extend(result.pipeline_result.results);

                let overall = if postcheck_success {
                    Status::Passed
                } else {
                    Status::Failed
                };

                PipelineResult {
                    overall,
                    results: combined_results,
                }
            }
            Err(e) => {
                // TempWorkspace creation failed - create error result
                let error_result = ryo_verification::VerificationResult::failure(
                    "postcheck",
                    std::time::Duration::ZERO,
                    vec![ryo_verification::Diagnostic::error(format!(
                        "Post-check failed: {}",
                        e
                    ))],
                );
                PipelineResult {
                    overall: Status::Failed,
                    results: vec![error_result],
                }
            }
        };

        Ok(VerifiedOrchestratedResult::new(
            orchestration_result,
            pipeline_result,
        ))
    }

    /// Collect source strings from files for diff calculation
    fn collect_sources(
        &self,
        files: &ImHashMap<WorkspaceFilePath, Arc<PureFile>>,
    ) -> Result<HashMap<WorkspaceFilePath, String>, ToSynError> {
        files
            .iter()
            .map(|(wfp, pf)| Ok((wfp.clone(), pf.to_source()?)))
            .collect()
    }

    /// Sequential execution: single BlueprintExecutor run
    fn run_sequential(&self, ctx: &mut AnalysisContext) -> OrchestratedResult {
        let blueprint = ParallelBlueprint::from_mutations(self.specs.clone());

        // Check for blueprint-level conflicts
        if blueprint.needs_escalation() {
            return OrchestratedResult::Error(OrchestratorError {
                kind: OrchestratorErrorKind::BlueprintConflict,
                message: format!(
                    "Blueprint has {} conflicts requiring escalation",
                    blueprint.conflicts.len()
                ),
            });
        }

        let executor = BlueprintExecutor::new().with_strategy(ExecutionStrategy::Wavefront);
        let result = executor.execute_v2(&blueprint, ctx);

        if result.success {
            let applied: Vec<usize> = (0..self.specs.len()).collect();
            OrchestratedResult::Success {
                applied,
                modified_files: result.modified_files,
                total_changes: result.total_changes,
            }
        } else {
            OrchestratedResult::Error(OrchestratorError {
                kind: OrchestratorErrorKind::ExecutionFailed,
                message: result.error.unwrap_or_else(|| "Unknown error".to_string()),
            })
        }
    }

    /// Speculative execution: partition specs, run in parallel, compose results
    ///
    /// Uses ItemRef-based conflict detection for fine-grained parallelism.
    fn run_speculative(&self, ctx: &mut AnalysisContext) -> OrchestratedResult {
        // 1. Classify specs into parallel groups and sequential specs
        let (parallel_groups, sequential_indices) = classify_for_parallel_execution(&self.specs);

        // If no parallelizable groups or only one group, fall back to sequential
        if parallel_groups.len() <= 1 && sequential_indices.is_empty() {
            return self.run_sequential(ctx);
        }

        // 2. Convert to SpecGroups for parallel execution
        let groups: Vec<SpecGroup> = parallel_groups
            .into_iter()
            .map(|indices| SpecGroup { indices })
            .collect();

        // If only one group after ItemRef analysis, fall back to sequential
        if groups.len() <= 1 {
            return self.run_sequential(ctx);
        }

        // 3. Execute parallel groups concurrently
        // Clone the context for each group to preserve registry and graphs
        let base_ctx = ctx.fork_clone();
        let group_results: Vec<(SpecGroup, Result<GroupResult, String>)> = groups
            .into_par_iter()
            .map(|group| {
                let result = self.execute_group(&group, &base_ctx);
                (group, result)
            })
            .collect();

        // 4. Compose parallel results
        let mut result = self.compose_results(ctx, group_results);

        // 5. Execute sequential specs after parallel ones complete
        if !sequential_indices.is_empty() {
            let sequential_specs: Vec<MutationSpec> = sequential_indices
                .iter()
                .map(|&i| self.specs[i].clone())
                .collect();

            let sequential_blueprint = ParallelBlueprint::from_mutations(sequential_specs);
            let executor = BlueprintExecutor::new().with_strategy(ExecutionStrategy::Wavefront);
            let seq_result = executor.execute_v2(&sequential_blueprint, ctx);

            // Merge sequential results
            result = match result {
                OrchestratedResult::Success {
                    mut applied,
                    mut modified_files,
                    total_changes,
                } => {
                    if seq_result.success {
                        applied.extend(sequential_indices);
                        modified_files.extend(seq_result.modified_files);
                        OrchestratedResult::Success {
                            applied,
                            modified_files,
                            total_changes: total_changes + seq_result.total_changes,
                        }
                    } else {
                        OrchestratedResult::PartialSuccess {
                            applied,
                            conflicts: vec![ConflictInfo {
                                file: None,
                                spec_indices: sequential_indices,
                                description: seq_result
                                    .error
                                    .unwrap_or_else(|| "Sequential execution failed".to_string()),
                            }],
                            modified_files,
                            total_changes,
                        }
                    }
                }
                OrchestratedResult::PartialSuccess {
                    mut applied,
                    mut conflicts,
                    mut modified_files,
                    total_changes,
                } => {
                    if seq_result.success {
                        applied.extend(sequential_indices);
                        modified_files.extend(seq_result.modified_files);
                    } else {
                        conflicts.push(ConflictInfo {
                            file: None,
                            spec_indices: sequential_indices,
                            description: seq_result
                                .error
                                .unwrap_or_else(|| "Sequential execution failed".to_string()),
                        });
                    }
                    OrchestratedResult::PartialSuccess {
                        applied,
                        conflicts,
                        modified_files,
                        total_changes: total_changes + seq_result.total_changes,
                    }
                }
                err @ OrchestratedResult::Error(_) => err,
            };
        }

        result
    }

    /// Execute a single group of specs on a cloned context
    fn execute_group(
        &self,
        group: &SpecGroup,
        base_ctx: &AnalysisContext,
    ) -> Result<GroupResult, String> {
        // Clone the context with full registry and graphs for this group
        let mut ctx = base_ctx.fork_clone();

        // Collect specs for this group
        let group_specs: Vec<MutationSpec> = group
            .indices
            .iter()
            .map(|&idx| self.specs[idx].clone())
            .collect();

        let blueprint = ParallelBlueprint::from_mutations(group_specs);

        if blueprint.needs_escalation() {
            return Err(format!("Group {:?} has conflicts", group.indices));
        }

        let executor = BlueprintExecutor::new().with_strategy(ExecutionStrategy::Wavefront);
        let result = executor.execute_v2(&blueprint, &mut ctx);

        if result.success {
            Ok(GroupResult {
                files: ctx.files,
                applied_indices: group.indices.clone(),
                modified_files: result.modified_files,
                changes: result.total_changes,
            })
        } else {
            Err(result.error.unwrap_or_else(|| "Unknown error".to_string()))
        }
    }

    /// Compose results from multiple group executions into the context
    fn compose_results(
        &self,
        ctx: &mut AnalysisContext,
        group_results: Vec<(SpecGroup, Result<GroupResult, String>)>,
    ) -> OrchestratedResult {
        let mut all_applied: Vec<usize> = Vec::new();
        let mut all_modified: HashSet<WorkspaceFilePath> = HashSet::new();
        let mut total_changes: usize = 0;
        let mut conflicts: Vec<ConflictInfo> = Vec::new();

        // Track which files were modified by which groups
        let mut file_to_groups: HashMap<WorkspaceFilePath, Vec<usize>> = HashMap::new();

        // First pass: collect successful results and detect file-level conflicts
        let mut successful_results: Vec<(usize, GroupResult)> = Vec::new();

        for (group_idx, (group, result)) in group_results.into_iter().enumerate() {
            match result {
                Ok(group_result) => {
                    // Track file modifications for conflict detection
                    for file in &group_result.modified_files {
                        file_to_groups
                            .entry(file.clone())
                            .or_default()
                            .push(group_idx);
                    }
                    successful_results.push((group_idx, group_result));
                }
                Err(err) => {
                    conflicts.push(ConflictInfo {
                        file: None,
                        spec_indices: group.indices,
                        description: err,
                    });
                }
            }
        }

        // Detect file-level conflicts (multiple groups modifying same file)
        let conflicting_files: HashSet<WorkspaceFilePath> = file_to_groups
            .iter()
            .filter(|(_, groups)| groups.len() > 1)
            .map(|(key, _)| key.clone())
            .collect();

        // Apply non-conflicting results
        for (_group_idx, group_result) in successful_results {
            let has_conflict = group_result
                .modified_files
                .iter()
                .any(|f| conflicting_files.contains(f));

            if has_conflict {
                // Record conflict for later resolution
                let conflicting: Vec<WorkspaceFilePath> = group_result
                    .modified_files
                    .iter()
                    .filter(|f| conflicting_files.contains(*f))
                    .cloned()
                    .collect();

                for file_path in conflicting {
                    if !conflicts
                        .iter()
                        .any(|c| c.file.as_ref() == Some(&file_path))
                    {
                        conflicts.push(ConflictInfo {
                            file: Some(file_path.clone()),
                            spec_indices: group_result.applied_indices.clone(),
                            description: format!("Multiple groups modified file: {:?}", file_path),
                        });
                    }
                }
            } else {
                // Apply this group's changes to ctx
                for file_path in &group_result.modified_files {
                    if let Some(pure_file) = group_result.files.get(file_path) {
                        ctx.files_mut().insert(file_path.clone(), pure_file.clone());
                    }
                    all_modified.insert(file_path.clone());
                }
                all_applied.extend(group_result.applied_indices);
                total_changes += group_result.changes;
            }
        }

        if conflicts.is_empty() {
            OrchestratedResult::Success {
                applied: all_applied,
                modified_files: all_modified.into_iter().collect(),
                total_changes,
            }
        } else if !all_applied.is_empty() {
            OrchestratedResult::PartialSuccess {
                applied: all_applied,
                conflicts,
                modified_files: all_modified.into_iter().collect(),
                total_changes,
            }
        } else {
            OrchestratedResult::Error(OrchestratorError {
                kind: OrchestratorErrorKind::ComposeFailed,
                message: "All groups had conflicts".to_string(),
            })
        }
    }
}

/// Extract a grouping key from a target string
/// Suggest the best orchestration strategy based on specs and context
pub fn suggest_orchestration(
    specs: &[MutationSpec],
    _ctx: &AnalysisContext,
) -> OrchestrationStrategy {
    let spec_count = specs.len();

    // Estimate number of independent groups using conflict detection
    let groups = conflict::group_by_conflicts(specs);
    let estimated_groups = groups.len();

    // Simple conflict density estimation based on target overlap
    let conflict_density = if spec_count == 0 {
        0.0
    } else {
        1.0 - (estimated_groups as f64 / spec_count as f64)
    };

    // Strategy selection (exclusive, evaluated top to bottom)
    match (spec_count, estimated_groups, conflict_density) {
        // 1. Few specs → Sequential
        (n, _, _) if n <= 5 => OrchestrationStrategy::Sequential,

        // 2. High conflict → Sequential (safety first)
        (_, _, d) if d >= 0.5 => OrchestrationStrategy::Sequential,

        // 3. Good independence & low conflict → Speculative
        (_, g, d) if g >= 3 && d < 0.15 => OrchestrationStrategy::Speculative,

        // 4. Otherwise → Murmuration (fallback to Speculative for now)
        _ => OrchestrationStrategy::Murmuration { tick_budget_ms: 10 },
    }
}

// ============================================================================
// Public API: Re-export from conflict module
// ============================================================================

/// Partition specs into conflict-free groups.
///
/// Uses Union-Find to efficiently group specs that transitively conflict.
/// Renamed from `partition_by_item_refs` but now uses MutationTargetSymbol-based detection.
pub use conflict::group_by_conflicts as partition_by_item_refs;

/// Classify specs into parallelizable groups (DEPRECATED).
///
/// This function is deprecated and will be removed. Use `conflict::group_by_conflicts` directly.
/// The old logic depended on `item_refs()` which doesn't work with lazy MutationTargetSymbol.
///
/// For now, this function simply calls `group_by_conflicts` for all specs.
pub fn classify_for_parallel_execution(specs: &[MutationSpec]) -> (Vec<Vec<usize>>, Vec<usize>) {
    // All specs are grouped by conflicts now
    // No special "sequential" handling needed - conflict detection handles project-wide mutations
    let groups = conflict::group_by_conflicts(specs);
    (groups, vec![])
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::executor::conflict::{find_conflicting_pairs, specs_conflict};
    use crate::executor::spec::{InsertPosition, MutationTargetSymbol, SymbolPath};
    use ryo_analysis::testing::{ContextBuilder, ContextTestExt};
    use ryo_symbol::SymbolId;

    fn create_multi_file_context() -> AnalysisContext {
        ContextBuilder::new()
            .with_file("src/lib.rs", "// lib\n")
            .with_file("src/models.rs", "struct User {}\n")
            .with_file("src/api.rs", "struct Api {}\n")
            .build()
    }

    /// Create a dummy SymbolId for testing (index starts from 1)
    fn dummy_id(index: u32) -> SymbolId {
        SymbolId::parse(&format!("{}v1", index)).expect("valid dummy id")
    }

    #[test]
    fn test_orchestrator_sequential_basic() {
        let mut ctx = create_multi_file_context();
        let user_id = ctx.registry().lookup_by_name("User").unwrap();

        let specs = vec![MutationSpec::AddDerive {
            target: MutationTargetSymbol::ById(user_id),
            derives: vec!["Debug".to_string()],
        }];

        let orchestrator =
            ExecutionOrchestrator::new(specs).with_strategy(OrchestrationStrategy::Sequential);

        let result = orchestrator.run(&mut ctx);

        assert!(result.is_success(), "Sequential execution failed");
        assert_eq!(result.applied_count(), 1);
    }

    #[test]
    fn test_orchestrator_speculative_independent_groups() {
        let mut ctx = create_multi_file_context();
        let user_id = ctx.registry().lookup_by_name("User").unwrap();
        let api_id = ctx.registry().lookup_by_name("Api").unwrap();

        // Specs targeting different structs (should be parallelizable)
        let specs = vec![
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(user_id),
                derives: vec!["Debug".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(api_id),
                derives: vec!["Clone".to_string()],
            },
        ];

        let orchestrator =
            ExecutionOrchestrator::new(specs).with_strategy(OrchestrationStrategy::Speculative);

        let result = orchestrator.run(&mut ctx);

        assert!(result.is_success(), "Speculative execution failed");
        assert_eq!(result.applied_count(), 2);
    }

    #[test]
    fn test_orchestrator_empty_specs() {
        let mut ctx = create_multi_file_context();
        let specs: Vec<MutationSpec> = vec![];

        let orchestrator = ExecutionOrchestrator::new(specs);
        let result = orchestrator.run(&mut ctx);

        assert!(result.is_success());
        assert_eq!(result.applied_count(), 0);
    }

    #[test]
    fn test_suggest_orchestration_few_specs() {
        let ctx = create_multi_file_context();

        let specs = vec![MutationSpec::AddDerive {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            derives: vec!["Debug".to_string()],
        }];

        let strategy = suggest_orchestration(&specs, &ctx);
        assert_eq!(strategy, OrchestrationStrategy::Sequential);
    }

    #[test]
    fn test_suggest_orchestration_many_independent() {
        let ctx = create_multi_file_context();

        // Many specs with different targets (use dummy IDs since we're testing orchestration logic)
        let specs = vec![
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(1)),
                derives: vec!["Debug".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(2)),
                derives: vec!["Clone".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(3)),
                derives: vec!["Default".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(4)),
                derives: vec!["Hash".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(5)),
                derives: vec!["Default".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(4)),
                derives: vec!["Hash".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(5)),
                derives: vec!["Eq".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(6)),
                derives: vec!["PartialEq".to_string()],
            },
        ];

        let strategy = suggest_orchestration(&specs, &ctx);
        // With 8 specs and some conflicts (id4 and id5 appear twice each),
        // we get 6 groups: [1], [2], [3], [4,4], [5,5], [6]
        // conflict_density = 1.0 - 6/8 = 0.25 (>= 0.15)
        // So Murmuration is returned (not Speculative which requires d < 0.15)
        assert_eq!(
            strategy,
            OrchestrationStrategy::Murmuration { tick_budget_ms: 10 }
        );
    }

    #[test]
    #[ignore = "V1 path disabled - needs V2 migration"]
    fn test_orchestrator_with_add_items() {
        let mut ctx = create_multi_file_context();

        let specs = vec![
            MutationSpec::AddItem {
                target: MutationTargetSymbol::ByPath(Box::new(
                    SymbolPath::parse("test_crate::models").unwrap(),
                )),
                content: "pub struct Order {}".to_string(),
                position: InsertPosition::Bottom,
            },
            MutationSpec::AddItem {
                target: MutationTargetSymbol::ByPath(Box::new(
                    SymbolPath::parse("test_crate::api").unwrap(),
                )),
                content: "pub fn handler() {}".to_string(),
                position: InsertPosition::Bottom,
            },
        ];

        // Use Sequential strategy as Speculative has known issues with module-based ItemRefs
        let orchestrator =
            ExecutionOrchestrator::new(specs).with_strategy(OrchestrationStrategy::Sequential);

        let result = orchestrator.run(&mut ctx);

        assert!(result.is_success(), "Speculative with AddItem failed");

        let models = ctx.test_file("src/models.rs").unwrap().to_source().unwrap();
        assert!(models.contains("Order"), "Order not added to models");

        let api = ctx.test_file("src/api.rs").unwrap().to_source().unwrap();
        assert!(api.contains("handler"), "handler not added to api");
    }

    #[test]
    #[ignore = "V1 path disabled - needs V2 migration"]
    fn test_orchestrator_verifies_changes_applied() {
        let mut ctx = create_multi_file_context();
        let user_id = ctx.registry().lookup_by_name("User").unwrap();
        let api_id = ctx.registry().lookup_by_name("Api").unwrap();

        let specs = vec![
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(user_id),
                derives: vec!["Debug".to_string(), "Clone".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(api_id),
                derives: vec!["Default".to_string()],
            },
        ];

        let orchestrator =
            ExecutionOrchestrator::new(specs).with_strategy(OrchestrationStrategy::Sequential);

        let result = orchestrator.run(&mut ctx);
        assert!(result.is_success());

        // Verify changes were applied to ctx
        let models = ctx.test_file("src/models.rs").unwrap().to_source().unwrap();
        assert!(models.contains("Debug"), "Debug not added to User");
        assert!(models.contains("Clone"), "Clone not added to User");

        let api = ctx.test_file("src/api.rs").unwrap().to_source().unwrap();
        assert!(api.contains("Default"), "Default not added to Api");
    }

    // ========================================================================
    // ItemRef-based Conflict Detection Tests
    // ========================================================================

    #[test]
    fn test_specs_conflict_different_fields_no_conflict() {
        // AddField to same struct but different fields → NO conflict
        let spec_a = MutationSpec::AddField {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            field_name: "name".to_string(),
            field_type: "String".to_string(),
            visibility: Default::default(),
        };
        let spec_b = MutationSpec::AddField {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            field_name: "email".to_string(),
            field_type: "String".to_string(),
            visibility: Default::default(),
        };

        // Different fields should NOT conflict - can run in parallel
        // AddField only references the field path, not the struct
        let conflicts = specs_conflict(&spec_a, &spec_b);
        assert!(
            !conflicts,
            "AddField to different fields should NOT conflict"
        );
    }

    #[test]
    fn test_specs_conflict_same_field_conflict() {
        // AddField and RemoveField on same field → CONFLICT
        let spec_add = MutationSpec::AddField {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            field_name: "email".to_string(),
            field_type: "String".to_string(),
            visibility: Default::default(),
        };
        let spec_remove = MutationSpec::RemoveField {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            field_name: "email".to_string(),
        };

        assert!(
            specs_conflict(&spec_add, &spec_remove),
            "Same field operations should conflict"
        );
    }

    #[test]
    fn test_specs_conflict_different_structs_no_conflict() {
        // AddField to different structs → no conflict
        let spec_a = MutationSpec::AddField {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            field_name: "name".to_string(),
            field_type: "String".to_string(),
            visibility: Default::default(),
        };
        let spec_b = MutationSpec::AddField {
            target: MutationTargetSymbol::ById(dummy_id(2)),
            field_name: "id".to_string(),
            field_type: "u64".to_string(),
            visibility: Default::default(),
        };

        assert!(
            !specs_conflict(&spec_a, &spec_b),
            "Different structs should not conflict"
        );
    }

    #[test]
    fn test_specs_conflict_parent_child_conflict() {
        use crate::executor::ItemKind;

        // RemoveItem on User vs AddField to User → conflict (parent-child)
        let spec_remove = MutationSpec::RemoveItem {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            item_kind: ItemKind::Struct,
        };
        let spec_add = MutationSpec::AddField {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            field_name: "email".to_string(),
            field_type: "String".to_string(),
            visibility: Default::default(),
        };

        assert!(
            specs_conflict(&spec_remove, &spec_add),
            "Remove parent should conflict with add child"
        );
    }

    #[test]
    fn test_partition_by_item_refs_independent_groups() {
        // Three specs targeting different structs → three groups
        let specs = vec![
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(1)),
                derives: vec!["Debug".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(2)),
                derives: vec!["Clone".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(3)),
                derives: vec!["Default".to_string()],
            },
        ];

        let groups = partition_by_item_refs(&specs);
        assert_eq!(
            groups.len(),
            3,
            "Three independent specs should form three groups"
        );
    }

    #[test]
    fn test_partition_by_item_refs_conflicting_merged() {
        // Two specs on same struct → one group
        let specs = vec![
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(1)),
                derives: vec!["Debug".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(1)),
                derives: vec!["Clone".to_string()],
            },
        ];

        let groups = partition_by_item_refs(&specs);
        assert_eq!(groups.len(), 1, "Conflicting specs should be in same group");
        assert_eq!(groups[0].len(), 2);
    }

    #[test]
    fn test_classify_for_parallel_execution() {
        use ryo_symbol::{SymbolKind, SymbolPath, SymbolRegistry};

        // Create dummy symbol IDs for testing
        let mut symbol_registry = SymbolRegistry::new();
        let path_user = SymbolPath::parse("test_crate::User").unwrap();
        let path_order = SymbolPath::parse("test_crate::Order").unwrap();
        let path_old = SymbolPath::parse("test_crate::old_name").unwrap();
        let symbol_user = symbol_registry
            .register(path_user, SymbolKind::Struct)
            .unwrap();
        let symbol_order = symbol_registry
            .register(path_order, SymbolKind::Struct)
            .unwrap();
        let symbol_old = symbol_registry
            .register(path_old, SymbolKind::Function)
            .unwrap();

        // All three specs have item_refs (different targets) → all parallel
        let specs = vec![
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(symbol_user),
                derives: vec!["Debug".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(symbol_order),
                derives: vec!["Clone".to_string()],
            },
            // Rename with "from" generates ItemRef for crate::old_name
            MutationSpec::Rename {
                target: MutationTargetSymbol::ById(symbol_old),
                to: "new_name".to_string(),
                scope: Default::default(),
            },
        ];

        let (parallel, sequential) = classify_for_parallel_execution(&specs);

        // User, Order, and old_name are all independent → 3 parallel groups
        assert_eq!(parallel.len(), 3, "Should have 3 parallel groups");

        // No specs are sequential (all have item_refs)
        assert_eq!(sequential.len(), 0, "No specs should be sequential");
    }

    #[test]
    fn test_find_conflicting_pairs() {
        let specs = vec![
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(1)),
                derives: vec!["Debug".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(1)),
                derives: vec!["Clone".to_string()],
            },
            MutationSpec::AddDerive {
                target: MutationTargetSymbol::ById(dummy_id(2)),
                derives: vec!["Default".to_string()],
            },
        ];

        let conflicts = find_conflicting_pairs(&specs);

        // specs[0] and specs[1] conflict (same target: User)
        assert!(conflicts.contains(&(0, 1)), "User specs should conflict");
        // specs[0]/specs[1] and specs[2] don't conflict
        assert!(
            !conflicts.contains(&(0, 2)),
            "User and Order should not conflict"
        );
        assert!(
            !conflicts.contains(&(1, 2)),
            "User and Order should not conflict"
        );
    }

    // ========================================================================
    // Verified Orchestration Tests
    // ========================================================================

    #[test]
    fn test_verified_orchestrated_result_success() {
        let orch_result = OrchestratedResult::Success {
            applied: vec![0, 1],
            modified_files: vec![],
            total_changes: 2,
        };

        let pipeline_result = PipelineResult {
            overall: ryo_verification::Status::Passed,
            results: vec![],
        };

        let verified = VerifiedOrchestratedResult::new(orch_result, pipeline_result);

        assert!(verified.is_success());
        assert!(verified.verified);
        assert_eq!(verified.applied_count(), 2);
    }

    #[test]
    fn test_verified_orchestrated_result_orchestration_failure() {
        let orch_result = OrchestratedResult::Error(OrchestratorError {
            kind: OrchestratorErrorKind::ExecutionFailed,
            message: "Test error".to_string(),
        });

        let verified = VerifiedOrchestratedResult::from_orchestration_failure(orch_result);

        assert!(!verified.is_success());
        assert!(!verified.verified);
        assert!(verified.verification.is_none());
        assert_eq!(verified.applied_count(), 0);
    }

    #[test]
    fn test_verified_orchestrated_result_verification_failure() {
        let orch_result = OrchestratedResult::Success {
            applied: vec![0],
            modified_files: vec![],
            total_changes: 1,
        };

        let pipeline_result = PipelineResult {
            overall: ryo_verification::Status::Failed,
            results: vec![ryo_verification::VerificationResult::failure(
                "test",
                std::time::Duration::ZERO,
                vec![ryo_verification::Diagnostic::error("Test error")],
            )],
        };

        let verified = VerifiedOrchestratedResult::new(orch_result, pipeline_result);

        assert!(!verified.is_success());
        assert!(!verified.verified);
        assert!(verified.orchestration.is_success());
    }

    #[test]
    fn test_run_speculative_verified_empty_specs() {
        let mut ctx = create_multi_file_context();

        let orchestrator = ExecutionOrchestrator::new(vec![]);
        let pipeline = VerificationPipeline::new();

        let result = orchestrator
            .run_speculative_verified(&mut ctx, &pipeline)
            .unwrap();

        // Empty specs should succeed trivially
        assert!(result.is_success());
        assert!(result.orchestration.is_success());
        assert!(result.verified);
    }

    #[test]
    fn test_run_speculative_verified_basic() {
        let mut ctx = create_multi_file_context();

        let specs = vec![MutationSpec::AddDerive {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            derives: vec!["Debug".to_string()],
        }];

        let orchestrator =
            ExecutionOrchestrator::new(specs).with_strategy(OrchestrationStrategy::Sequential);

        // Use empty pipeline (no verifiers)
        let pipeline = VerificationPipeline::new();

        let result = orchestrator
            .run_speculative_verified(&mut ctx, &pipeline)
            .unwrap();

        // Orchestration should succeed
        assert!(result.orchestration.is_success());
        // With no verifiers, verification passes trivially
        assert!(result.verified);
    }

    #[test]
    fn test_run_speculative_verified_with_graph_verifier() {
        use ryo_verification::GraphVerifier;

        let mut ctx = create_multi_file_context();

        let specs = vec![MutationSpec::AddDerive {
            target: MutationTargetSymbol::ById(dummy_id(1)),
            derives: vec!["Debug".to_string()],
        }];

        let orchestrator =
            ExecutionOrchestrator::new(specs).with_strategy(OrchestrationStrategy::Sequential);

        // Pipeline with GraphVerifier only (fast pre-check)
        let pipeline = VerificationPipeline::new().add_in_memory(GraphVerifier::new());

        let result = orchestrator
            .run_speculative_verified(&mut ctx, &pipeline)
            .unwrap();

        // Both orchestration and pre-check should succeed
        assert!(result.orchestration.is_success());
        // Verification status depends on pre-check result
        assert!(result.verification.is_some() || result.verified);
    }
}