Pipeline Flow

#[derive(Debug, Clone)]
pub struct PipelineConfig {
    pub repo_dir: PathBuf,
    pub base_branch: String,
    pub test_command: String,
    pub lint_command: String,
    pub max_ci_rounds: u32,
    pub plane: Option<PlaneConfig>,
    /// When true, agent steps are replaced with shell echo stubs (for testing).
    pub dry_run: bool,
    /// GitHub org to restrict repo access to. When set, the pipeline resolves
    /// the target repo from the task message and clones it into a temp workspace.
    pub github_org: Option<String>,
    /// Directory for JSONL trace files. When set, all agent calls are traced.
    pub trace_dir: Option<PathBuf>,
    /// When set, pipeline runs execute inside a Daytona sandbox.
    pub daytona: Option<DaytonaConfig>,
    /// Warm sandbox pool. When set, the pipeline tries to acquire a pre-built
    /// sandbox before falling back to cold creation.
    #[cfg(feature = "daytona")]
    pub pool: Option<Arc<crate::sandbox::pool::WarmPool>>,
}

impl Default for PipelineConfig {
    fn default() -> Self {
        Self {
            repo_dir: PathBuf::from("."),
            base_branch: "main".to_string(),
            test_command: "cargo test".to_string(),
            lint_command: "cargo clippy".to_string(),
            max_ci_rounds: 2,
            plane: None,
            dry_run: false,
            github_org: None,
            trace_dir: None,
            daytona: None,
            #[cfg(feature = "daytona")]
            pool: None,
        }
    }
}

