swarm_engine_eval/

runner.rs

//! EvalRunner - Orchestrator を直接使用した評価実行

use std::time::Duration;

use tokio::runtime::Handle;

use std::path::PathBuf;

use swarm_engine_core::actions::ActionDef;
use swarm_engine_core::agent::{
    BatchInvoker, DefaultBatchManagerAgent, GenericWorker, ManagementStrategy, ManagerAgent,
    ManagerId, WorkerAgent,
};
use swarm_engine_core::environment::EnvironmentBox;
use swarm_engine_core::events::{
    ActionEventPublisher, JsonlWriter, LearningEventChannel, LearningLifecycleHook,
};
use swarm_engine_core::exploration::{LearnedDependencyProvider, NodeRules, OperatorProvider};
use swarm_engine_core::extensions::Extensions;
use swarm_engine_core::learn::{
    CountTrigger, LearningSnapshot, LearningStore, OfflineModel, SnapshotMetadata, TrainTrigger,
};
use swarm_engine_core::orchestrator::{OrchestratorBuilder, SwarmConfig};
use swarm_engine_core::types::{GroupId, SwarmTask};

use crate::environments::{
    CodeEnvironment, DeepSearchEnvironment, InternalDiagnosisEnvironment, MazeEnvironment,
    SearchEnvironment, TroubleshootingEnvironment,
};

use crate::aggregator::Aggregator;
use crate::error::Result;
use crate::metrics::RunMetrics;
use crate::reporter::{ConfigSummary, EvalReport, SeedInfo};
use crate::run::{EvalRun, TerminationReason};
use crate::scenario::conditions::{ConditionValue, TimeoutBehavior};
use crate::scenario::{EvalScenario, ManagementStrategyConfig};

/// Evaluation seed for reproducibility
///
/// Stored in Extensions to allow Environment and other components to access it.
#[derive(Debug, Clone, Copy)]
pub struct EvalSeed(pub u64);

/// Factory for creating ManagerAgent
pub type ManagerFactory = Box<dyn Fn() -> Box<dyn ManagerAgent> + Send + Sync>;

/// Factory for creating BatchInvoker
pub type BatchInvokerFactory = Box<dyn Fn() -> Box<dyn BatchInvoker> + Send + Sync>;

/// Factory for creating OperatorProvider
pub type OperatorProviderFactory =
    Box<dyn Fn() -> Box<dyn OperatorProvider<NodeRules>> + Send + Sync>;

/// Evaluation runner using Orchestrator directly
///
/// # Example
///
/// ```ignore
/// let runner = EvalRunner::new(scenario, runtime.handle().clone())
///     .with_runs(5)
///     .with_seed(42)
///     .with_task(SwarmTask::new("Find the auth handler"))
///     .with_manager_factory(|| Box::new(MyManager::new()))
///     .with_batch_invoker_factory(|| Box::new(MyInvoker::new()));
///
/// let report = runner.run()?;
/// ```
pub struct EvalRunner {
    scenario: EvalScenario,
    runtime: Handle,
    runs: usize,
    seed: u64,
    /// Task to execute (optional)
    task: Option<SwarmTask>,
    /// Manager factory (creates new instance per run)
    manager_factory: Option<ManagerFactory>,
    /// BatchInvoker factory (creates new instance per run)
    batch_invoker_factory: Option<BatchInvokerFactory>,
    /// Extensions factory (creates new instance per run)
    extensions_factory: Option<Box<dyn Fn() -> Extensions + Send + Sync>>,
    /// OperatorProvider factory (creates new instance per run)
    operator_provider_factory: Option<OperatorProviderFactory>,
    /// Verbose output (print tick snapshots)
    verbose: bool,
    /// Enable ExplorationSpaceV2 tracking
    enable_exploration: bool,
    /// Dependency graph for action sequencing
    dependency_graph: Option<swarm_engine_core::exploration::DependencyGraph>,
    /// Path to save action events (JSONL format)
    action_events_path: Option<PathBuf>,
    /// LearningStore for cross-session learning
    learning_store: Option<LearningStore>,
    /// Prior learning snapshot to load before each run
    prior_snapshot: Option<LearningSnapshot>,
    /// Offline model for optimized parameters
    offline_model: Option<OfflineModel>,
    /// Shared eval count for LifecycleHook (tracks total runs across iterations)
    eval_count: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    /// TrainTrigger for Learning (default: run after every eval)
    train_trigger: Option<std::sync::Arc<dyn TrainTrigger>>,
}

impl EvalRunner {
    pub fn new(scenario: EvalScenario, runtime: Handle) -> Self {
        Self {
            scenario,
            runtime,
            runs: 1,
            seed: 42,
            task: None,
            manager_factory: None,
            batch_invoker_factory: None,
            extensions_factory: None,
            operator_provider_factory: None,
            verbose: false,
            enable_exploration: false,
            dependency_graph: None,
            action_events_path: None,
            learning_store: None,
            prior_snapshot: None,
            offline_model: None,
            eval_count: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            train_trigger: None,
        }
    }

    /// Enable verbose output (print tick snapshots)
    pub fn with_verbose(mut self, verbose: bool) -> Self {
        self.verbose = verbose;
        self
    }

    /// Enable ExplorationSpace tracking
    pub fn with_exploration(mut self, enable: bool) -> Self {
        self.enable_exploration = enable;
        self
    }

