oris_evolution/
pipeline.rs

1//! Evolution Pipeline - Complete detect/select/mutate runtime pipeline.
2//!
3//! This module implements the full evolution loop as separate runtime stages:
4//! - Detect: Extract signals from task context
5//! - Select: Choose gene candidates based on signals
6//! - Mutate: Prepare mutation proposals
7//! - Execute: Run the mutation in sandbox
8//! - Validate: Verify mutation correctness
9//! - Evaluate: Assess mutation quality
10//! - Solidify: Create gene/capsule events
11//! - Reuse: Mark capsule as reusable
12
13use serde::{Deserialize, Serialize};
14use std::collections::HashMap;
15use std::sync::Arc;
16use std::time::{Duration, Instant};
17use thiserror::Error;
18
19use crate::core::{GeneCandidate, PreparedMutation, Selector, SelectorInput};
20use crate::evolver::{EvolutionSignal, MutationProposal, MutationRiskLevel, ValidationResult};
21use crate::port::{
22    EvaluateInput, EvaluatePort, GeneStorePersistPort, SandboxPort, SignalExtractorInput,
23    SignalExtractorPort, ValidateInput, ValidatePort,
24};
25use crate::task_class::{load_task_classes, TaskClassInferencer};
26
27/// Pipeline configuration
28#[derive(Clone, Debug, Serialize, Deserialize)]
29pub struct EvolutionPipelineConfig {
30    /// Enable/disable specific stages
31    pub enable_detect: bool,
32    pub enable_select: bool,
33    pub enable_mutate: bool,
34    pub enable_execute: bool,
35    pub enable_validate: bool,
36    pub enable_evaluate: bool,
37    pub enable_solidify: bool,
38    pub enable_reuse: bool,
39
40    /// Stage timeout in seconds
41    pub detect_timeout_secs: u64,
42    pub select_timeout_secs: u64,
43    pub mutate_timeout_secs: u64,
44    pub execute_timeout_secs: u64,
45    pub validate_timeout_secs: u64,
46    pub evaluate_timeout_secs: u64,
47    pub solidify_timeout_secs: u64,
48    pub reuse_timeout_secs: u64,
49
50    /// Max candidates to select
51    pub max_candidates: usize,
52
53    /// Min signal confidence threshold
54    pub min_signal_confidence: f32,
55}
56
57impl Default for EvolutionPipelineConfig {
58    fn default() -> Self {
59        Self {
60            enable_detect: true,
61            enable_select: true,
62            enable_mutate: true,
63            enable_execute: true,
64            enable_validate: true,
65            enable_evaluate: true,
66            enable_solidify: true,
67            enable_reuse: true,
68            detect_timeout_secs: 30,
69            select_timeout_secs: 30,
70            mutate_timeout_secs: 60,
71            execute_timeout_secs: 300,
72            validate_timeout_secs: 60,
73            evaluate_timeout_secs: 30,
74            solidify_timeout_secs: 30,
75            reuse_timeout_secs: 30,
76            max_candidates: 10,
77            min_signal_confidence: 0.5,
78        }
79    }
80}
81
82/// Stage state
83#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
84pub enum PipelineStageState {
85    /// Stage not started
86    Pending,
87    /// Stage currently running
88    Running,
89    /// Stage completed successfully
90    Completed,
91    /// Stage failed with error
92    Failed(String),
93    /// Stage was skipped
94    Skipped(String),
95}
96
97/// Pipeline execution context (internal use, not serialized)
98pub struct PipelineContext {
99    /// Input task context
100    pub task_input: serde_json::Value,
101    /// Optional extractor input for the Detect stage.
102    /// When set and a `SignalExtractorPort` is injected into the pipeline,
103    /// the Detect stage will call the extractor to populate `signals`.
104    pub extractor_input: Option<SignalExtractorInput>,
105    /// Signals extracted in Detect phase
106    pub signals: Vec<EvolutionSignal>,
107    /// Gene candidates selected in Select phase
108    pub candidates: Vec<GeneCandidate>,
109    /// Mutation proposals prepared in Mutate phase
110    pub proposals: Vec<MutationProposal>,
111    /// Execution result
112    pub execution_result: Option<serde_json::Value>,
113    /// Validation result
114    pub validation_result: Option<ValidationResult>,
115    /// Evaluation result
116    pub evaluation_result: Option<EvaluationResult>,
117    /// Solidified genes
118    pub solidified_genes: Vec<String>,
119    /// Reused capsules
120    pub reused_capsules: Vec<String>,
121    /// Wall-clock duration recorded for each stage that ran.
122    pub stage_timings: HashMap<String, Duration>,
123    /// Task class ID inferred by the Detect stage, if a `TaskClassInferencer` is attached.
124    pub inferred_task_class_id: Option<String>,
125}
126
127impl Default for PipelineContext {
128    fn default() -> Self {
129        Self {
130            task_input: serde_json::json!({}),
131            extractor_input: None,
132            signals: Vec::new(),
133            candidates: Vec::new(),
134            proposals: Vec::new(),
135            execution_result: None,
136            validation_result: None,
137            evaluation_result: None,
138            solidified_genes: Vec::new(),
139            reused_capsules: Vec::new(),
140            stage_timings: HashMap::new(),
141            inferred_task_class_id: None,
142        }
143    }
144}
145
146/// Evaluation result
147#[derive(Clone, Debug, Serialize, Deserialize)]
148pub struct EvaluationResult {
149    pub score: f32,
150    pub improvements: Vec<String>,
151    pub regressions: Vec<String>,
152    pub recommendation: EvaluationRecommendation,
153}
154
155#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
156pub enum EvaluationRecommendation {
157    Accept,
158    Reject,
159    NeedsRevision,
160    RequiresHumanReview,
161}
162
163/// Pipeline execution result
164#[derive(Clone, Debug, Serialize, Deserialize)]
165pub struct PipelineResult {
166    /// Whether the pipeline completed successfully
167    pub success: bool,
168    /// Stage states
169    pub stage_states: Vec<StageState>,
170    /// Error message if failed
171    pub error: Option<String>,
172    /// Task class inferred during the Detect stage (present when a
173    /// `TaskClassInferencer` was attached to the pipeline).
174    pub inferred_task_class_id: Option<String>,
175}
176
177/// Individual stage state
178#[derive(Clone, Debug, Serialize, Deserialize)]
179pub struct StageState {
180    pub stage_name: String,
181    pub state: PipelineStageState,
182    pub duration_ms: Option<u64>,
183}
184
185impl StageState {
186    pub fn new(name: &str) -> Self {
187        Self {
188            stage_name: name.to_string(),
189            state: PipelineStageState::Pending,
190            duration_ms: None,
191        }
192    }
193}
194
195/// Pipeline errors
196#[derive(Error, Debug)]
197pub enum PipelineError {
198    #[error("Detect stage error: {0}")]
199    DetectError(String),
200
201    #[error("Select stage error: {0}")]
202    SelectError(String),
203
204    #[error("Mutate stage error: {0}")]
205    MutateError(String),
206
207    #[error("Execute stage error: {0}")]
208    ExecuteError(String),
209
210    #[error("Validate stage error: {0}")]
211    ValidateError(String),
212
213    #[error("Evaluate stage error: {0}")]
214    EvaluateError(String),
215
216    #[error("Solidify stage error: {0}")]
217    SolidifyError(String),
218
219    #[error("Reuse stage error: {0}")]
220    ReuseError(String),
221
222    #[error("Pipeline timeout: {0}")]
223    Timeout(String),
224}
225
226/// Evolution Pipeline trait
227pub trait EvolutionPipeline: Send + Sync {
228    /// Get pipeline name
229    fn name(&self) -> &str;
230
231    /// Get pipeline configuration
232    fn config(&self) -> &EvolutionPipelineConfig;
233
234    /// Execute the full pipeline
235    fn execute(&self, context: PipelineContext) -> Result<PipelineResult, PipelineError>;
236
237    /// Execute a specific stage
238    #[deprecated(note = "Use execute() instead")]
239    fn execute_stage(
240        &self,
241        stage: PipelineStage,
242        context: &mut PipelineContext,
243    ) -> Result<PipelineStageState, PipelineError>;
244}
245
246/// Pipeline stages
247#[derive(Clone, Debug, Copy, PartialEq, Eq, Serialize, Deserialize)]
248pub enum PipelineStage {
249    Detect,
250    Select,
251    Mutate,
252    Execute,
253    Validate,
254    Evaluate,
255    Solidify,
256    Reuse,
257}
258
259impl PipelineStage {
260    pub fn as_str(&self) -> &'static str {
261        match self {
262            PipelineStage::Detect => "detect",
263            PipelineStage::Select => "select",
264            PipelineStage::Mutate => "mutate",
265            PipelineStage::Execute => "execute",
266            PipelineStage::Validate => "validate",
267            PipelineStage::Evaluate => "evaluate",
268            PipelineStage::Solidify => "solidify",
269            PipelineStage::Reuse => "reuse",
270        }
271    }
272
273    pub fn all() -> Vec<PipelineStage> {
274        vec![
275            PipelineStage::Detect,
276            PipelineStage::Select,
277            PipelineStage::Mutate,
278            PipelineStage::Execute,
279            PipelineStage::Validate,
280            PipelineStage::Evaluate,
281            PipelineStage::Solidify,
282            PipelineStage::Reuse,
283        ]
284    }
285}
286
287/// Standard evolution pipeline implementation
288pub struct StandardEvolutionPipeline {
289    name: String,
290    config: EvolutionPipelineConfig,
291    selector: Arc<dyn Selector>,
292    /// Optional signal extractor for the Detect stage.
293    signal_extractor: Option<Arc<dyn SignalExtractorPort>>,
294    /// Optional task-class inferencer for the Detect stage.
295    task_class_inferencer: Option<TaskClassInferencer>,
296    /// Optional sandbox for the Execute stage.
297    sandbox: Option<Arc<dyn SandboxPort>>,
298    /// Optional gene store for the Solidify/Reuse stages.
299    gene_store: Option<Arc<dyn GeneStorePersistPort>>,
300    /// Optional validator for the Validate stage.
301    validate_port: Option<Arc<dyn ValidatePort>>,
302    /// Optional evaluator for the Evaluate stage.
303    evaluate_port: Option<Arc<dyn EvaluatePort>>,
304}
305
306impl StandardEvolutionPipeline {
307    /// Create a pipeline with a mandatory selector and no injected ports.
308    /// Detect will pass through pre-populated signals; Execute will use a
309    /// no-op stub that records a synthetic success result.
310    pub fn new(config: EvolutionPipelineConfig, selector: Arc<dyn Selector>) -> Self {
311        Self {
312            name: "standard".to_string(),
313            config,
314            selector,
315            signal_extractor: None,
316            task_class_inferencer: None,
317            sandbox: None,
318            gene_store: None,
319            validate_port: None,
320            evaluate_port: None,
321        }
322    }
323
324    /// Attach a `SignalExtractorPort` for the Detect stage.
325    pub fn with_signal_extractor(mut self, extractor: Arc<dyn SignalExtractorPort>) -> Self {
326        self.signal_extractor = Some(extractor);
327        self
328    }
329
330    /// Attach a `TaskClassInferencer` for the Detect stage.
331    ///
332    /// When attached the Detect stage automatically infers the task class from
333    /// collected signals and stores the result in
334    /// `PipelineResult::inferred_task_class_id`.  If not attached, inference is
335    /// skipped and the field is `None`.
336    pub fn with_task_class_inferencer(mut self, inferencer: TaskClassInferencer) -> Self {
337        self.task_class_inferencer = Some(inferencer);
338        self
339    }
340
341    /// Attach a `TaskClassInferencer` pre-loaded with the current
342    /// `load_task_classes()` registry (builtin + optional TOML override).
343    pub fn with_default_task_class_inferencer(self) -> Self {
344        let inferencer = TaskClassInferencer::new(load_task_classes());
345        self.with_task_class_inferencer(inferencer)
346    }
347
348    /// Attach a `SandboxPort` for the Execute stage.
349    pub fn with_sandbox(mut self, sandbox: Arc<dyn SandboxPort>) -> Self {
350        self.sandbox = Some(sandbox);
351        self
352    }
353
354    /// Attach a `GeneStorePersistPort` for the Solidify/Reuse stages.
355    pub fn with_gene_store(mut self, gene_store: Arc<dyn GeneStorePersistPort>) -> Self {
356        self.gene_store = Some(gene_store);
357        self
358    }
359
360    /// Attach a `ValidatePort` for the Validate stage.
361    pub fn with_validate_port(mut self, validator: Arc<dyn ValidatePort>) -> Self {
362        self.validate_port = Some(validator);
363        self
364    }
365
366    /// Attach an `EvaluatePort` for the Evaluate stage.
367    pub fn with_evaluate_port(mut self, evaluator: Arc<dyn EvaluatePort>) -> Self {
368        self.evaluate_port = Some(evaluator);
369        self
370    }
371}
372
373impl EvolutionPipeline for StandardEvolutionPipeline {
374    fn name(&self) -> &str {
375        &self.name
376    }
377
378    fn config(&self) -> &EvolutionPipelineConfig {
379        &self.config
380    }
381
382    fn execute(&self, mut context: PipelineContext) -> Result<PipelineResult, PipelineError> {
383        let mut stage_states = Vec::new();
384
385        // Detect phase
386        if self.config.enable_detect {
387            let mut stage = StageState::new(PipelineStage::Detect.as_str());
388            stage.state = PipelineStageState::Running;
389            stage_states.push(stage);
390
391            let t0 = Instant::now();
392            if let Some(ref extractor) = self.signal_extractor {
393                // Use the injected extractor to populate signals from raw input.
394                let input = context.extractor_input.clone().unwrap_or_default();
395                let extracted = extractor.extract(&input);
396                // Merge: keep any pre-populated signals and append new ones.
397                context.signals.extend(extracted);
398            }
399            // When no extractor is injected, signals already in context are
400            // used as-is (pass-through, backward-compatible behaviour).
401
402            // Infer task class from collected signals when an inferencer is set.
403            if let Some(ref inferencer) = self.task_class_inferencer {
404                if !context.signals.is_empty() {
405                    let combined: String = context
406                        .signals
407                        .iter()
408                        .map(|s| s.description.as_str())
409                        .collect::<Vec<_>>()
410                        .join(" ");
411                    context.inferred_task_class_id = Some(inferencer.infer(&combined));
412                }
413            }
414
415            let elapsed = t0.elapsed();
416            context
417                .stage_timings
418                .insert(PipelineStage::Detect.as_str().to_string(), elapsed);
419            let d_ms = elapsed.as_millis() as u64;
420            let last = stage_states.last_mut().unwrap();
421            last.state = PipelineStageState::Completed;
422            last.duration_ms = Some(d_ms);
423        } else {
424            stage_states.push(StageState {
425                stage_name: PipelineStage::Detect.as_str().to_string(),
426                state: PipelineStageState::Skipped("disabled".to_string()),
427                duration_ms: None,
428            });
429        }
430
431        // Select phase
432        if self.config.enable_select {
433            let mut stage = StageState::new(PipelineStage::Select.as_str());
434            stage.state = PipelineStageState::Running;
435            stage_states.push(stage);
436
437            let t0 = Instant::now();
438            let input = SelectorInput {
439                signals: context
440                    .signals
441                    .iter()
442                    .map(|s| s.description.clone())
443                    .collect(),
444                env: crate::core::EnvFingerprint {
445                    rustc_version: String::new(),
446                    cargo_lock_hash: String::new(),
447                    target_triple: String::new(),
448                    os: String::new(),
449                },
450                spec_id: None,
451                limit: self.config.max_candidates,
452            };
453
454            context.candidates = self.selector.select(&input);
455            let elapsed = t0.elapsed();
456            context
457                .stage_timings
458                .insert(PipelineStage::Select.as_str().to_string(), elapsed);
459            let last = stage_states.last_mut().unwrap();
460            last.state = PipelineStageState::Completed;
461            last.duration_ms = Some(elapsed.as_millis() as u64);
462        } else {
463            stage_states.push(StageState {
464                stage_name: PipelineStage::Select.as_str().to_string(),
465                state: PipelineStageState::Skipped("disabled".to_string()),
466                duration_ms: None,
467            });
468        }
469
470        // Mutate phase - prepare proposals from candidates
471        if self.config.enable_mutate {
472            let mut stage = StageState::new(PipelineStage::Mutate.as_str());
473            stage.state = PipelineStageState::Running;
474            stage_states.push(stage);
475
476            let t0 = Instant::now();
477            context.proposals = context
478                .candidates
479                .iter()
480                .enumerate()
481                .map(|(i, candidate)| MutationProposal {
482                    proposal_id: format!("proposal_{}", i),
483                    signal_ids: vec![],
484                    gene_id: candidate.gene.id.clone(),
485                    description: format!("Mutation for gene {}", candidate.gene.id),
486                    estimated_impact: candidate.score,
487                    risk_level: MutationRiskLevel::Medium,
488                    proposed_changes: serde_json::json!({}),
489                })
490                .collect();
491            let elapsed = t0.elapsed();
492            context
493                .stage_timings
494                .insert(PipelineStage::Mutate.as_str().to_string(), elapsed);
495            let last = stage_states.last_mut().unwrap();
496            last.state = PipelineStageState::Completed;
497            last.duration_ms = Some(elapsed.as_millis() as u64);
498        } else {
499            stage_states.push(StageState {
500                stage_name: PipelineStage::Mutate.as_str().to_string(),
501                state: PipelineStageState::Skipped("disabled".to_string()),
502                duration_ms: None,
503            });
504        }
505
506        // Execute phase
507        if self.config.enable_execute {
508            let mut stage = StageState::new(PipelineStage::Execute.as_str());
509            stage.state = PipelineStageState::Running;
510            stage_states.push(stage);
511
512            let t0 = Instant::now();
513            if let (Some(ref sb), Some(proposal)) = (&self.sandbox, context.proposals.first()) {
514                // Build a minimal PreparedMutation from the first proposal so
515                // the sandbox can apply the change in an isolated workspace.
516                let mutation = build_prepared_mutation(proposal);
517                let result = sb.execute(&mutation);
518                context.execution_result = Some(result.to_json());
519                let last = stage_states.last_mut().unwrap();
520                last.state = if result.success {
521                    PipelineStageState::Completed
522                } else {
523                    PipelineStageState::Failed(result.message.clone())
524                };
525            } else {
526                // No sandbox injected — fall back to stub (backward-compatible).
527                context.execution_result = Some(serde_json::json!({
528                    "success": true,
529                    "stdout": "",
530                    "stderr": "",
531                    "duration_ms": 0,
532                    "message": "Mutation executed successfully (stub)"
533                }));
534                stage_states.last_mut().unwrap().state = PipelineStageState::Completed;
535            }
536            let elapsed = t0.elapsed();
537            context
538                .stage_timings
539                .insert(PipelineStage::Execute.as_str().to_string(), elapsed);
540            stage_states.last_mut().unwrap().duration_ms = Some(elapsed.as_millis() as u64);
541        } else {
542            stage_states.push(StageState {
543                stage_name: PipelineStage::Execute.as_str().to_string(),
544                state: PipelineStageState::Skipped("disabled".to_string()),
545                duration_ms: None,
546            });
547        }
548
549        // Validate phase
550        if self.config.enable_validate {
551            let mut stage = StageState::new(PipelineStage::Validate.as_str());
552            stage.state = PipelineStageState::Running;
553            stage_states.push(stage);
554
555            let t0 = Instant::now();
556            if let Some(proposal) = context.proposals.first() {
557                let vresult = if let Some(ref vp) = self.validate_port {
558                    // Build input from execution result stored in context.
559                    let exec = context.execution_result.as_ref();
560                    let exec_success = exec
561                        .and_then(|v| v.get("success"))
562                        .and_then(|v| v.as_bool())
563                        .unwrap_or(false);
564                    let stdout = exec
565                        .and_then(|v| v.get("stdout"))
566                        .and_then(|v| v.as_str())
567                        .unwrap_or("")
568                        .to_string();
569                    let stderr = exec
570                        .and_then(|v| v.get("stderr"))
571                        .and_then(|v| v.as_str())
572                        .unwrap_or("")
573                        .to_string();
574                    let input = ValidateInput {
575                        proposal_id: proposal.proposal_id.clone(),
576                        execution_success: exec_success,
577                        stdout,
578                        stderr,
579                    };
580                    vp.validate(&input)
581                } else {
582                    // Backward-compatible stub when no validator is injected.
583                    ValidationResult {
584                        proposal_id: proposal.proposal_id.clone(),
585                        passed: true,
586                        score: 0.8,
587                        issues: vec![],
588                        simulation_results: None,
589                    }
590                };
591                context.validation_result = Some(vresult);
592            }
593            let elapsed = t0.elapsed();
594            context
595                .stage_timings
596                .insert(PipelineStage::Validate.as_str().to_string(), elapsed);
597            let last = stage_states.last_mut().unwrap();
598            // Mark stage Failed when validation did not pass so callers can detect it.
599            last.state = match &context.validation_result {
600                Some(r) if !r.passed => PipelineStageState::Failed("validation failed".to_string()),
601                _ => PipelineStageState::Completed,
602            };
603            last.duration_ms = Some(elapsed.as_millis() as u64);
604        } else {
605            stage_states.push(StageState {
606                stage_name: PipelineStage::Validate.as_str().to_string(),
607                state: PipelineStageState::Skipped("disabled".to_string()),
608                duration_ms: None,
609            });
610        }
611
612        // Evaluate phase
613        if self.config.enable_evaluate {
614            let mut stage = StageState::new(PipelineStage::Evaluate.as_str());
615            stage.state = PipelineStageState::Running;
616            stage_states.push(stage);
617
618            let t0 = Instant::now();
619            context.evaluation_result = if let Some(ref ep) = self.evaluate_port {
620                if let Some(proposal) = context.proposals.first() {
621                    let input = EvaluateInput {
622                        proposal_id: proposal.proposal_id.clone(),
623                        intent: proposal.description.clone(),
624                        original: String::new(),
625                        proposed: String::new(),
626                        signals: context
627                            .signals
628                            .iter()
629                            .map(|s| s.description.clone())
630                            .collect(),
631                    };
632                    Some(ep.evaluate(&input))
633                } else {
634                    None
635                }
636            } else {
637                // Backward-compatible stub when no evaluator is injected.
638                Some(EvaluationResult {
639                    score: 0.8,
640                    improvements: vec!["Mutation applied successfully".to_string()],
641                    regressions: vec![],
642                    recommendation: EvaluationRecommendation::Accept,
643                })
644            };
645            let elapsed = t0.elapsed();
646            context
647                .stage_timings
648                .insert(PipelineStage::Evaluate.as_str().to_string(), elapsed);
649            let last = stage_states.last_mut().unwrap();
650            last.state = PipelineStageState::Completed;
651            last.duration_ms = Some(elapsed.as_millis() as u64);
652        } else {
653            stage_states.push(StageState {
654                stage_name: PipelineStage::Evaluate.as_str().to_string(),
655                state: PipelineStageState::Skipped("disabled".to_string()),
656                duration_ms: None,
657            });
658        }
659
660        // Solidify phase - persist promoted genes via the GeneStorePersistPort
661        if self.config.enable_solidify {
662            let mut stage = StageState::new(PipelineStage::Solidify.as_str());
663            stage.state = PipelineStageState::Running;
664            stage_states.push(stage);
665
666            let t0 = Instant::now();
667            let mut solidified: Vec<String> = Vec::new();
668            for candidate in &context.candidates {
669                let gene = &candidate.gene;
670                // Persist via injected port when available.
671                if let Some(ref store) = self.gene_store {
672                    store.persist_gene(&gene.id, &gene.signals, &gene.strategy, &gene.validation);
673                }
674                solidified.push(gene.id.clone());
675            }
676            context.solidified_genes = solidified;
677            let elapsed = t0.elapsed();
678            context
679                .stage_timings
680                .insert(PipelineStage::Solidify.as_str().to_string(), elapsed);
681            let last = stage_states.last_mut().unwrap();
682            last.state = PipelineStageState::Completed;
683            last.duration_ms = Some(elapsed.as_millis() as u64);
684        } else {
685            stage_states.push(StageState {
686                stage_name: PipelineStage::Solidify.as_str().to_string(),
687                state: PipelineStageState::Skipped("disabled".to_string()),
688                duration_ms: None,
689            });
690        }
691
692        // Reuse phase - record capsule reuse via the GeneStorePersistPort
693        if self.config.enable_reuse {
694            let mut stage = StageState::new(PipelineStage::Reuse.as_str());
695            stage.state = PipelineStageState::Running;
696            stage_states.push(stage);
697
698            let t0 = Instant::now();
699            let mut reused: Vec<String> = Vec::new();
700            for candidate in &context.candidates {
701                let cap_ids: Vec<String> =
702                    candidate.capsules.iter().map(|c| c.id.clone()).collect();
703                if let Some(ref store) = self.gene_store {
704                    store.mark_reused(&candidate.gene.id, &cap_ids);
705                }
706                reused.extend(cap_ids);
707            }
708            context.reused_capsules = reused;
709            let elapsed = t0.elapsed();
710            context
711                .stage_timings
712                .insert(PipelineStage::Reuse.as_str().to_string(), elapsed);
713            let last = stage_states.last_mut().unwrap();
714            last.state = PipelineStageState::Completed;
715            last.duration_ms = Some(elapsed.as_millis() as u64);
716        } else {
717            stage_states.push(StageState {
718                stage_name: PipelineStage::Reuse.as_str().to_string(),
719                state: PipelineStageState::Skipped("disabled".to_string()),
720                duration_ms: None,
721            });
722        }
723
724        // Propagate validation failure to the overall pipeline result.
725        let validation_passed = context
726            .validation_result
727            .as_ref()
728            .map(|r| r.passed)
729            .unwrap_or(true);
730
731        Ok(PipelineResult {
732            success: validation_passed,
733            stage_states,
734            error: if validation_passed {
735                None
736            } else {
737                Some("Validation stage did not pass".to_string())
738            },
739            inferred_task_class_id: context.inferred_task_class_id.clone(),
740        })
741    }
742
743    fn execute_stage(
744        &self,
745        stage: PipelineStage,
746        context: &mut PipelineContext,
747    ) -> Result<PipelineStageState, PipelineError> {
748        match stage {
749            PipelineStage::Detect => {
750                // Signals already in context
751                Ok(PipelineStageState::Completed)
752            }
753            PipelineStage::Select => {
754                let input = SelectorInput {
755                    signals: context
756                        .signals
757                        .iter()
758                        .map(|s| s.description.clone())
759                        .collect(),
760                    env: crate::core::EnvFingerprint {
761                        rustc_version: String::new(),
762                        cargo_lock_hash: String::new(),
763                        target_triple: String::new(),
764                        os: String::new(),
765                    },
766                    spec_id: None,
767                    limit: self.config.max_candidates,
768                };
769                context.candidates = self.selector.select(&input);
770                Ok(PipelineStageState::Completed)
771            }
772            PipelineStage::Mutate => {
773                context.proposals = context
774                    .candidates
775                    .iter()
776                    .enumerate()
777                    .map(|(i, candidate)| MutationProposal {
778                        proposal_id: format!("proposal_{}", i),
779                        signal_ids: vec![],
780                        gene_id: candidate.gene.id.clone(),
781                        description: format!("Mutation for gene {}", candidate.gene.id),
782                        estimated_impact: candidate.score,
783                        risk_level: MutationRiskLevel::Medium,
784                        proposed_changes: serde_json::json!({}),
785                    })
786                    .collect();
787                Ok(PipelineStageState::Completed)
788            }
789            PipelineStage::Execute => {
790                context.execution_result = Some(serde_json::json!({
791                    "status": "success"
792                }));
793                Ok(PipelineStageState::Completed)
794            }
795            PipelineStage::Validate => {
796                if let Some(proposal) = context.proposals.first() {
797                    let validation_result = if let Some(ref validator) = self.validate_port {
798                        let exec = context.execution_result.as_ref();
799                        let input = ValidateInput {
800                            proposal_id: proposal.proposal_id.clone(),
801                            execution_success: exec
802                                .and_then(|value| value.get("success"))
803                                .and_then(|value| value.as_bool())
804                                .unwrap_or(false),
805                            stdout: exec
806                                .and_then(|value| value.get("stdout"))
807                                .and_then(|value| value.as_str())
808                                .unwrap_or("")
809                                .to_string(),
810                            stderr: exec
811                                .and_then(|value| value.get("stderr"))
812                                .and_then(|value| value.as_str())
813                                .unwrap_or("")
814                                .to_string(),
815                        };
816                        validator.validate(&input)
817                    } else {
818                        ValidationResult {
819                            proposal_id: proposal.proposal_id.clone(),
820                            passed: true,
821                            score: 0.8,
822                            issues: vec![],
823                            simulation_results: None,
824                        }
825                    };
826                    context.validation_result = Some(validation_result);
827                }
828                Ok(match context.validation_result.as_ref() {
829                    Some(result) if !result.passed => {
830                        PipelineStageState::Failed("validation failed".to_string())
831                    }
832                    _ => PipelineStageState::Completed,
833                })
834            }
835            PipelineStage::Evaluate => {
836                context.evaluation_result = if let Some(ref evaluator) = self.evaluate_port {
837                    context.proposals.first().map(|proposal| {
838                        evaluator.evaluate(&EvaluateInput {
839                            proposal_id: proposal.proposal_id.clone(),
840                            intent: proposal.description.clone(),
841                            original: String::new(),
842                            proposed: String::new(),
843                            signals: context
844                                .signals
845                                .iter()
846                                .map(|signal| signal.description.clone())
847                                .collect(),
848                        })
849                    })
850                } else {
851                    Some(EvaluationResult {
852                        score: 0.8,
853                        improvements: vec!["Mutation applied successfully".to_string()],
854                        regressions: vec![],
855                        recommendation: EvaluationRecommendation::Accept,
856                    })
857                };
858                Ok(PipelineStageState::Completed)
859            }
860            PipelineStage::Solidify => {
861                for candidate in &context.candidates {
862                    let gene = &candidate.gene;
863                    if let Some(ref store) = self.gene_store {
864                        store.persist_gene(
865                            &gene.id,
866                            &gene.signals,
867                            &gene.strategy,
868                            &gene.validation,
869                        );
870                    }
871                }
872                context.solidified_genes = context
873                    .candidates
874                    .iter()
875                    .map(|c| c.gene.id.clone())
876                    .collect();
877                Ok(PipelineStageState::Completed)
878            }
879            PipelineStage::Reuse => {
880                for candidate in &context.candidates {
881                    let cap_ids: Vec<String> =
882                        candidate.capsules.iter().map(|c| c.id.clone()).collect();
883                    if let Some(ref store) = self.gene_store {
884                        store.mark_reused(&candidate.gene.id, &cap_ids);
885                    }
886                    context.reused_capsules.extend(cap_ids);
887                }
888                Ok(PipelineStageState::Completed)
889            }
890        }
891    }
892}
893
894/// Build a minimal `PreparedMutation` from a `MutationProposal`.
895///
896/// Used by the Execute stage when a `SandboxPort` is injected. The resulting
897/// mutation carries the proposal identifier and an empty unified-diff payload;
898/// real mutation generation (LLM or rule-based) will replace this in a later
899/// phase of the evolution pipeline.
900fn build_prepared_mutation(proposal: &MutationProposal) -> PreparedMutation {
901    use crate::core::{
902        ArtifactEncoding, MutationArtifact, MutationIntent, MutationTarget, RiskLevel,
903    };
904
905    PreparedMutation {
906        intent: MutationIntent {
907            id: proposal.proposal_id.clone(),
908            intent: proposal.description.clone(),
909            target: MutationTarget::WorkspaceRoot,
910            expected_effect: format!("Apply mutation for gene {}", proposal.gene_id),
911            risk: match proposal.risk_level {
912                MutationRiskLevel::Low => RiskLevel::Low,
913                MutationRiskLevel::Medium => RiskLevel::Medium,
914                MutationRiskLevel::High | MutationRiskLevel::Critical => RiskLevel::High,
915            },
916            signals: proposal.signal_ids.clone(),
917            spec_id: None,
918        },
919        artifact: MutationArtifact {
920            encoding: ArtifactEncoding::UnifiedDiff,
921            payload: String::new(),
922            base_revision: None,
923            content_hash: String::new(),
924        },
925    }
926}
927
928#[cfg(test)]
929mod tests {
930    use super::*;
931    use std::sync::atomic::{AtomicUsize, Ordering};
932
933    #[test]
934    fn test_pipeline_config_default() {
935        let config = EvolutionPipelineConfig::default();
936        assert!(config.enable_detect);
937        assert!(config.enable_select);
938        assert!(config.enable_mutate);
939    }
940
941    #[test]
942    fn test_pipeline_stage_states() {
943        let config = EvolutionPipelineConfig {
944            enable_detect: false,
945            enable_select: true,
946            enable_mutate: false,
947            ..Default::default()
948        };
949
950        // Just test that config works
951        assert!(!config.enable_detect);
952        assert!(config.enable_select);
953    }
954
955    // ─── GeneStorePersistPort integration test ─────────────────────────────
956
957    /// A minimal in-memory mock that records which genes/capsules were persisted.
958    struct MockGeneStore {
959        genes: std::sync::Mutex<Vec<String>>,
960        reused: std::sync::Mutex<Vec<String>>,
961    }
962
963    impl MockGeneStore {
964        fn new() -> Self {
965            Self {
966                genes: std::sync::Mutex::new(Vec::new()),
967                reused: std::sync::Mutex::new(Vec::new()),
968            }
969        }
970    }
971
972    impl GeneStorePersistPort for MockGeneStore {
973        fn persist_gene(
974            &self,
975            gene_id: &str,
976            _signals: &[String],
977            _strategy: &[String],
978            _validation: &[String],
979        ) -> bool {
980            self.genes.lock().unwrap().push(gene_id.to_string());
981            true
982        }
983
984        fn mark_reused(&self, gene_id: &str, _capsule_ids: &[String]) -> bool {
985            self.reused.lock().unwrap().push(gene_id.to_string());
986            true
987        }
988    }
989
990    /// A minimal `Selector` that always returns one hard-coded candidate.
991    struct SingleCandidateSelector;
992
993    impl Selector for SingleCandidateSelector {
994        fn select(&self, _input: &SelectorInput) -> Vec<GeneCandidate> {
995            use crate::core;
996            vec![GeneCandidate {
997                gene: core::Gene {
998                    id: "gene-abc-001".to_string(),
999                    signals: vec!["test-signal".to_string()],
1000                    strategy: vec!["apply-fix".to_string()],
1001                    validation: vec!["cargo test".to_string()],
1002                    state: core::AssetState::default(),
1003                    task_class_id: None,
1004                },
1005                capsules: vec![],
1006                score: 0.9,
1007            }]
1008        }
1009    }
1010
1011    struct RecordingValidatePort {
1012        passed: bool,
1013        calls: std::sync::Mutex<Vec<ValidateInput>>,
1014    }
1015
1016    impl RecordingValidatePort {
1017        fn new(passed: bool) -> Self {
1018            Self {
1019                passed,
1020                calls: std::sync::Mutex::new(Vec::new()),
1021            }
1022        }
1023    }
1024
1025    impl ValidatePort for RecordingValidatePort {
1026        fn validate(&self, input: &ValidateInput) -> ValidationResult {
1027            self.calls.lock().unwrap().push(input.clone());
1028            ValidationResult {
1029                proposal_id: input.proposal_id.clone(),
1030                passed: self.passed,
1031                score: if self.passed { 1.0 } else { 0.2 },
1032                issues: vec![],
1033                simulation_results: None,
1034            }
1035        }
1036    }
1037
1038    struct RecordingEvaluatePort {
1039        calls: std::sync::Mutex<Vec<EvaluateInput>>,
1040    }
1041
1042    impl RecordingEvaluatePort {
1043        fn new() -> Self {
1044            Self {
1045                calls: std::sync::Mutex::new(Vec::new()),
1046            }
1047        }
1048    }
1049
1050    struct CountingValidatePort {
1051        call_count: Arc<AtomicUsize>,
1052        passed: bool,
1053    }
1054
1055    impl CountingValidatePort {
1056        fn new(call_count: Arc<AtomicUsize>, passed: bool) -> Self {
1057            Self { call_count, passed }
1058        }
1059    }
1060
1061    impl ValidatePort for CountingValidatePort {
1062        fn validate(&self, input: &ValidateInput) -> ValidationResult {
1063            self.call_count.fetch_add(1, Ordering::SeqCst);
1064            ValidationResult {
1065                proposal_id: input.proposal_id.clone(),
1066                passed: self.passed,
1067                score: if self.passed { 0.95 } else { 0.15 },
1068                issues: vec![],
1069                simulation_results: None,
1070            }
1071        }
1072    }
1073
1074    struct CountingEvaluatePort {
1075        call_count: Arc<AtomicUsize>,
1076    }
1077
1078    impl CountingEvaluatePort {
1079        fn new(call_count: Arc<AtomicUsize>) -> Self {
1080            Self { call_count }
1081        }
1082    }
1083
1084    impl EvaluatePort for CountingEvaluatePort {
1085        fn evaluate(&self, _input: &EvaluateInput) -> EvaluationResult {
1086            self.call_count.fetch_add(1, Ordering::SeqCst);
1087            EvaluationResult {
1088                score: 0.91,
1089                improvements: vec!["evaluate port called".to_string()],
1090                regressions: vec![],
1091                recommendation: EvaluationRecommendation::Accept,
1092            }
1093        }
1094    }
1095
1096    impl EvaluatePort for RecordingEvaluatePort {
1097        fn evaluate(&self, input: &EvaluateInput) -> EvaluationResult {
1098            self.calls.lock().unwrap().push(input.clone());
1099            EvaluationResult {
1100                score: 0.33,
1101                improvements: vec!["used injected evaluator".to_string()],
1102                regressions: vec!["none".to_string()],
1103                recommendation: EvaluationRecommendation::NeedsRevision,
1104            }
1105        }
1106    }
1107
1108    #[test]
1109    fn test_solidify_reuse_calls_gene_store() {
1110        let store = Arc::new(MockGeneStore::new());
1111        let config = EvolutionPipelineConfig {
1112            enable_detect: false,
1113            enable_select: true,
1114            enable_mutate: false,
1115            enable_execute: false,
1116            enable_validate: false,
1117            enable_evaluate: false,
1118            enable_solidify: true,
1119            enable_reuse: true,
1120            ..Default::default()
1121        };
1122
1123        let pipeline = StandardEvolutionPipeline::new(config, Arc::new(SingleCandidateSelector))
1124            .with_gene_store(store.clone());
1125
1126        let ctx = PipelineContext::default();
1127        let result = pipeline.execute(ctx).expect("pipeline should succeed");
1128        assert!(result.success);
1129
1130        // Verify Solidify stage persisted the gene
1131        let persisted_genes = store.genes.lock().unwrap();
1132        assert!(
1133            persisted_genes.contains(&"gene-abc-001".to_string()),
1134            "Solidify stage should have persisted gene-abc-001, got: {:?}",
1135            persisted_genes
1136        );
1137    }
1138
1139    #[test]
1140    fn test_execute_stage_validate_uses_injected_port_and_fallback() {
1141        let validator = Arc::new(RecordingValidatePort::new(false));
1142        let pipeline = StandardEvolutionPipeline::new(
1143            EvolutionPipelineConfig::default(),
1144            Arc::new(SingleCandidateSelector),
1145        )
1146        .with_validate_port(validator.clone());
1147
1148        let mut context = PipelineContext::default();
1149        context.proposals.push(MutationProposal {
1150            proposal_id: "proposal-1".to_string(),
1151            signal_ids: vec![],
1152            gene_id: "gene-1".to_string(),
1153            description: "validate proposal".to_string(),
1154            estimated_impact: 0.5,
1155            risk_level: MutationRiskLevel::Medium,
1156            proposed_changes: serde_json::json!({}),
1157        });
1158        context.execution_result = Some(serde_json::json!({
1159            "success": true,
1160            "stdout": "validator stdout",
1161            "stderr": ""
1162        }));
1163
1164        let state = pipeline
1165            .execute_stage(PipelineStage::Validate, &mut context)
1166            .expect("validate stage should succeed");
1167
1168        assert_eq!(
1169            state,
1170            PipelineStageState::Failed("validation failed".to_string())
1171        );
1172        let calls = validator.calls.lock().unwrap();
1173        assert_eq!(calls.len(), 1);
1174        assert_eq!(calls[0].proposal_id, "proposal-1");
1175        assert!(calls[0].execution_success);
1176        assert_eq!(calls[0].stdout, "validator stdout");
1177        drop(calls);
1178        assert!(!context.validation_result.as_ref().unwrap().passed);
1179
1180        let fallback_pipeline = StandardEvolutionPipeline::new(
1181            EvolutionPipelineConfig::default(),
1182            Arc::new(SingleCandidateSelector),
1183        );
1184        let mut fallback_context = PipelineContext::default();
1185        fallback_context.proposals.push(MutationProposal {
1186            proposal_id: "proposal-2".to_string(),
1187            signal_ids: vec![],
1188            gene_id: "gene-2".to_string(),
1189            description: "fallback validate".to_string(),
1190            estimated_impact: 0.4,
1191            risk_level: MutationRiskLevel::Low,
1192            proposed_changes: serde_json::json!({}),
1193        });
1194
1195        let fallback_state = fallback_pipeline
1196            .execute_stage(PipelineStage::Validate, &mut fallback_context)
1197            .expect("fallback validate stage should succeed");
1198
1199        assert_eq!(fallback_state, PipelineStageState::Completed);
1200        assert!(fallback_context.validation_result.as_ref().unwrap().passed);
1201        assert_eq!(
1202            fallback_context.validation_result.as_ref().unwrap().score,
1203            0.8
1204        );
1205    }
1206
1207    #[test]
1208    fn test_execute_stage_evaluate_uses_injected_port_and_fallback() {
1209        let evaluator = Arc::new(RecordingEvaluatePort::new());
1210        let pipeline = StandardEvolutionPipeline::new(
1211            EvolutionPipelineConfig::default(),
1212            Arc::new(SingleCandidateSelector),
1213        )
1214        .with_evaluate_port(evaluator.clone());
1215
1216        let mut context = PipelineContext::default();
1217        context.signals.push(EvolutionSignal {
1218            signal_id: "signal-1".to_string(),
1219            signal_type: crate::evolver::SignalType::ErrorPattern {
1220                error_type: "panic".to_string(),
1221                frequency: 1,
1222            },
1223            source_task_id: "task-1".to_string(),
1224            confidence: 0.9,
1225            description: "improve evaluator path".to_string(),
1226            metadata: serde_json::json!({}),
1227        });
1228        context.proposals.push(MutationProposal {
1229            proposal_id: "proposal-3".to_string(),
1230            signal_ids: vec!["signal-1".to_string()],
1231            gene_id: "gene-3".to_string(),
1232            description: "evaluate proposal".to_string(),
1233            estimated_impact: 0.7,
1234            risk_level: MutationRiskLevel::Medium,
1235            proposed_changes: serde_json::json!({}),
1236        });
1237
1238        let state = pipeline
1239            .execute_stage(PipelineStage::Evaluate, &mut context)
1240            .expect("evaluate stage should succeed");
1241
1242        assert_eq!(state, PipelineStageState::Completed);
1243        let calls = evaluator.calls.lock().unwrap();
1244        assert_eq!(calls.len(), 1);
1245        assert_eq!(calls[0].proposal_id, "proposal-3");
1246        assert_eq!(calls[0].intent, "evaluate proposal");
1247        assert_eq!(calls[0].signals, vec!["improve evaluator path".to_string()]);
1248        drop(calls);
1249        assert_eq!(
1250            context.evaluation_result.as_ref().unwrap().recommendation,
1251            EvaluationRecommendation::NeedsRevision
1252        );
1253        assert_eq!(context.evaluation_result.as_ref().unwrap().score, 0.33);
1254
1255        let fallback_pipeline = StandardEvolutionPipeline::new(
1256            EvolutionPipelineConfig::default(),
1257            Arc::new(SingleCandidateSelector),
1258        );
1259        let mut fallback_context = PipelineContext::default();
1260
1261        let fallback_state = fallback_pipeline
1262            .execute_stage(PipelineStage::Evaluate, &mut fallback_context)
1263            .expect("fallback evaluate stage should succeed");
1264
1265        assert_eq!(fallback_state, PipelineStageState::Completed);
1266        assert_eq!(
1267            fallback_context
1268                .evaluation_result
1269                .as_ref()
1270                .unwrap()
1271                .recommendation,
1272            EvaluationRecommendation::Accept
1273        );
1274        assert_eq!(
1275            fallback_context
1276                .evaluation_result
1277                .as_ref()
1278                .unwrap()
1279                .improvements,
1280            vec!["Mutation applied successfully".to_string()]
1281        );
1282    }
1283
1284    #[test]
1285    fn test_execute_invokes_injected_validate_and_evaluate_ports() {
1286        let validate_calls = Arc::new(AtomicUsize::new(0));
1287        let evaluate_calls = Arc::new(AtomicUsize::new(0));
1288        let config = EvolutionPipelineConfig {
1289            enable_detect: false,
1290            enable_select: true,
1291            enable_mutate: true,
1292            enable_execute: true,
1293            enable_validate: true,
1294            enable_evaluate: true,
1295            enable_solidify: false,
1296            enable_reuse: false,
1297            ..Default::default()
1298        };
1299
1300        let pipeline = StandardEvolutionPipeline::new(config, Arc::new(SingleCandidateSelector))
1301            .with_validate_port(Arc::new(CountingValidatePort::new(
1302                validate_calls.clone(),
1303                true,
1304            )))
1305            .with_evaluate_port(Arc::new(CountingEvaluatePort::new(evaluate_calls.clone())));
1306
1307        let result = pipeline
1308            .execute(PipelineContext::default())
1309            .expect("pipeline should execute");
1310
1311        assert!(result.success);
1312        assert_eq!(validate_calls.load(Ordering::SeqCst), 1);
1313        assert_eq!(evaluate_calls.load(Ordering::SeqCst), 1);
1314        assert!(result.error.is_none());
1315        assert!(result.stage_states.iter().any(|stage| {
1316            stage.stage_name == PipelineStage::Validate.as_str()
1317                && stage.state == PipelineStageState::Completed
1318        }));
1319        assert!(result.stage_states.iter().any(|stage| {
1320            stage.stage_name == PipelineStage::Evaluate.as_str()
1321                && stage.state == PipelineStageState::Completed
1322        }));
1323    }
1324
1325    #[test]
1326    fn test_execute_propagates_validate_port_failure_to_pipeline_result() {
1327        let validate_calls = Arc::new(AtomicUsize::new(0));
1328        let evaluate_calls = Arc::new(AtomicUsize::new(0));
1329        let config = EvolutionPipelineConfig {
1330            enable_detect: false,
1331            enable_select: true,
1332            enable_mutate: true,
1333            enable_execute: true,
1334            enable_validate: true,
1335            enable_evaluate: true,
1336            enable_solidify: false,
1337            enable_reuse: false,
1338            ..Default::default()
1339        };
1340
1341        let pipeline = StandardEvolutionPipeline::new(config, Arc::new(SingleCandidateSelector))
1342            .with_validate_port(Arc::new(CountingValidatePort::new(
1343                validate_calls.clone(),
1344                false,
1345            )))
1346            .with_evaluate_port(Arc::new(CountingEvaluatePort::new(evaluate_calls.clone())));
1347
1348        let result = pipeline
1349            .execute(PipelineContext::default())
1350            .expect("pipeline should execute");
1351
1352        assert!(!result.success);
1353        assert_eq!(validate_calls.load(Ordering::SeqCst), 1);
1354        assert_eq!(evaluate_calls.load(Ordering::SeqCst), 1);
1355        assert_eq!(
1356            result.error.as_deref(),
1357            Some("Validation stage did not pass")
1358        );
1359        assert!(result.stage_states.iter().any(|stage| {
1360            stage.stage_name == PipelineStage::Validate.as_str()
1361                && stage.state == PipelineStageState::Failed("validation failed".to_string())
1362        }));
1363    }
1364}
oris_evolution/pipeline.rs

oris_evolution/
pipeline.rs