pub async fn run_pipeline(
    platform: &dyn ChatPlatform,
    channel_id: &str,
    user: &str,
    task: &str,
    config: &PipelineConfig,
) -> Result<PipelineResult> {
    // 1. Fetch conversation history
    let history = platform.fetch_history(channel_id).await?;

    let trigger = TriggerContext {
        source: platform.name().to_string(),
        channel_id: channel_id.to_string(),
        user: user.to_string(),
        message: task.to_string(),
        chat_history: history,
    };

// If github_org is set, resolve and clone the repo
let sandbox: Box<dyn Sandbox> = if let Some(ref org) = config.github_org {
    let repo_name = match repo::parse_repo_from_message(task) {
        Some(name) => name,
        None => {
            return Ok(PipelineResult {
                output: "Setup failed: could not identify a target repo".to_string(),
                status: PipelineStatus::SetupFailed,
                // ...
            });
        }
    };

    let full_name = format!("{org}/{repo_name}");
    repo::validate_org(&full_name, org)?;

    // Try pool-first acquisition (warm sandbox), then cold clone
    #[cfg(feature = "daytona")]
    if let Some(ref pool) = config.pool {
        match pool.acquire(&full_name).await {
            Some(ws) => Box::new(PooledSandbox::new(ws, pool, config.base_branch.clone())),
            None => {
                // Fallback to Daytona cold clone or LocalSandbox
            }
        }
    }
} else {
    Box::new(LocalSandbox::from_path(config.repo_dir.clone()))
};

let plane_issue_id = if let Some(plane_cfg) = &config.plane {
    match PlaneClient::new(plane_cfg.clone()) {
        Ok(client) => {
            let desc = format!("<p>Task from {user} via {}: {task}</p>", platform.name());
            match client.create_issue(task, &desc).await {
                Ok(id) => {
                    info!(issue_id = %id, "created Plane issue");
                    Some(id)
                }
                Err(e) => {
                    warn!("failed to create Plane issue: {e}");
                    None
                }
            }
        }
        Err(e) => {
            warn!("failed to create Plane client: {e}");
            None
        }
    }
} else {
    None
};

let branch_slug = if config.dry_run {
    crate::git::slugify(task)
} else {
    generate_branch_slug(task, config.trace_dir.as_ref()).await
};

let mut git = GitOps::new(&*sandbox, config.base_branch.clone());
if let Err(e) = git.create_branch_from_slug(&branch_slug).await {
    error!("git branch creation failed: {e:#}");
    sandbox.destroy().await?;
    return Ok(PipelineResult {
        output: format!("Setup failed: could not create git branch: {e:#}"),
        pr_url: None,
        plane_issue_id,
        ci_passed: false,
        rounds_used: 0,
        status: PipelineStatus::SetupFailed,
    });
}

progress(platform, channel_id, "Analyzing task...").await;
let complexity = classify_task(task, config.dry_run, config.trace_dir.as_ref()).await;
info!(?complexity, "task classified");

match complexity {
    TaskComplexity::Standard => {
        progress(platform, channel_id, "Planning approach...").await;
        let (bp, ctx) = build_tdd_blueprint(&trigger, config, &working_dir)?;
        match BlueprintRunner::new(ctx, &*sandbox).run(&bp).await {
            Ok(ctx) => {
                last_output = ctx.last_output.clone().unwrap_or_default();
                tdd_tests_passed = ctx.last_exit_code == Some(0);
            }
            Err(e) => {
                return Ok(PipelineResult {
                    output: format!("Agent failed in TDD pipeline: {e}"),
                    status: PipelineStatus::AgentFailed,
                    // ...
                });
            }
        }
    }
    TaskComplexity::BugFix => {
        progress(platform, channel_id, "Investigating bug...").await;
        let (bp, ctx) = build_diagnostic_blueprint(&trigger, config, &working_dir)?;
        // ... similar execution
    }
    TaskComplexity::Simple => {
        progress(platform, channel_id, "Working on it...").await;
        let (bp, ctx) = build_main_blueprint(&trigger, config, &working_dir)?;
        // ... similar execution
    }
}

let changed = git.changed_files().await.unwrap_or_default();
let run_ci = if changed.is_empty() {
    info!("no files changed — skipping CI");
    false
} else {
    let needs = needs_ci(&changed);
    if needs {
        info!(file_count = changed.len(), "code changes detected → running CI");
    } else {
        info!(file_count = changed.len(), files = ?changed, "docs-only changes → skipping CI");
    }
    needs
};

if run_ci {
    for round in 1..=config.max_ci_rounds {
        rounds_used = round;
        info!(round, "CI round");

        if round > 1 {
            // Fix round: agent gets test failure output
            let (bp, ctx) = build_fix_blueprint(&trigger, config, &last_test_output, &working_dir)?;
            BlueprintRunner::new(ctx, &*sandbox).run(&bp).await?;
        }

        // Run lint + test
        let ci_bp = Blueprint::new("magpie-ci")
            .add_step(Step {
                name: "lint-check".to_string(),
                kind: StepKind::Shell(ShellStep::new(lint_cmd).with_args(lint_args)),
                condition: Condition::Always,
                continue_on_error: true,
            })
            .add_step(Step {
                name: "test".to_string(),
                kind: StepKind::Shell(ShellStep::new(test_cmd).with_args(test_args)),
                condition: Condition::Always,
                continue_on_error: true,
            });

        match BlueprintRunner::new(ci_ctx, &*sandbox).run(&ci_bp).await {
            Ok(ctx) => {
                if ctx.last_exit_code == Some(0) {
                    ci_passed = true;
                    info!(round, "tests passed");
                    break;
                } else {
                    warn!(round, "tests failed");
                }
            }
            Err(e) => warn!(round, error = %e, "CI blueprint error"),
        }
    }
}

progress(platform, channel_id, "Creating pull request...").await;
let pr_url = if config.dry_run {
    None
} else {
    let diff = git.diff_content().await?;
    if diff.trim().is_empty() {
        None
    } else {
        let commit_message = generate_commit_message(
            task,
            &diff,
            config.trace_dir.as_ref()
        ).await?;
        info!(commit_msg = %commit_message, "agent-generated commit message");
        commit_push_pr(&git, task, &changed, ci_passed, &commit_message).await?
    }
};

if let (Some(issue_id), Some(plane_cfg)) = (&plane_issue_id, &config.plane) {
    if let Ok(client) = PlaneClient::new(plane_cfg.clone()) {
        let state = if ci_passed { "done" } else { "in_progress" };
        let _ = client
            .update_issue(
                issue_id,
                &IssueUpdate {
                    state: Some(state.to_string()),
                    ..Default::default()
                },
            )
            .await;

        if let Some(url) = &pr_url {
            let comment = format!("<p>PR created: <a href=\"{url}\">{url}</a></p>");
            let _ = client.add_comment(issue_id, &comment).await;
        }
    }
}

// Switch back to base branch so repo is clean for next run
if let Err(e) = git.checkout_base().await {
    warn!("failed to checkout base branch: {e}");
}

// Destroy sandbox (clean up temp resources)
if let Err(e) = sandbox.destroy().await {
    warn!("failed to destroy sandbox: {e}");
}

let status = if ci_passed {
    PipelineStatus::Success
} else if pr_url.is_some() {
    PipelineStatus::PartialSuccess
} else {
    PipelineStatus::AgentFailed
};

Ok(PipelineResult {
    output: last_output,
    pr_url,
    plane_issue_id,
    ci_passed,
    rounds_used,
    status,
})