    /// Set dependency graph for action sequencing
    ///
    /// The graph defines valid action transitions and terminal conditions.
    /// When set, actions will be filtered based on the graph structure.
    pub fn with_dependency_graph(
        mut self,
        graph: swarm_engine_core::exploration::DependencyGraph,
    ) -> Self {
        self.dependency_graph = Some(graph);
        self
    }

    /// Set path to save action events (JSONL format)
    ///
    /// When set, all action events will be saved to the specified path.
    /// Each run will append to a file with the run index suffix.
    ///
    /// # Example
    ///
    /// ```ignore
    /// runner.with_action_events_path("./logs/events.jsonl")
    /// // Creates: ./logs/events_run0.jsonl, ./logs/events_run1.jsonl, ...
    /// ```
    pub fn with_action_events_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.action_events_path = Some(path.into());
        self
    }

    /// Enable LearningStore for cross-session learning
    ///
    /// When enabled, statistics will be saved after each run and loaded as prior
    /// for subsequent runs. This enables incremental learning across sessions.
    ///
    /// # Example
    ///
    /// ```ignore
    /// runner.with_learning_store("~/.swarm-engine/learning")
    /// ```
    pub fn with_learning_store(mut self, path: impl AsRef<std::path::Path>) -> Self {
        match LearningStore::new(path) {
            Ok(store) => {
                // Load prior snapshot for this scenario
                // Use learning_key() for consistent directory naming (e.g., "troubleshooting" from "user:troubleshooting:v2")
                let scenario_key = self.scenario.meta.id.learning_key();
                self.prior_snapshot = store.load_scenario(&scenario_key).ok();
                // Load offline model for optimized parameters
                self.offline_model = store.load_offline_model(&scenario_key).ok();
                if let Some(ref model) = self.offline_model {
                    tracing::debug!(
                        ucb1_c = model.parameters.ucb1_c,
                        strategy = %model.strategy_config.initial_strategy,
                        action_order = model.action_order.is_some(),
                        "Offline model loaded"
                    );
                    println!(
                        "Offline model loaded: ucb1_c={:.3}, strategy={}",
                        model.parameters.ucb1_c, model.strategy_config.initial_strategy
                    );
                }
                self.learning_store = Some(store);

                // Enable LearningEventChannel for recording strategy advice events
                LearningEventChannel::global().enable();
            }
            Err(e) => {
                eprintln!("Warning: Failed to create LearningStore: {}", e);
            }
        }
        self
    }

    /// Set TrainTrigger for Learning
    ///
    /// Controls when offline learning is executed after eval runs.
    /// Default: Learn after every run (if learning_store is configured).
    ///
    /// # Example
    ///
    /// ```ignore
    /// use swarm_engine_core::learn::CountTrigger;
    ///
    /// // Run learning after every 5 eval iterations
    /// runner.with_train_trigger(Arc::new(CountTrigger::new(5)))
    /// ```
    pub fn with_train_trigger(mut self, trigger: std::sync::Arc<dyn TrainTrigger>) -> Self {
        self.train_trigger = Some(trigger);
        self
    }

    pub fn with_runs(mut self, runs: usize) -> Self {
        self.runs = runs;
        self
    }

    pub fn with_seed(mut self, seed: u64) -> Self {
        self.seed = seed;
        self
    }

    /// Set the task to execute
    pub fn with_task(mut self, task: SwarmTask) -> Self {
        self.task = Some(task);
        self
    }

    /// Set manager factory (creates new Manager for each run)
    pub fn with_manager_factory<F>(mut self, factory: F) -> Self
    where
        F: Fn() -> Box<dyn ManagerAgent> + Send + Sync + 'static,
    {
        self.manager_factory = Some(Box::new(factory));
        self
    }

    /// Set batch invoker factory (creates new BatchInvoker for each run)
    pub fn with_batch_invoker_factory<F>(mut self, factory: F) -> Self
    where
        F: Fn() -> Box<dyn BatchInvoker> + Send + Sync + 'static,
    {
        self.batch_invoker_factory = Some(Box::new(factory));
        self
    }

    /// Set extensions factory (creates new Extensions for each run)
    pub fn with_extensions_factory<F>(mut self, factory: F) -> Self
    where
        F: Fn() -> Extensions + Send + Sync + 'static,
    {
        self.extensions_factory = Some(Box::new(factory));
        self
    }

    /// Set OperatorProvider factory (creates new provider for each run)
    ///
    /// Use this to configure the Selection strategy for exploration.
    /// Default is `AdaptiveProvider` if not specified.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use swarm_engine_core::exploration::{HybridLlmProvider, ReviewPolicy};
    /// use swarm_engine_llm::LlmStrategyAdvisor;
    ///
    /// runner.with_operator_provider_factory(|| {
    ///     let advisor = LlmStrategyAdvisor::new(decider.clone(), handle.clone());
    ///     let policy = ReviewPolicy::default();
    ///     Box::new(HybridLlmProvider::new(advisor, policy))
    /// })
    /// ```
    pub fn with_operator_provider_factory<F>(mut self, factory: F) -> Self
    where
        F: Fn() -> Box<dyn OperatorProvider<NodeRules>> + Send + Sync + 'static,
    {
        self.operator_provider_factory = Some(Box::new(factory));
        self
    }

    pub fn run(&self) -> Result<EvalReport> {
        let mut eval_runs = Vec::with_capacity(self.runs);
        let mut run_seeds = Vec::with_capacity(self.runs);
        let mut first_dependency_graph = None;

        // Generate GroupId for this eval run (all iterations share the same GroupId)
        // This enables DPO learning to compare multiple executions under the same conditions
        let group_id = GroupId::new();

        for i in 0..self.runs {
            let run_seed = self.seed.wrapping_add(i as u64);
            run_seeds.push(run_seed);

            let (result, dep_graph) = self.run_single(i, run_seed, group_id)?;
            eval_runs.push(result);

            // Keep the first run's DependencyGraph for action_order caching
            if i == 0 {
                first_dependency_graph = dep_graph;
            }
        }

        let aggregated = Aggregator::aggregate(&eval_runs);

        // Offline learning is now executed via LifecycleHook (on_terminate)
        // The TrainTrigger controls when learning runs (default: after all runs complete)
        //
        // TODO: action_order saving should be moved to LearningLifecycleHook's learn_callback
        // For now, save action_order separately if DependencyGraph is available
        if let (Some(ref store), Some(ref graph)) = (&self.learning_store, &first_dependency_graph)
        {
            if graph.has_action_order() {
                let scenario_key = self.scenario.meta.id.learning_key();

                // Load existing offline model and update action_order
                if let Ok(mut model) = store.load_offline_model(&scenario_key) {
                    use swarm_engine_core::learn::LearnedActionOrder;

                    let action_order = LearnedActionOrder::new(
                        graph.discover_order().to_vec(),
                        graph.not_discover_order().to_vec(),
                        graph.available_actions(),
                    );
                    model.action_order = Some(action_order.clone());

                    if let Err(e) = store.save_offline_model(&scenario_key, &model) {
                        tracing::warn!("Failed to save action_order: {}", e);
                    } else {
                        println!(
                            "Action order saved: discover={:?}, not_discover={:?}",
                            action_order.discover, action_order.not_discover
                        );
                    }
                }
            }
        }

        Ok(EvalReport {
            config_summary: ConfigSummary {
                scenario_name: self.scenario.meta.name.clone(),
                scenario_id: self.scenario.meta.id.to_string(),
                worker_count: self.scenario.agents.workers.iter().map(|w| w.count).sum(),
                max_ticks: self.scenario.app_config.max_ticks,
                run_count: self.runs,
            },
            seed_info: SeedInfo {
                base_seed: self.seed,
                run_seeds,
            },
            runs: eval_runs,
            aggregated,
            assertion_results: vec![],
        })
    }

    /// Run a single evaluation iteration
    ///
    /// # Arguments
    /// * `index` - The iteration index (0-based)
    /// * `seed` - The random seed for this iteration
    /// * `group_id` - The group ID shared by all iterations in this eval run.
    ///    Used for DPO learning to compare multiple executions.
    fn run_single(
        &self,
        index: usize,
        seed: u64,
        group_id: GroupId,
    ) -> Result<(
        EvalRun,
        Option<swarm_engine_core::exploration::DependencyGraph>,
    )> {
        let workers = self.build_workers();

        // Build ManagementStrategy from scenario config
        let management_strategy = self.build_management_strategy();

        let config = SwarmConfig {
            tick_duration: Duration::from_millis(self.scenario.app_config.tick_duration_ms),
            max_ticks: self.scenario.app_config.max_ticks,
            management_strategy,
        };

        // Start building Orchestrator
        let mut builder = OrchestratorBuilder::new().config(config);

        // Add workers
        for worker in workers {
            builder = builder.add_worker_boxed(worker);
        }

        // Add Managers: from factory if provided, otherwise from scenario templates
        if let Some(factory) = &self.manager_factory {
            let manager = factory();
            builder = builder.manager(DynManagerWrapper(manager));
        } else {
            // Build managers from scenario templates
            let managers = self.build_managers();
            for manager in managers {
                builder = builder.manager(manager);
            }
        }

        // Add BatchInvoker if factory provided
        if let Some(factory) = &self.batch_invoker_factory {
            let invoker = factory();
            builder = builder.batch_invoker(DynBatchInvokerWrapper(invoker));
        }

        // Build Extensions from scenario (includes LlmConfig, ActionsConfig, EvalSeed)
        let extensions = self.build_extensions_from_scenario(seed);
        builder = builder.extensions(extensions);

        // Enable ExplorationSpace if requested (from scenario or explicit setting)
        if self.enable_exploration || self.scenario.app_config.enable_exploration {
            builder = builder.with_exploration();
        }

        // Apply offline model for optimized parameters (if no explicit OperatorProvider factory)
        if self.operator_provider_factory.is_none() {
            if let Some(ref model) = self.offline_model {
                builder = builder.with_offline_model(model.clone());
                println!(
                    "Offline model applied: ucb1_c={:.3}, maturity={}, strategy={}",
                    model.parameters.ucb1_c,
                    model.strategy_config.maturity_threshold,
                    model.strategy_config.initial_strategy
                );
            }
        }

        // Apply learned action order if available (independent of OperatorProvider)
        // This allows skipping LLM for DependencyGraph generation
        if let Some(ref model) = self.offline_model {
            if let Some(ref action_order) = model.action_order {
                let provider = LearnedDependencyProvider::new(action_order.clone());
                builder = builder.dependency_provider(provider);
                println!(
                    "Learned action order applied: discover={:?}, not_discover={:?}",
                    action_order.discover, action_order.not_discover
                );
            }
        }

        // Set OperatorProvider if factory provided (overrides offline model)
        if let Some(factory) = &self.operator_provider_factory {
            let provider = factory();
            builder = builder.operator_provider(DynOperatorProviderWrapper(provider));
        }

        // Set ActionEventPublisher if action_events_path is configured
        let jsonl_handle = if let Some(base_path) = &self.action_events_path {
            // Create path with run index suffix
            let path = if self.runs > 1 {
                let stem = base_path.file_stem().unwrap_or_default().to_string_lossy();
                let ext = base_path.extension().unwrap_or_default().to_string_lossy();
                let parent = base_path.parent().unwrap_or(std::path::Path::new("."));
                parent.join(format!("{}_run{}.{}", stem, index, ext))
            } else {
                base_path.clone()
            };

            let (publisher, _rx) = ActionEventPublisher::new(1024);
            let jsonl_rx = publisher.subscribe();
            let writer = JsonlWriter::new(jsonl_rx, path);

            // Spawn JsonlWriter in background
            let handle = self.runtime.spawn(async move {
                if let Err(e) = writer.run().await {
                    eprintln!("JsonlWriter error: {}", e);
                }
            });

            builder = builder.action_collector(publisher);
            Some(handle)
        } else {
            None
        };

        // Set LifecycleHook for automatic learning trigger
        if let Some(ref store) = self.learning_store {
            let scenario_key = self.scenario.meta.id.learning_key();
            let mut hook = LearningLifecycleHook::new(store.storage().base_dir())
                .with_scenario(scenario_key)
                .with_shared_eval_count(std::sync::Arc::clone(&self.eval_count));

            // Apply custom trigger if configured, otherwise use CountTrigger(runs)
            // This means learning runs after all iterations complete
            if let Some(ref trigger) = self.train_trigger {
                hook = hook.with_trigger(std::sync::Arc::clone(trigger));
            } else {
                // Default: run learning after all runs in this eval complete
                hook = hook.with_trigger(std::sync::Arc::new(CountTrigger::new(self.runs)));
            }

            builder = builder.lifecycle_hook(Box::new(hook));
        }

        let mut orchestrator = builder.build(self.runtime.clone());

        // Enable partitioning when multiple managers are configured
        let manager_count = self
            .scenario
            .agents
            .managers
            .iter()
            .map(|t| t.count)
            .sum::<usize>();
        if manager_count > 1 {
            orchestrator.enable_partitioning();
        }

        // Determine task: prefer explicit task, fallback to scenario task
        // Set group_id for DPO learning (all iterations in this eval share the same GroupId)
        let task_to_run = self
            .task
            .clone()
            .or_else(|| self.build_task_from_scenario())
            .map(|task| task.with_group_id(group_id));

        // Run with task if available, otherwise run without task
        // run_task() returns Result to enforce DependencyGraph requirement
        let result = if let Some(task) = task_to_run {
            orchestrator.run_task(task)?
        } else {
            orchestrator.run()
        };

        let state = orchestrator.state();
        let timed_out = result.total_ticks >= self.scenario.app_config.max_ticks;
        let environment_done = state.shared.is_environment_done();
        let total_actions = state.shared.stats.total_visits() as u64;
        let successful_actions = state.shared.stats.total_successes() as u64;
        let llm_invocations = state.shared.llm_invocations();
        let llm_invoke_errors = state.shared.llm_errors();

        let metrics = RunMetrics {
            task: crate::metrics::TaskMetrics {
                total_ticks: result.total_ticks,
                total_tasks: 0,
                completed_tasks: 0,
                total_actions,
                successful_actions,
                success_rate: state.shared.stats.success_rate(),
            },
            coordination: crate::metrics::CoordinationMetrics {
                // manager_activations は廃止。LLM 呼び出し回数で近似（1 batch = 1 activation として）
                manager_activations: llm_invocations,
                manager_intervention_rate: if result.total_ticks > 0 {
                    llm_invocations as f64 / result.total_ticks as f64
                } else {
                    0.0
                },
                ..Default::default()
            },
            performance: {
                let llm_error_rate = if llm_invocations > 0 {
                    llm_invoke_errors as f64 / llm_invocations as f64
                } else {
                    0.0
                };
                crate::metrics::PerformanceMetrics {
                    total_duration_ms: result.total_duration.as_millis() as f64,
                    avg_tick_latency_ms: if result.total_ticks > 0 {
                        result.total_duration.as_millis() as f64 / result.total_ticks as f64
                    } else {
                        0.0
                    },
                    raw_throughput_per_sec: if result.total_duration.as_secs_f64() > 0.0 {
                        total_actions as f64 / result.total_duration.as_secs_f64()
                    } else {
                        0.0
                    },
                    effective_throughput_per_sec: if result.total_duration.as_secs_f64() > 0.0 {
                        successful_actions as f64 / result.total_duration.as_secs_f64()
                    } else {
                        0.0
                    },
                    llm_invocations,
                    llm_invoke_errors,
                    llm_error_rate,
                    ..Default::default()
                }
            },
            robustness: Default::default(),
        };

        // Evaluate success/failure conditions
        let (success, termination_reason) = if !result.completed {
            (false, TerminationReason::Stopped)
        } else {
            self.evaluate_conditions(&metrics, environment_done, timed_out)
        };

        // Save statistics to LearningStore if configured
        if let Some(ref store) = self.learning_store {
            // Use learning_key for directory naming (e.g., "troubleshooting")
            // but keep display name in metadata for human readability
            let scenario_key = self.scenario.meta.id.learning_key();
            let metadata = SnapshotMetadata::default()
                .with_scenario(&scenario_key)
                .with_task(&self.scenario.task.goal);

            // Get LearnStats from LearnedProvider if available
            let learn_stats_opt = orchestrator.learned_provider().and_then(|p| p.stats());

            let snapshot = if let Some(learn_stats) = learn_stats_opt {
                LearningSnapshot {
                    version: swarm_engine_core::learn::SNAPSHOT_VERSION,
                    metadata,
                    episode_transitions: learn_stats.episode_transitions.clone(),
                    ngram_stats: learn_stats.ngram_stats.clone(),
                    selection_performance: learn_stats.selection_performance.clone(),
                    contextual_stats: learn_stats
                        .contextual_stats
                        .iter()
                        .map(|(k, v)| {
                            (
                                k.clone(),
                                swarm_engine_core::online_stats::ActionStats {
                                    visits: v.visits,
                                    successes: v.successes,
                                    failures: v.failures,
                                    ..Default::default()
                                },
                            )
                        })
                        .collect(),
                    action_stats: state
                        .shared
                        .stats
                        .all_action_stats()
                        .map(|(k, v)| (k.clone(), v.clone()))
                        .collect(),
                }
            } else {
                // No LearnStats available - create snapshot with only action_stats
                LearningSnapshot {
                    version: swarm_engine_core::learn::SNAPSHOT_VERSION,
                    metadata,
                    episode_transitions: Default::default(),
                    ngram_stats: Default::default(),
                    selection_performance: Default::default(),
                    contextual_stats: Default::default(),
                    action_stats: state
                        .shared
                        .stats
                        .all_action_stats()
                        .map(|(k, v)| (k.clone(), v.clone()))
                        .collect(),
                }
            };

            if let Err(e) = store.save_session(&scenario_key, &snapshot) {
                eprintln!("Warning: Failed to save learning data: {}", e);
            }
        }

        // Extract DependencyGraph and learn_record before dropping orchestrator
        let dependency_graph = orchestrator.dependency_graph().cloned();
        let learn_record = dependency_graph
            .as_ref()
            .and_then(|g| g.learn_record().cloned());

        // Save Episode with DependencyGraphRecord if learn_record is available and learning is enabled
        if let (Some(ref store), Some(record)) = (&self.learning_store, learn_record) {
            use std::fs::OpenOptions;
            use std::io::Write;
            use swarm_engine_core::learn::{Episode, EpisodeContext, Outcome};

            let scenario_key = self.scenario.meta.id.learning_key();
            let ticks = result.total_ticks as u32;
            let max_ticks = self.scenario.app_config.max_ticks as u32;

            // Determine Outcome based on task completion status
            // - Success: Environment reached its goal (normalized score by ticks)
            // - Timeout: Hit max_ticks without completing
            // - Failure: Environment did not complete for other reasons
            let outcome = if environment_done {
                // Score: higher is better, normalized by max_ticks
                // Fast completion (low ticks) = high score
                let score = 1.0 - (ticks as f64 / max_ticks as f64).min(1.0);
                Outcome::success(score.max(0.01)) // Minimum score for successful completion
            } else if timed_out {
                // Timed out without completing - use partial score based on progress
                let partial_score = (ticks as f64 / max_ticks as f64).min(0.99);
                Outcome::timeout(Some(partial_score))
            } else {
                // Failed for other reasons (e.g., all workers idle, error)
                Outcome::failure(format!("Task not completed at tick {}", ticks))
            };
            let is_success = outcome.is_success();

            // Build Episode with DependencyGraphRecord in context
            let context = EpisodeContext::new().with_record(record);
            let episode = Episode::builder()
                .learn_model("dependency_graph")
                .context(context)
                .outcome(outcome)
                .scenario(&scenario_key)
                .build();

            // Save to JSONL file
            let path = store.storage().base_dir().join(format!(
                "scenarios/{}/dep_graph_episodes.jsonl",
                scenario_key
            ));

            if let Some(parent) = path.parent() {
                let _ = std::fs::create_dir_all(parent);
            }

            if let Ok(mut file) = OpenOptions::new().create(true).append(true).open(&path) {
                if let Ok(json) = serde_json::to_string(&episode) {
                    let _ = writeln!(file, "{}", json);
                    tracing::debug!(
                        path = %path.display(),
                        success = is_success,
                        "Saved Episode with DependencyGraphRecord"
                    );
                }
            }
        }

        // Drop orchestrator to close ActionEventPublisher channel
        drop(orchestrator);

        // Wait for JsonlWriter to finish writing
        if let Some(handle) = jsonl_handle {
            // Use block_on to wait synchronously
            let _ = self.runtime.block_on(handle);
        }

        Ok((
            EvalRun::new(index, seed, success, termination_reason, metrics),
            dependency_graph,
        ))
    }

    fn build_workers(&self) -> Vec<Box<dyn WorkerAgent>> {
        let mut workers: Vec<Box<dyn WorkerAgent>> = Vec::new();

        for template in &self.scenario.agents.workers {
            for i in 0..template.count {
                let id = workers.len();
                let name = template.id_pattern.replace("{i}", &i.to_string());

                let worker = GenericWorker::new(id)
                    .with_name(name)
                    .with_require_guidance(true);

                workers.push(Box::new(worker));
            }
        }

        workers
    }

    fn build_managers(&self) -> Vec<DefaultBatchManagerAgent> {
        let mut managers = Vec::new();
        let mut manager_index = 0;

        for template in &self.scenario.agents.managers {
            let ids = template.generate_ids();
            for name in ids {
                let manager = DefaultBatchManagerAgent::new(ManagerId(manager_index))
                    .with_name(name)
                    .with_interval(self.scenario.manager.process_interval_ticks);

                managers.push(manager);
                manager_index += 1;
            }
        }

        // デフォルト: Manager テンプレートがない場合は 1 つ作成
        if managers.is_empty() {
            managers.push(
                DefaultBatchManagerAgent::new(ManagerId(0))
                    .with_name("default_manager")
                    .with_interval(self.scenario.manager.process_interval_ticks),
            );
        }

        managers
    }

    fn build_management_strategy(&self) -> ManagementStrategy {
        match &self.scenario.app_config.management_strategy {
            ManagementStrategyConfig::EveryTick {} => ManagementStrategy::EveryTick,
            ManagementStrategyConfig::IntervalBased { max_interval } => {
                ManagementStrategy::FixedInterval {
                    interval: *max_interval,
                }
            }
            ManagementStrategyConfig::EventDriven { triggers: _ } => {
                // Event-driven maps to completion-based
                ManagementStrategy::CompletionBased { max_wait_ticks: 50 }
            }
            ManagementStrategyConfig::Hybrid {
                max_interval,
                triggers: _,
            } => ManagementStrategy::Hybrid {
                preferred_interval: *max_interval,
                force_after_ticks: max_interval * 2,
            },
            ManagementStrategyConfig::Disabled {} => {
                // Disabled = very large interval (effectively never)
                ManagementStrategy::FixedInterval { interval: u64::MAX }
            }
        }
    }

    /// Build SwarmTask from scenario task config
    ///
    /// Returns None if task goal is empty
    fn build_task_from_scenario(&self) -> Option<SwarmTask> {
        let task_config = &self.scenario.task;

        if task_config.goal.is_empty() {
            return None;
        }

        // Build context JSON object
        let mut context = serde_json::Map::new();

        if let Some(target_path) = &task_config.context.target_path {
            context.insert(
                "target_path".to_string(),
                serde_json::Value::String(target_path.clone()),
            );
        }
        if let Some(working_dir) = &task_config.context.working_dir {
            context.insert(
                "working_dir".to_string(),
                serde_json::Value::String(working_dir.clone()),
            );
        }
        if let Some(max_depth) = task_config.context.max_depth {
            context.insert(
                "max_depth".to_string(),
                serde_json::Value::Number(serde_json::Number::from(max_depth)),
            );
        }

        // Add extra context (convert toml::Value to serde_json::Value)
        for (key, value) in &task_config.context.extra {
            if let Ok(json_value) = serde_json::to_value(value) {
                context.insert(key.clone(), json_value);
            }
        }

        let task =
            SwarmTask::new(&task_config.goal).with_context(serde_json::Value::Object(context));

        Some(task)
    }

    /// Build Extensions with LlmConfig, ActionsConfig, and EvalSeed from scenario
    fn build_extensions_from_scenario(&self, seed: u64) -> Extensions {
        let mut extensions = if let Some(factory) = &self.extensions_factory {
            factory()
        } else {
            Extensions::new()
        };

        // Insert EvalSeed for reproducibility
        extensions.insert(EvalSeed(seed));

        // Insert LlmConfig for BatchInvoker/Manager to use
        extensions.insert(self.scenario.llm.clone());

        // Insert LoRA config if specified (for BatchInvoker to use)
        if let Some(ref lora) = self.scenario.llm.lora {
            extensions.insert(lora.clone());
        }

        // Insert ManagerConfig for Manager to use
        extensions.insert(self.scenario.manager.clone());

        // Insert BatchProcessorConfig for BatchInvoker to use
        extensions.insert(self.scenario.batch_processor.clone());

        // Convert EvalActionsConfig to Core ActionsConfig for Worker/Manager to use
        let core_actions_config = self.scenario.actions.to_core_config();
        extensions.insert(core_actions_config);

        // Create and insert Environment based on env_type
        let env_type = self.scenario.environment.env_type.as_str();
        let env_params = &self.scenario.environment.params;

        let env_box: Option<EnvironmentBox> = match env_type {
            "maze" => {
                let map = env_params.get("map").and_then(|v| v.as_str()).unwrap_or("");
                let worker_count = env_params
                    .get("worker_count")
                    .and_then(|v| v.as_u64())
                    .unwrap_or(1) as usize;
                Some(Box::new(MazeEnvironment::from_str(map, worker_count)))
            }
            "code" => {
                // Currently only "auth" scenario is supported, default to it
                Some(Box::new(CodeEnvironment::auth_scenario()))
            }
            "troubleshooting" => {
                let scenario_name = env_params
                    .get("scenario")
                    .and_then(|v| v.as_str())
                    .unwrap_or("memory_leak");
                let env = match scenario_name {
                    "memory_leak" => TroubleshootingEnvironment::memory_leak_scenario(),
                    "cpu_spike" => TroubleshootingEnvironment::cpu_spike_scenario(),
                    "network_timeout" => TroubleshootingEnvironment::network_timeout_scenario(),
                    "medium" => TroubleshootingEnvironment::complex_scenario(15, 3, 2, seed),
                    "high" => TroubleshootingEnvironment::complex_scenario(30, 8, 3, seed),
                    "extreme" => TroubleshootingEnvironment::complex_scenario(50, 15, 4, seed),
                    "complex" => {
                        let total_services = env_params
                            .get("total_services")
                            .and_then(|v| v.as_u64())
                            .unwrap_or(15) as usize;
                        let noise_services = env_params
                            .get("noise_services")
                            .and_then(|v| v.as_u64())
                            .unwrap_or(3) as usize;
                        let cascade_depth = env_params
                            .get("cascade_depth")
                            .and_then(|v| v.as_u64())
                            .unwrap_or(2) as usize;
                        TroubleshootingEnvironment::complex_scenario(
                            total_services,
                            noise_services,
                            cascade_depth,
                            seed,
                        )
                    }
                    _ => TroubleshootingEnvironment::memory_leak_scenario(),
                };
                Some(Box::new(env))
            }
            "search" => {
                let scenario_name = env_params
                    .get("scenario")
                    .and_then(|v| v.as_str())
                    .unwrap_or("basic");
                let env = match scenario_name {
                    "basic" => SearchEnvironment::basic_scenario(),
                    "medium" => SearchEnvironment::medium_scenario(),
                    "large" => SearchEnvironment::large_scenario(),
                    "custom" => {
                        let file_count = env_params
                            .get("file_count")
                            .and_then(|v| v.as_u64())
                            .unwrap_or(5) as usize;
                        let target_index = env_params
                            .get("target_index")
                            .and_then(|v| v.as_u64())
                            .unwrap_or(2) as usize;
                        SearchEnvironment::custom_scenario(file_count, target_index, seed)
                    }
                    _ => SearchEnvironment::basic_scenario(),
                };
                Some(Box::new(env))
            }
            "internal_diagnosis" => {
                let scenario_name = env_params
                    .get("scenario")
                    .and_then(|v| v.as_str())
                    .unwrap_or("routing");
                let env = match scenario_name {
                    "routing" => InternalDiagnosisEnvironment::routing_error_scenario(),
                    "failover" => InternalDiagnosisEnvironment::failover_error_scenario(),
                    "worker_pool" => InternalDiagnosisEnvironment::worker_pool_scenario(),
                    "strategy" => InternalDiagnosisEnvironment::strategy_mismatch_scenario(),
                    "exploration" => InternalDiagnosisEnvironment::exploration_depth_scenario(),
                    "complex" => InternalDiagnosisEnvironment::complex_scenario(seed),
                    _ => InternalDiagnosisEnvironment::routing_error_scenario(),
                };
                Some(Box::new(env))
            }
            "deep_search" => {
                let scenario_name = env_params
                    .get("scenario")
                    .and_then(|v| v.as_str())
                    .unwrap_or("tech_question");
                let env = match scenario_name {
                    "tech_question" | _ => DeepSearchEnvironment::tech_question_scenario(),
                };
                Some(Box::new(env))
            }
            // Real filesystem environment (Bash, Read, Write, Grep, Glob)
            "default" | "realworld" => {
                use swarm_engine_core::environment::DefaultEnvironment;
                let working_dir = env_params
                    .get("working_dir")
                    .and_then(|v| v.as_str())
                    .map(std::path::PathBuf::from);
                let env = if let Some(dir) = working_dir {
                    DefaultEnvironment::with_working_dir(dir)
                } else {
                    DefaultEnvironment::new()
                };
                Some(Box::new(env))
            }
            _ => None, // Unknown env_type - no Environment inserted
        };

        if let Some(env) = env_box {
            extensions.insert(env);
        }

        // Insert DependencyGraph if specified (explicit or from scenario)
        let graph = self.dependency_graph.clone().or_else(|| {
            self.scenario.dependency_graph.as_ref().and_then(|cfg| {
                let action_names = self.scenario.actions.action_names();
                cfg.to_core_graph(&action_names)
            })
        });
        if let Some(g) = graph {
            extensions.insert(g);
        }

        // Insert prior learning snapshot if available
        // OrchestratorBuilder.build() will call stats.load_prior()
        if let Some(ref prior) = self.prior_snapshot {
            extensions.insert(prior.clone());
        }

        extensions
    }

    /// Evaluate scenario conditions to determine success/failure
    ///
    /// Returns (success, termination_reason)
    fn evaluate_conditions(
        &self,
        metrics: &RunMetrics,
        environment_done: bool,
        timed_out: bool,
    ) -> (bool, TerminationReason) {
        let conditions = &self.scenario.conditions;

        // 1. Check failure conditions first (any match = fail)
        for condition in &conditions.failure {
            if let Some(actual) =
                self.get_metric_value(&condition.metric, metrics, environment_done)
            {
                if condition.evaluate(&actual) {
                    return (false, TerminationReason::Failure);
                }
            }
        }

        // 2. Handle timeout
        if timed_out {
            return match conditions.on_timeout {
                TimeoutBehavior::Fail => (false, TerminationReason::Timeout),
                TimeoutBehavior::PartialSuccess => {
                    // Check if success conditions are met
                    let success = self.check_success_conditions(metrics, environment_done);
                    (success, TerminationReason::Timeout)
                }
                TimeoutBehavior::MilestoneScore => {
                    // TODO: Implement milestone scoring
                    (false, TerminationReason::Timeout)
                }
            };
        }

        // 3. Check success conditions (all must match)
        let success = self.check_success_conditions(metrics, environment_done);
        if success {
            (true, TerminationReason::Success)
        } else {
            // Not yet successful, but no failure conditions met either
            // This shouldn't happen if called after completion
            (false, TerminationReason::Stopped)
        }
    }

    /// Check if all success conditions are met
    fn check_success_conditions(&self, metrics: &RunMetrics, environment_done: bool) -> bool {
        let conditions = &self.scenario.conditions;

        // If no success conditions defined, consider successful
        if conditions.success.is_empty() {
            return true;
        }

        // All conditions must pass
        conditions.success.iter().all(|condition| {
            self.get_metric_value(&condition.metric, metrics, environment_done)
                .map(|actual| condition.evaluate(&actual))
                .unwrap_or(false)
        })
    }

    /// Get metric value by path (e.g., "environment.done", "task.success_rate")
    fn get_metric_value(
        &self,
        path: &str,
        metrics: &RunMetrics,
        environment_done: bool,
    ) -> Option<ConditionValue> {
        match path {
            // Environment metrics
            "environment.done" => Some(ConditionValue::Bool(environment_done)),

            // Task metrics
            "task.total_ticks" | "total_ticks" => {
                Some(ConditionValue::Integer(metrics.task.total_ticks as i64))
            }
            "task.success_rate" | "success_rate" => {
                Some(ConditionValue::Float(metrics.task.success_rate))
            }
            "task.total_actions" | "total_actions" => {
                Some(ConditionValue::Integer(metrics.task.total_actions as i64))
            }
            "task.successful_actions" | "successful_actions" => Some(ConditionValue::Integer(
                metrics.task.successful_actions as i64,
            )),

            // Performance metrics
            "performance.llm_error_rate" | "llm_error_rate" => {
                Some(ConditionValue::Float(metrics.performance.llm_error_rate))
            }
            "performance.llm_invocations" | "llm_invocations" => Some(ConditionValue::Integer(
                metrics.performance.llm_invocations as i64,
            )),

            // Coordination metrics
            "coordination.manager_activations" | "manager_activations" => Some(
                ConditionValue::Integer(metrics.coordination.manager_activations as i64),
            ),

            // Error metrics (from failed_actions)
            "errors.count" => {
                let failed = metrics
                    .task
                    .total_actions
                    .saturating_sub(metrics.task.successful_actions);
                Some(ConditionValue::Integer(failed as i64))
            }

            // Unknown metric
            _ => None,
        }
    }
}

