bot_forge/execution/

executor.rs

//! Validated execution entry point. The executor receives a frozen plan and its lowered runtime.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::atomic::AtomicUsize;
use std::sync::{Arc, Mutex};

use crate::backends::typed::CommandInvocation;
use crate::cancellation::CancellationSession;
use crate::config::schema::{InstallSpec, NetworkPolicy};
use crate::error::ForgeError;
use crate::execution::apt_mirror::{AptSourceOverride, configure_for_install};
use crate::execution::command::{ShellDisplayMode, run_invocation_labeled};
use crate::execution::environment::{
    apply_planned_environment, refresh_install_finish_environment,
};
use crate::execution::install::{
    finalize_scheduled_install, install_skill_item, install_typed_tool, record_tool_install,
    run_certificate_preflight, select_install_components, select_outdated_components,
};
use crate::execution::managed_cargo::{CoordinatedCargoResources, install_coordinated};
use crate::execution::runtime::from_plan;
use crate::execution::scheduler::{
    CancellationToken, NodeOutcome, NodeRunner, ResourceBudget, ResourceCoordinator, Scheduler,
};
use crate::execution::verifier::{preview_install, preview_plan};
use crate::model::{
    BackendKind, InstallConfig, InstallOptions, InstallReport, RegistryEntry, ToolDef,
    ToolInstallOutcome, ToolInstallResult,
};
use crate::paths::managed_bin_dir;
use crate::planning::{ExecutionNode, ExecutionPlan, NodeKind};
use crate::state::cache::acquire_usage_lease;
use crate::telemetry::{estimates, flush, record_cancellation, record_failure};
use crate::ui::{StatusKind, print_install_preview, stderr_status, stdout_status};
use crate::util::exe_name;

/// Frozen plan, approved selection, and runtime options required by [`execute`].
///
/// [`execute`] validates dependency closure again before performing side effects; callers cannot
/// use this carrier to inject components outside `plan`.
pub struct PreparedExecution {
    /// Immutable plan whose hashes and nodes define the execution contract.
    pub plan: ExecutionPlan,
    /// Runtime interaction and output options.
    pub options: InstallOptions,
    /// Component selection approved by the caller or interactive preview.
    pub selection: ExecutionSelection,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Component IDs approved for execution; dependencies are added before execution begins.
pub struct ExecutionSelection {
    /// Selected component IDs in plan order.
    pub components: Vec<String>,
    /// Installed components explicitly approved for replacement because their versions are older.
    pub reinstall: Vec<String>,
}

impl ExecutionSelection {
    /// Select every component in plan order.
    pub fn all(plan: &ExecutionPlan) -> Self {
        Self {
            components: plan
                .components
                .iter()
                .map(|component| component.id.clone())
                .collect(),
            reinstall: Vec::new(),
        }
    }

