feat: pipeline stage progress, duplicate sweep fix, strategy evaluation

- Add stage_label to TaskInfo/TaskSnapshot for pipeline stage tracking
- SSE progress events include stage_label (Sweep, Walk-Forward, etc.)
- Thread progress/cancel callbacks through pipeline for sweep bar progress
- Reset progress to indeterminate when moving past sweep stage
- Fix duplicate sweep rows in /runs when multiple runs share max sharpe
- submit_pipeline runs StrategyEvaluation (baseline + robustness + verdict)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: michaelchu

src/application/pipeline.rs

@@ -5,6 +5,7 @@ use serde_json::Value;
 use std::collections::HashMap;
 
 use crate::application::sweeps;
+use crate::scripting::engine::{CancelCallback, ProgressCallback};
 use crate::server::OptopsyServer;
 use crate::tools::pipeline::StageCallback;
 use crate::tools::response_types::pipeline::PipelineResponse;
@@ -41,15 +42,18 @@ pub async fn execute(
     request: &PipelineRequest,
     source: &str,
 ) -> Result<PipelineResponse> {
-    execute_with_stage(server, request, source, &None).await
+    execute_with_stage(server, request, source, &None, None, None).await
 }
 
 /// Execute the full pipeline with an optional stage progress callback.
+#[allow(clippy::too_many_arguments)]
 pub async fn execute_with_stage(
     server: &OptopsyServer,
     request: &PipelineRequest,
     source: &str,
     on_stage: &StageCallback,
+    progress: Option<ProgressCallback>,
+    is_cancelled: Option<&CancelCallback>,
 ) -> Result<PipelineResponse> {
     let run_store = server.require_run_store()?;
 
@@ -73,8 +77,8 @@ pub async fn execute_with_stage(
         &sweep_req,
         source,
         request.thread_id.as_deref(),
-        None,
-        None,
+        progress,
+        is_cancelled,
     )
     .await?;
 

src/application/workflows.rs

@@ -34,21 +34,39 @@ pub async fn execute(
     request: &WorkflowRequest,
     source: &str,
 ) -> Result<WorkflowResponse> {
-    execute_with_stage(server, request, source, &None).await
+    execute_with_stage(server, request, source, &None, None, None).await
 }
 
 pub async fn execute_with_stage(
     server: &OptopsyServer,
     request: &WorkflowRequest,
     source: &str,
     on_stage: &crate::tools::pipeline::StageCallback,
+    progress: Option<crate::scripting::engine::ProgressCallback>,
+    is_cancelled: Option<&crate::scripting::engine::CancelCallback>,
 ) -> Result<WorkflowResponse> {
     match request.kind {
         WorkflowKind::BaselineValidation => Ok(WorkflowResponse::BaselineValidation(
-            pipeline::execute_with_stage(server, &request.pipeline, source, on_stage).await?,
+            pipeline::execute_with_stage(
+                server,
+                &request.pipeline,
+                source,
+                on_stage,
+                progress,
+                is_cancelled,
+            )
+            .await?,
         )),
         WorkflowKind::StrategyEvaluation => Ok(WorkflowResponse::StrategyEvaluation(
-            execute_strategy_evaluation(server, &request.pipeline, source, on_stage).await?,
+            execute_strategy_evaluation(
+                server,
+                &request.pipeline,
+                source,
+                on_stage,
+                progress,
+                is_cancelled,
+            )
+            .await?,
         )),
     }
 }
@@ -58,14 +76,24 @@ async fn execute_strategy_evaluation(
     request: &PipelineRequest,
     source: &str,
     on_stage: &crate::tools::pipeline::StageCallback,
+    progress: Option<crate::scripting::engine::ProgressCallback>,
+    is_cancelled: Option<&crate::scripting::engine::CancelCallback>,
 ) -> Result<StrategyEvaluationResponse> {
     let started_at = Instant::now();
     let mut eval_request = request.clone();
     if eval_request.num_permutations == 0 {
         eval_request.num_permutations = DEFAULT_STRATEGY_EVAL_PERMUTATIONS;
     }
 
-    let pipeline = pipeline::execute_with_stage(server, &eval_request, source, on_stage).await?;
+    let pipeline = pipeline::execute_with_stage(
+        server,
+        &eval_request,
+        source,
+        on_stage,
+        progress,
+        is_cancelled,
+    )
+    .await?;
     let robustness_checks = if pipeline.walk_forward.is_some() {
         if let Some(cb) = on_stage {
             cb("Robustness Checks");

src/data/run_store.rs

@@ -347,14 +347,18 @@ impl RunStore for SqliteRunStore {
                      FROM sweeps sw
                      LEFT JOIN strategies s ON s.id = sw.strategy_id
                      LEFT JOIN (
-                         SELECT r1.*
-                         FROM runs r1
-                         INNER JOIN (
-                             SELECT sweep_id, MAX(sharpe) as max_sharpe
-                             FROM runs
-                             WHERE sweep_id IS NOT NULL
-                             GROUP BY sweep_id
-                         ) r2 ON r1.sweep_id = r2.sweep_id AND r1.sharpe = r2.max_sharpe
+                         SELECT sweep_id,
+                                MAX(sharpe) as sharpe,
+                                MAX(sortino) as sortino,
+                                MAX(cagr) as cagr,
+                                MAX(profit_factor) as profit_factor,
+                                MAX(total_return) as total_return,
+                                MAX(win_rate) as win_rate,
+                                MIN(max_drawdown) as max_drawdown,
+                                MAX(trade_count) as trade_count
+                         FROM runs
+                         WHERE sweep_id IS NOT NULL
+                         GROUP BY sweep_id
                      ) br ON br.sweep_id = sw.id
                      LEFT JOIN (
                          SELECT sweep_id,

src/server/handlers/tasks.rs

@@ -355,16 +355,31 @@ pub async fn submit_pipeline(
                 };
 
                 // Stage progress callback — updates the task's stage_label
+                // and resets bar-level progress for non-sweep stages
                 let task_ref = Arc::clone(&task);
                 let on_stage: crate::tools::pipeline::StageCallback =
                     Some(Box::new(move |label: &str| {
                         *task_ref.stage_label.lock().unwrap() = label.to_string();
+                        if label != "Sweep" {
+                            task_ref.progress_current.store(0, Ordering::Relaxed);
+                            task_ref.progress_total.store(0, Ordering::Relaxed);
+                        }
                     }));
 
-                let result =
-                    workflows::execute_with_stage(&server, &wf_request, "manual", &on_stage)
-                        .await
-                        .map_err(|e| e.to_string())?;
+                // Progress + cancel callbacks for sweep bar-level progress
+                let progress = app_tasks::progress_callback(&task);
+                let is_cancelled = app_tasks::cancel_callback(&task);
+
+                let result = workflows::execute_with_stage(
+                    &server,
+                    &wf_request,
+                    "manual",
+                    &on_stage,
+                    Some(progress),
+                    Some(&is_cancelled),
+                )
+                .await
+                .map_err(|e| e.to_string())?;
 
                 // Extract sweep_id and persist analysis
                 let (sweep_id, result_json) = match &result {

src/tools/pipeline.rs

@@ -29,7 +29,6 @@ const TOP_COMBOS_FOR_WF: usize = 3;
     clippy::too_many_lines,
     clippy::implicit_hasher
 )]
-#[allow(clippy::too_many_arguments)]
 pub async fn run_pipeline(
     server: &OptopsyServer,
     strategy: &str,

tests/pipeline.rs

@@ -111,6 +111,7 @@ async fn significance_gate_fails_skips_downstream() {
         vec!["run-1".to_string(), "run-2".to_string()],
         sweep,
         base_params,
+        &None,
     )
     .await;
 
@@ -165,6 +166,7 @@ async fn no_permutation_passes_significance_gate() {
         vec!["run-1".to_string()],
         sweep,
         base_params,
+        &None,
     )
     .await;
 
@@ -210,6 +212,7 @@ async fn full_pipeline_with_nvda_fixture() {
         vec!["run-1".to_string()],
         sweep,
         base_params,
+        &None,
     )
     .await;
 
@@ -295,6 +298,7 @@ async fn pipeline_preserves_sweep_metadata() {
         run_ids.clone(),
         sweep,
         base_params,
+        &None,
     )
     .await
     .expect("Pipeline should not error");