// Wrapper to convert Box<dyn ManagerAgent> to impl ManagerAgent
struct DynManagerWrapper(Box<dyn ManagerAgent>);

impl ManagerAgent for DynManagerWrapper {
    fn prepare(
        &self,
        context: &swarm_engine_core::agent::TaskContext,
    ) -> swarm_engine_core::agent::BatchDecisionRequest {
        self.0.prepare(context)
    }

    fn finalize(
        &self,
        context: &swarm_engine_core::agent::TaskContext,
        responses: Vec<(
            swarm_engine_core::types::WorkerId,
            swarm_engine_core::agent::DecisionResponse,
        )>,
    ) -> swarm_engine_core::agent::ManagementDecision {
        self.0.finalize(context, responses)
    }

    fn id(&self) -> swarm_engine_core::agent::ManagerId {
        self.0.id()
    }

    fn name(&self) -> &str {
        self.0.name()
    }
}

// Wrapper to convert Box<dyn BatchInvoker> to impl BatchInvoker
struct DynBatchInvokerWrapper(Box<dyn BatchInvoker>);

impl BatchInvoker for DynBatchInvokerWrapper {
    fn invoke(
        &self,
        request: swarm_engine_core::agent::BatchDecisionRequest,
        extensions: &swarm_engine_core::extensions::Extensions,
    ) -> swarm_engine_core::agent::BatchInvokeResult {
        self.0.invoke(request, extensions)
    }