    fn dependency_closed(&self, plan: &ExecutionPlan) -> Result<Self, ForgeError> {
        let known = plan
            .components
            .iter()
            .map(|component| (component.id.as_str(), component))
            .collect::<std::collections::BTreeMap<_, _>>();
        let mut selected = self.components.iter().cloned().collect::<BTreeSet<_>>();
        for id in &self.reinstall {
            if !known.contains_key(id.as_str()) {
                return Err(ForgeError::Config(format!(
                    "execution reinstall selection contains an unplanned component: {id}"
                )));
            }
            if !selected.contains(id) {
                return Err(ForgeError::Config(format!(
                    "execution reinstall selection is not part of the approved components: {id}"
                )));
            }
        }
        let mut pending = self.components.clone();
        while let Some(id) = pending.pop() {
            let component = known.get(id.as_str()).ok_or_else(|| {
                ForgeError::Config(format!(
                    "execution selection contains an unplanned component: {id}"
                ))
            })?;
            for dependency in &component.dependencies {
                if selected.insert(dependency.clone()) {
                    pending.push(dependency.clone());
                }
            }
        }
        Ok(Self {
            components: plan
                .components
                .iter()
                .filter(|component| selected.contains(&component.id))
                .map(|component| component.id.clone())
                .collect(),
            reinstall: self.reinstall.clone(),
        })
    }
}

/// Select components automatically or through the interactive installer preview.
///
/// # Errors
///
/// Returns [`ForgeError`] when detection, interactive input, or dependency-closure validation
/// fails.
pub fn select(plan: &ExecutionPlan, yes: bool) -> Result<ExecutionSelection, ForgeError> {
    let preview = preview_plan(plan)?;
    if yes {
        let mut selection = ExecutionSelection::all(plan);
        selection.reinstall = preview
            .tools
            .iter()
            .filter(|status| status.outdated && status.installable)
            .map(|status| status.name.clone())
            .collect();
        return Ok(selection);
    }
    let mut components = select_install_components(&preview, plan)?;
    let reinstall = select_outdated_components(&preview)?;
    components.extend(reinstall.iter().cloned());
    components.sort();
    components.dedup();
    let selection = ExecutionSelection {
        components,
        reinstall,
    };
    selection.dependency_closed(plan)
}

/// Execute an approved frozen plan and return its final installation report.
///
/// # Errors
///
/// Returns [`ForgeError`] for invalid selections, detection failures, unavailable backends,
/// scheduler or installation failures, cancellation-related persistence failures, or final report
/// construction failures.
pub fn execute(prepared: PreparedExecution) -> Result<InstallReport, ForgeError> {
    let started = std::time::Instant::now();
    let cancellation = CancellationSession::begin();
    let _cache_lease = acquire_usage_lease()?;
    let selection = prepared.selection.dependency_closed(&prepared.plan)?;
    let runtime = from_plan(&prepared.plan);
    run_certificate_preflight(
        &prepared.plan,
        if prepared.options.status_bar {
            ShellDisplayMode::StatusBar
        } else {
            ShellDisplayMode::Plain
        },
    )?;
    let preview = preview_install(&runtime)?;
    print_install_preview(&preview);
    let selected = selection
        .components
        .iter()
        .cloned()
        .collect::<BTreeSet<_>>();
    let reinstall = selection.reinstall.iter().cloned().collect::<BTreeSet<_>>();
    let missing_tools = tools_requiring_install(&preview, &selected, &reinstall);
    let missing_skills = preview
        .missing_skills()
        .into_iter()
        .filter(|status| selected.contains(&status.name))
        .map(|status| status.name.clone())
        .collect::<BTreeSet<_>>();
    let not_installable = preview
        .missing_tools()
        .into_iter()
        .filter(|status| selected.contains(&status.name) && !status.installable)
        .map(|status| status.name.as_str())
        .collect::<Vec<_>>();
    if !not_installable.is_empty() {
        return Err(ForgeError::Config(format!(
            "selected dependency is missing an installation backend: {}",
            not_installable.join(", ")
        )));
    }
    if missing_tools.is_empty() && missing_skills.is_empty() {
        stdout_status(
            StatusKind::Success,
            "All selected tools and skills are already installed.",
        );
        return finalize_scheduled_install(&runtime, &preview, Vec::new(), Vec::new(), started);
    }

    let required_apt_packages = required_apt_packages(&runtime, &selected, &missing_tools);
    let apt_source = configure_for_install(
        &prepared.plan,
        &runtime,
        &prepared.options,
        &required_apt_packages,
    )?;

    let completed = run_selected_plan(
        &prepared.plan,
        &runtime,
        ScheduledSelection {
            components: selected,
            missing_tools,
            missing_skills,
            apt_source,
        },
        &prepared.options,
        cancellation.token(),
    );
    flush();
    let completed = completed?;
    refresh_install_finish_environment(&runtime)?;
    finalize_scheduled_install(
        &runtime,
        &preview,
        completed.tools,
        completed.entries,
        started,
    )
}

fn tools_requiring_install(
    preview: &crate::model::InstallPreview,
    selected: &BTreeSet<String>,
    reinstall: &BTreeSet<String>,
) -> BTreeSet<String> {
    preview
        .tools
        .iter()
        .filter(|status| {
            selected.contains(&status.name)
                && (!status.installed || (status.outdated && reinstall.contains(&status.name)))
        })
        .map(|status| status.name.clone())
        .collect()
}

struct ScheduledResults {
    tools: Vec<ToolInstallResult>,
    entries: Vec<RegistryEntry>,
}

struct ScheduledSelection {
    components: BTreeSet<String>,
    missing_tools: BTreeSet<String>,
    missing_skills: BTreeSet<String>,
    apt_source: Option<AptSourceOverride>,
}

fn required_apt_packages(
    runtime: &InstallConfig,
    selected: &BTreeSet<String>,
    missing_tools: &BTreeSet<String>,
) -> BTreeSet<String> {
    runtime
        .tools
        .iter()
        .filter(|tool| selected.contains(&tool.name) && missing_tools.contains(&tool.name))
        .filter_map(|tool| match tool.install.as_ref() {
            Some(InstallSpec::Apt(apt)) => Some(&apt.packages),
            _ => None,
        })
        .flatten()
        .cloned()
        .collect()
}

fn run_selected_plan(
    plan: &ExecutionPlan,
    runtime: &InstallConfig,
    selection: ScheduledSelection,
    options: &InstallOptions,
    cancellation: CancellationToken,
) -> Result<ScheduledResults, ForgeError> {
    let budget = ResourceBudget::for_plan(plan);
    let cargo_components = runtime
        .tools
        .iter()
        .filter(|tool| selection.components.contains(&tool.name))
        .filter(|tool| selection.missing_tools.contains(&tool.name))
        .filter(|tool| matches!(tool.install, Some(InstallSpec::Cargo(_))))
        .count();
    let parallelism = CargoParallelism::for_run(plan, &budget, cargo_components.max(1));
    if options.status_bar && cargo_components > 0 {
        stderr_status(
            StatusKind::Info,
            &format!(
                "Cargo resources: {} tools · {} concurrent builds · {} jobs/build · {} CPU tokens",
                parallelism.processes,
                parallelism.build_slots,
                parallelism.cargo_jobs,
                parallelism.cpu_tokens
            ),
        );
    }
    let runner = Arc::new(PlanNodeRunner {
        plan: plan.clone(),
        runtime: runtime.clone(),
        profile: options.profile.as_str().to_string(),
        selected: selection.components,
        missing_tools: selection.missing_tools,
        missing_skills: selection.missing_skills,
        display_mode: if options.status_bar {
            ShellDisplayMode::StatusBar
        } else {
            ShellDisplayMode::Plain
        },
        tool_results: Mutex::new(Vec::new()),
        entries: Mutex::new(Vec::new()),
        error: Mutex::new(None),
        cargo_jobs: parallelism.cargo_jobs,
        cargo_build_slots: parallelism.build_slots,
        cargo_remaining: AtomicUsize::new(cargo_components),
        apt_source: selection.apt_source,
    });
    let reports = Scheduler::new(plan.policy.max_parallel.max(1), budget)
        .with_priorities(cargo_priorities(plan, runtime, &runner.missing_tools))
        .execute(plan, Arc::clone(&runner), cancellation)?;
    if let Some(error) = runner
        .error
        .lock()
        .map_err(|_| ForgeError::Command("execution error lock is corrupted".into()))?
        .take()
    {
        return Err(error);
    }
    if reports
        .iter()
        .any(|report| report.outcome == NodeOutcome::Cancelled)
    {
        return Err(ForgeError::Command(
            "installation execution was cancelled".into(),
        ));
    }
    if let Some(report) = reports
        .iter()
        .find(|report| matches!(report.outcome, NodeOutcome::Failed | NodeOutcome::Blocked))
    {
        return Err(ForgeError::Command(format!(
            "execution plan node did not complete: {}",
            report.node
        )));
    }
    let mut tools = runner
        .tool_results
        .lock()
        .map_err(|_| ForgeError::Command("installation result lock is corrupted".into()))?
        .clone();
    tools.sort_by(|left, right| left.name.cmp(&right.name));
    let mut entries = runner
        .entries
        .lock()
        .map_err(|_| ForgeError::Command("managed record result lock is corrupted".into()))?
        .clone();
    entries.sort_by_key(RegistryEntry::stable_id);
    Ok(ScheduledResults { tools, entries })
}

fn cargo_priorities(
    plan: &ExecutionPlan,
    runtime: &InstallConfig,
    components: &BTreeSet<String>,
) -> std::collections::BTreeMap<String, u64> {
    let estimates = estimates();
    let priorities = runtime
        .tools
        .iter()
        .filter(|tool| components.contains(&tool.name))
        .filter_map(|tool| {
            let InstallSpec::Cargo(specification) = tool.install.as_ref()? else {
                return None;
            };
            let source = specification.source_digest();
            let fingerprint =
                specification.artifact_fingerprint(&source, &specification.lock_digest());
            Some((
                tool.name.as_str(),
                estimates.get(&fingerprint)?.average_build_ms,
            ))
        })
        .collect::<std::collections::BTreeMap<_, _>>();
    let own = plan
        .nodes
        .iter()
        .filter_map(|node| {
            priorities
                .get(node.component.as_str())
                .copied()
                .map(|priority| (node.id.clone(), priority.max(1)))
        })
        .collect::<std::collections::BTreeMap<_, _>>();
    let mut dependents = std::collections::BTreeMap::<String, Vec<String>>::new();
    for node in &plan.nodes {
        for dependency in &node.dependencies {
            dependents
                .entry(dependency.clone())
                .or_default()
                .push(node.id.clone());
        }
    }
    let mut critical = std::collections::BTreeMap::new();
    for node in plan.nodes.iter().rev() {
        let downstream = dependents
            .get(&node.id)
            .into_iter()
            .flatten()
            .filter_map(|dependent| critical.get(dependent).copied())
            .max()
            .unwrap_or(0);
        let own = own.get(&node.id).copied().unwrap_or(1);
        critical.insert(node.id.clone(), own.saturating_add(downstream));
    }
    critical
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct CargoParallelism {
    processes: usize,
    build_slots: usize,
    cpu_tokens: u32,
    cargo_jobs: usize,
}

impl CargoParallelism {
    fn for_run(plan: &ExecutionPlan, budget: &ResourceBudget, components: usize) -> Self {
        let cpu_tokens = budget.capacity("cpu").max(1);
        let memory_slots = budget.capacity("memory-mib") / 512;
        let build_slots = components
            .max(1)
            .min(plan.policy.max_parallel.max(1))
            .min(budget.capacity("network").max(1) as usize)
            .min(budget.capacity("disk-io").max(1) as usize)
            .min(memory_slots.max(1) as usize)
            .min(cpu_tokens as usize);
        let processes = components.max(1).min(plan.policy.max_parallel.max(1));
        Self {
            processes,
            build_slots,
            cpu_tokens,
            cargo_jobs: (cpu_tokens as usize / build_slots).max(1),
        }
    }
}

struct PlanNodeRunner {
    plan: ExecutionPlan,
    runtime: InstallConfig,
    profile: String,
    selected: BTreeSet<String>,
    missing_tools: BTreeSet<String>,
    missing_skills: BTreeSet<String>,
    display_mode: ShellDisplayMode,
    tool_results: Mutex<Vec<ToolInstallResult>>,
    entries: Mutex<Vec<RegistryEntry>>,
    error: Mutex<Option<ForgeError>>,
    cargo_jobs: usize,
    cargo_build_slots: usize,
    cargo_remaining: AtomicUsize,
    apt_source: Option<AptSourceOverride>,
}

impl NodeRunner for PlanNodeRunner {
    fn run(
        &self,
        node: &ExecutionNode,
        cancellation: &CancellationToken,
    ) -> Result<NodeOutcome, ForgeError> {
        if node.kind == NodeKind::Environment {
            return apply_planned_environment(&self.runtime)
                .map(|()| NodeOutcome::Completed)
                .or_else(|error| {
                    self.store_error(error);
                    Ok(NodeOutcome::Failed)
                });
        }
        if !self.selected.contains(&node.component) {
            return Ok(NodeOutcome::Skipped);
        }
        if cancellation.is_cancelled() {
            return Ok(NodeOutcome::Cancelled);
        }
        if node.kind != NodeKind::Acquire {
            self.store_error(ForgeError::Config(format!(
                "execution plan contains an unsupported node type: {}",
                node.id
            )));
            return Ok(NodeOutcome::Failed);
        }
        if let Some(tool) = self
            .runtime
            .tools
            .iter()
            .find(|tool| tool.name == node.component)
        {
            if !self.missing_tools.contains(&tool.name) {
                return Ok(NodeOutcome::Skipped);
            }
            return self.install_tool(tool, None, cancellation);
        }
        if let Some(skill) = self
            .runtime
            .skills
            .iter()
            .find(|skill| skill.name == node.component)
        {
            if !self.missing_skills.contains(&skill.name) {
                return Ok(NodeOutcome::Skipped);
            }
            return match install_skill_item(skill, &self.profile, false, self.plan.policy.network) {
                Ok(entries) => {
                    self.entries
                        .lock()
                        .map_err(|_| ForgeError::Command("skill result lock is corrupted".into()))?
                        .extend(entries);
                    Ok(NodeOutcome::Completed)
                }
                Err(error) => {
                    self.store_error(error);
                    Ok(NodeOutcome::Failed)
                }
            };
        }
        self.store_error(ForgeError::Config(format!(
            "execution plan component {} has no runtime definition",
            node.component
        )));
        Ok(NodeOutcome::Failed)
    }

    fn manages_resources(&self, node: &ExecutionNode) -> bool {
        node.kind == NodeKind::Acquire
            && self.selected.contains(&node.component)
            && self.missing_tools.contains(&node.component)
            && self
                .runtime
                .tools
                .iter()
                .find(|tool| tool.name == node.component)
                .and_then(|tool| tool.install.as_ref())
                .is_some_and(|install| matches!(install, InstallSpec::Cargo(_)))
    }

    fn run_with_resources(
        &self,
        node: &ExecutionNode,
        cancellation: &CancellationToken,
        coordinator: &ResourceCoordinator,
    ) -> Result<NodeOutcome, ForgeError> {
        if !self.manages_resources(node) {
            return self.run(node, cancellation);
        }
        let Some(tool) = self
            .runtime
            .tools
            .iter()
            .find(|tool| tool.name == node.component)
        else {
            self.store_error(ForgeError::Config(format!(
                "execution plan Cargo component {} has no runtime definition",
                node.component
            )));
            return Ok(NodeOutcome::Failed);
        };
        let Some(InstallSpec::Cargo(specification)) = tool.install.as_ref() else {
            self.store_error(ForgeError::Config(format!(
                "component {} is not a Cargo typed backend",
                node.component
            )));
            return Ok(NodeOutcome::Failed);
        };
        if cancellation.is_cancelled() {
            record_cancellation(
                &specification.artifact_fingerprint(
                    &specification.source_digest(),
                    &specification.lock_digest(),
                ),
                &tool.name,
            );
            return Ok(NodeOutcome::Cancelled);
        }
        match install_coordinated(
            &self.plan,
            tool,
            specification,
            self.display_mode,
            self.cargo_jobs,
            CoordinatedCargoResources {
                coordinator,
                remaining_builds: &self.cargo_remaining,
                baseline_jobs: self.cargo_jobs,
                max_builds: self.cargo_build_slots,
                cancellation,
            },
        ) {
            Ok(result) => self.store_tool_result(result, tool, cancellation),
            Err(error) => {
                let fingerprint = specification.artifact_fingerprint(
                    &specification.source_digest(),
                    &specification.lock_digest(),
                );
                if cancellation.is_cancelled() {
                    record_cancellation(&fingerprint, &tool.name);
                } else {
                    record_failure(&fingerprint, &tool.name);
                }
                self.store_error(error);
                Ok(NodeOutcome::Failed)
            }
        }
    }
}

impl PlanNodeRunner {
    fn install_tool(
        &self,
        tool: &ToolDef,
        coordinator: Option<&ResourceCoordinator>,
        cancellation: &CancellationToken,
    ) -> Result<NodeOutcome, ForgeError> {
        if cancellation.is_cancelled() {
            return Ok(NodeOutcome::Cancelled);
        }
        let Some(specification) = tool.install.as_ref() else {
            self.store_error(ForgeError::Config(format!(
                "tool {} is missing an installation backend",
                tool.name
            )));
            return Ok(NodeOutcome::Failed);
        };
        let result = match (specification, coordinator) {
            (InstallSpec::Cargo(cargo), Some(coordinator)) => install_coordinated(
                &self.plan,
                tool,
                cargo,
                self.display_mode,
                self.cargo_jobs,
                CoordinatedCargoResources {
                    coordinator,
                    remaining_builds: &self.cargo_remaining,
                    baseline_jobs: self.cargo_jobs,
                    max_builds: self.cargo_build_slots,
                    cancellation,
                },
            ),
            _ => install_typed_tool(
                &self.plan,
                tool,
                specification,
                self.display_mode,
                None,
                self.apt_source
                    .as_ref()
                    .map(|source| source.source_file.as_path()),
                self.apt_source
                    .as_ref()
                    .map(|source| source.lists_dir.as_path()),
            ),
        };
        match result {
            Ok(result) => self.store_tool_result(result, tool, cancellation),
            Err(error) => {
                self.store_error(error);
                Ok(NodeOutcome::Failed)
            }
        }
    }
    fn store_tool_result(
        &self,
        result: ToolInstallResult,
        tool: &ToolDef,
        cancellation: &CancellationToken,
    ) -> Result<NodeOutcome, ForgeError> {
        if result.outcome == ToolInstallOutcome::Installed
            && tool.name == "rust-bot"
            && let Err(error) = self.install_rust_bot_skills(cancellation)
        {
            self.store_error(error);
            return Ok(NodeOutcome::Failed);
        }
        if result.outcome == ToolInstallOutcome::Installed
            && !matches!(tool.backend(), Some(BackendKind::Cargo | BackendKind::Git))
        {
            record_tool_install(tool, &self.profile)?;
        }
        let outcome = if result.outcome == ToolInstallOutcome::VerificationFailed {
            NodeOutcome::Failed
        } else {
            NodeOutcome::Completed
        };
        self.tool_results
            .lock()
            .map_err(|_| ForgeError::Command("installation result lock is corrupted".into()))?
            .push(result);
        Ok(outcome)
    }

    fn store_error(&self, error: ForgeError) {
        if let Ok(mut slot) = self.error.lock()
            && slot.is_none()
        {
            *slot = Some(error);
        }
    }

    fn install_rust_bot_skills(&self, cancellation: &CancellationToken) -> Result<(), ForgeError> {
        if cancellation.is_cancelled() {
            return Err(ForgeError::Command(
                "rust-bot skill installation was cancelled".into(),
            ));
        }
        if self.plan.policy.network != NetworkPolicy::Online {
            return Err(ForgeError::Config(
                "rust-bot skill installation requires network=online".into(),
            ));
        }
        let invocation = rust_bot_skill_install_invocation();
        run_invocation_labeled(
            "rust-bot skill packages",
            &invocation,
            self.display_mode,
            None,
        )
        .map_err(|error| {
            ForgeError::Command(format!("rust-bot skill installation failed: {error}"))
        })
    }
}

fn rust_bot_skill_install_invocation() -> CommandInvocation {
    let managed = managed_bin_dir().join(exe_name("rust-bot"));
    let program = if managed.is_file() {
        managed.display().to_string()
    } else {
        "rust-bot".to_string()
    };
    CommandInvocation {
        program,
        args: vec!["install".into(), "stable".into()],
        env: BTreeMap::new(),
        current_dir: None,
        clear_env: false,
        null_stdin: true,
        timeout_secs: Some(300),
        inactivity_timeout_secs: Some(120),
        idempotent: true,
        success_codes: vec![0],
        stdout_contains: None,
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeSet;
    use std::sync::Mutex;
    use std::sync::atomic::AtomicUsize;

    use crate::config::schema::{AptInstall, InstallSpec, NetworkPolicy};
    use crate::execution::command::ShellDisplayMode;
    use crate::execution::executor::{
        CargoParallelism, ExecutionSelection, PlanNodeRunner, required_apt_packages,
        rust_bot_skill_install_invocation, tools_requiring_install,
    };
    use crate::execution::scheduler::{CancellationToken, NodeOutcome, NodeRunner, ResourceBudget};
    use crate::model::{InstallConfig, InstallKind, InstallPreview, ToolDef, ToolStatus};
    use crate::planning::{
        ExecutionNode, ExecutionPlan, NodeKind, PlanPolicy, ResolvedComponent, TargetPlatform,
    };

    fn component(id: &str, dependencies: &[&str]) -> ResolvedComponent {
        ResolvedComponent {
            id: id.into(),
            display_name: None,
            version: None,
            optional: false,
            allow_insecure_hosts: Vec::new(),
            kind: InstallKind::Tool,
            variant: None,
            requested_by: Vec::new(),
            dependencies: dependencies.iter().map(|value| (*value).into()).collect(),
            provides: Vec::new(),
            conflicts: Vec::new(),
            detect: None,
            install: None,
            verify: None,
            source: None,
            revision: None,
            agents: Vec::new(),
        }
    }

    fn plan() -> ExecutionPlan {
        ExecutionPlan {
            profile: "test".into(),
            target: TargetPlatform::host(),
            config_hash: "a".repeat(64),
            plan_hash: "b".repeat(64),
            policy: PlanPolicy {
                network: NetworkPolicy::Online,
                max_parallel: 1,
                max_downloads: 1,
                max_memory_mib: None,
            },
            certificate_preflight: None,
            components: vec![component("base", &[]), component("tool", &["base"])],
            unsupported_components: Vec::new(),
            nodes: Vec::new(),
            environment: Vec::new(),
            apt_mirror: None,
            origins: Default::default(),
        }
    }

    #[test]
    fn selection_is_closed_over_plan_dependencies() {
        let selection = ExecutionSelection {
            components: vec!["tool".into()],
            reinstall: Vec::new(),
        }
        .dependency_closed(&plan())
        .unwrap();
        assert_eq!(selection.components, ["base", "tool"]);
    }

    #[test]
    fn selection_rejects_components_outside_the_frozen_plan() {
        let error = ExecutionSelection {
            components: vec!["unknown".into()],
            reinstall: Vec::new(),
        }
        .dependency_closed(&plan())
        .unwrap_err();
        assert!(error.to_string().contains("unplanned component"));
    }

    #[test]
    fn reinstall_selection_must_be_explicitly_approved_and_planned() {
        let error = ExecutionSelection {
            components: vec!["tool".into()],
            reinstall: vec!["base".into()],
        }
        .dependency_closed(&plan())
        .unwrap_err();
        assert!(
            error
                .to_string()
                .contains("not part of the approved components")
        );
    }

    #[test]
    fn overwrite_approval_controls_the_actual_scheduled_tool_set() {
        let preview = InstallPreview {
            tools: ["first", "second"]
                .into_iter()
                .map(|name| ToolStatus {
                    name: name.into(),
                    display_name: None,
                    optional: false,
                    installed: true,
                    version: Some("1.0.0".into()),
                    required_version: Some("2.0.0".into()),
                    outdated: true,
                    installable: true,
                })
                .collect(),
            skills: Vec::new(),
        };
        let selected = BTreeSet::from(["first".into(), "second".into()]);

        assert!(tools_requiring_install(&preview, &selected, &BTreeSet::new()).is_empty());
        assert_eq!(
            tools_requiring_install(
                &preview,
                &selected,
                &BTreeSet::from(["first".into(), "second".into()]),
            ),
            selected
        );
        assert_eq!(
            tools_requiring_install(&preview, &selected, &BTreeSet::from(["second".into()]),),
            BTreeSet::from(["second".into()])
        );
        let mut refreshed = preview;
        refreshed.tools[1].outdated = false;
        refreshed.tools[1].version = Some("2.0.0".into());
        assert!(
            tools_requiring_install(&refreshed, &selected, &BTreeSet::from(["second".into()]),)
                .is_empty()
        );
    }

    #[test]
    fn custom_overwrite_selection_reaches_only_the_chosen_runner_path() {
        let preview = InstallPreview {
            tools: ["first", "second"]
                .into_iter()
                .map(|name| ToolStatus {
                    name: name.into(),
                    display_name: None,
                    optional: false,
                    installed: true,
                    version: Some("1.0.0".into()),
                    required_version: Some("2.0.0".into()),
                    outdated: true,
                    installable: true,
                })
                .collect(),
            skills: Vec::new(),
        };
        let selected = BTreeSet::from(["first".into(), "second".into()]);
        let chosen = BTreeSet::from(["second".into()]);
        let runtime = InstallConfig {
            tools: ["first", "second"]
                .into_iter()
                .map(|name| ToolDef {
                    name: name.into(),
                    ..ToolDef::default()
                })
                .collect(),
            ..InstallConfig::default()
        };
        let runner = PlanNodeRunner {
            plan: plan(),
            runtime,
            profile: "test".into(),
            selected: selected.clone(),
            missing_tools: tools_requiring_install(&preview, &selected, &chosen),
            missing_skills: BTreeSet::new(),
            display_mode: ShellDisplayMode::Plain,
            tool_results: Mutex::new(Vec::new()),
            entries: Mutex::new(Vec::new()),
            error: Mutex::new(None),
            cargo_jobs: 1,
            cargo_build_slots: 1,
            cargo_remaining: AtomicUsize::new(0),
            apt_source: None,
        };
        let run = |component: &str| {
            runner.run(
                &ExecutionNode {
                    id: format!("{component}:acquire"),
                    component: component.into(),
                    kind: NodeKind::Acquire,
                    dependencies: Vec::new(),
                    resources: Vec::new(),
                },
                &CancellationToken::default(),
            )
        };
        assert_eq!(run("first").unwrap(), NodeOutcome::Skipped);
        assert_eq!(run("second").unwrap(), NodeOutcome::Failed);
        assert!(
            runner
                .error
                .lock()
                .unwrap()
                .as_ref()
                .is_some_and(|error| error.to_string().contains("second"))
        );
    }

    #[test]
    fn apt_probe_is_scoped_to_selected_missing_packages() {
        let mut runtime = InstallConfig::default();
        for (name, package) in [
            ("selected", "cmake"),
            ("installed", "git"),
            ("other", "ninja"),
        ] {
            runtime.tools.push(ToolDef {
                name: name.into(),
                display_name: None,
                version: None,
                optional: false,
                allow_insecure_hosts: Vec::new(),
                detect: None,
                install: Some(InstallSpec::Apt(AptInstall {
                    packages: vec![package.into()],
                    update: true,
                })),
                verify: None,
            });
        }
        let selected = BTreeSet::from(["selected".into(), "installed".into()]);
        let missing = BTreeSet::from(["selected".into(), "other".into()]);

        assert_eq!(
            required_apt_packages(&runtime, &selected, &missing),
            BTreeSet::from(["cmake".into()])
        );
    }

    #[test]
    fn cargo_parallelism_coordinates_processes_and_inner_jobs() {
        let mut plan = plan();
        plan.policy.max_parallel = 8;
        plan.policy.max_downloads = 4;
        plan.policy.max_memory_mib = Some(4096);
        let budget = ResourceBudget::for_plan(&plan);
        let allocation = CargoParallelism::for_run(&plan, &budget, 12);
        let cores = std::thread::available_parallelism().map_or(1, usize::from);

        assert_eq!(allocation.processes, 8);
        let expected = cores.min(8).min(budget.capacity("disk-io") as usize).min(4);
        assert_eq!(allocation.build_slots, expected);
        assert_eq!(allocation.cpu_tokens as usize, cores);
        assert!(allocation.build_slots * allocation.cargo_jobs <= cores);
    }

    #[test]
    fn cargo_parallelism_respects_memory_and_download_budgets() {
        let mut plan = plan();
        plan.policy.max_parallel = 8;
        plan.policy.max_downloads = 1;
        plan.policy.max_memory_mib = Some(512);
        let budget = ResourceBudget::for_plan(&plan);

        let allocation = CargoParallelism::for_run(&plan, &budget, 8);
        assert_eq!(allocation.processes, 8);
        assert_eq!(allocation.build_slots, 1);
    }

    #[test]
    fn rust_bot_skill_install_targets_latest_stable_skills() {
        let invocation = rust_bot_skill_install_invocation();
        assert_eq!(invocation.args, ["install", "stable"]);
        assert_eq!(invocation.timeout_secs, Some(300));
        assert_eq!(invocation.inactivity_timeout_secs, Some(120));
        assert!(invocation.null_stdin);
        assert!(invocation.idempotent);
    }
}