    fn plan_dependencies(
        &self,
        task: &str,
        actions: &[ActionDef],
    ) -> Option<swarm_engine_core::exploration::DependencyGraph> {
        self.0.plan_dependencies(task, actions)
    }

    fn name(&self) -> &str {
        self.0.name()
    }
}

// Wrapper to convert Box<dyn OperatorProvider<NodeRules>> to impl OperatorProvider<NodeRules>
struct DynOperatorProviderWrapper(Box<dyn OperatorProvider<NodeRules>>);

impl OperatorProvider<NodeRules> for DynOperatorProviderWrapper {
    fn provide(
        &self,
        rules: NodeRules,
        context: Option<
            &swarm_engine_core::exploration::ProviderContext<
                '_,
                swarm_engine_core::exploration::ActionNodeData,
                String,
                swarm_engine_core::exploration::MapNodeState,
            >,
        >,
    ) -> swarm_engine_core::exploration::ConfigurableOperator<NodeRules> {
        self.0.provide(rules, context)
    }

    fn reevaluate(
        &self,
        operator: &mut swarm_engine_core::exploration::ConfigurableOperator<NodeRules>,
        ctx: &swarm_engine_core::exploration::ProviderContext<
            '_,
            swarm_engine_core::exploration::ActionNodeData,
            String,
            swarm_engine_core::exploration::MapNodeState,
        >,
    ) {
        self.0.reevaluate(operator, ctx)
    }

    fn name(&self) -> &str {
        self.0.name()
    }
}