xianyun_cli/commands/

run.rs

//! `caproni run`: run reloaders in edit mode until interrupted.
//!
//! Port of Go's `cmd/run` package.
//!
//! `runRun`'s setup pipeline (repository cloning, cluster DNS domain
//! resolution) and its concurrent core (`runAndWait`: one goroutine per
//! reloader, signal handling) are both ported for real below. The
//! concurrent core uses [`tokio::task::JoinSet`] for structured
//! concurrency -- see [`execute_reloaders_concurrently`]'s doc comment
//! for the full design and what's intentionally out of scope (the TUI,
//! mirror deployment-copy orchestration). See
//! `PORT_NOTES.md`'s "fifth pass" entry for the complete rationale.

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

use anyhow::Context as _;
use clap::Args;
use xianyun_backends::git::Backend as GitBackend;
use xianyun_backends::kubectl;
use xianyun_cluster::clusterdomain;
use xianyun_core::config::caproni::load_from_context;
use xianyun_core::config::global::Context;
use xianyun_core::config::latest::CaproniConfig;
pub use xianyun_core::config::latest::ReloaderConfig;
use xianyun_executor::exec::CommandRunner;
use xianyun_update::lifecyclehooks::{self, HookType};

#[derive(Args, Debug)]
pub struct RunArgs {}

/// Port of `run.ErrNoEditModeReloaders`/`run.ErrRepositoryNotDefined`.
#[derive(Debug, thiserror::Error)]
pub enum RunError {
    #[error("no reloaders have edit_mode=true, set edit_mode: true on at least one reloader")]
    NoEditModeReloaders,
    #[error("repository {0} referenced by reloader: not defined in config")]
    RepositoryNotDefined(String),
    #[error("unsupported reloader type for `caproni run`: {0:?}")]
    UnsupportedReloaderType(String),
}

/// A single reloader's execution, as an owned, boxed future. Production
/// implementations wrap a blocking call (mirror/skaffold's `run`) in
/// [`tokio::task::spawn_blocking`]; test doubles can be a plain `async`
/// block.
pub type ReloaderFuture = Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send>>;

/// Runs one reloader by name, returning its outcome. Mirrors Go's
/// `runReloader`/`runWithHooks` folded into a single injected function,
/// the same DI pattern `up`/`ca`/`deploy`/`run`'s setup pipeline already
/// use for their `KubectlFactory`/`Connector`/`git::Backend` parameters.
pub type ReloaderRunFn = Arc<dyn Fn(String, ReloaderConfig) -> ReloaderFuture + Send + Sync>;

pub fn run(_args: &RunArgs, ctx: &Context, cli: &crate::Cli) -> anyhow::Result<()> {
    let executor: Arc<dyn CommandRunner + Send + Sync> =
        Arc::new(xianyun_executor::exec::Client::new());
    let git_backend: Arc<dyn GitBackend + Send + Sync> =
        Arc::new(xianyun_backends::git::Client::new(executor.clone()));

    let builder_executor = executor.clone();

    // Resolved up front (pure parsing, no I/O) so `run_run` can build the
    // TUI-routed dispatch once it creates the TUI itself -- see
    // `run_run`'s doc comment on the TUI section for why the `Config` is
    // threaded through rather than a `Dispatch` built here: the `Dispatch`
    // needs `tui.new_writer()`, and the `Tui` doesn't exist yet at this
    // point.
    let logging_config = if xianyun_executor::exec::streamed::is_stderr_tty() {
        let mut config = crate::hooks::resolve_logging_config(cli)?;
        // Ported from `setupTUI`'s `cfgCopy.Color = &colorEnabled`: force
        // color on for the TUI-routed logger regardless of the configured
        // default, since the TUI only ever runs when stderr is already a
        // real terminal.
        config.color = Some(true);
        Some(config)
    } else {
        None
    };

    run_run(
        ctx,
        &executor,
        &git_backend,
        crate::commands::up::connect_kubectl,
        logging_config,
        &move |caproni_cfg, static_vars, kubeconfig, tui_dispatch| {
            build_reloader_run_fn(
                builder_executor.clone(),
                caproni_cfg,
                static_vars,
                kubeconfig,
                tui_dispatch,
            )
        },
    )
}

/// Builds the [`ReloaderRunFn`] `run_run` hands to
/// [`execute_reloaders_concurrently`], given the resolved configuration,
/// cluster kubeconfig, and (if the log-filter TUI is active) the
/// `tracing::Dispatch` each reloader task should log through for the
/// duration of its own work. Injected (rather than hard-coded inside
/// `run_run`) so tests can substitute a fake, instantly-resolving
/// `ReloaderRunFn` without a live cluster, real `mirrord`/`skaffold`
/// binaries, or real subprocess execution -- exactly like every other
/// injected seam in this crate (see the module doc comment).
pub type RunOneBuilder = dyn Fn(
    &CaproniConfig,
    &HashMap<String, String>,
    &str,
    Option<tracing::Dispatch>,
) -> ReloaderRunFn;

/// Port of `run.runRun`: loads the configuration, finds `edit_mode`
/// reloaders, clones their repositories, resolves the cluster's DNS
/// domain, fetches its kubeconfig, runs `edit_mode_start` lifecycle
/// hooks, launches every reloader concurrently via
/// [`execute_reloaders_concurrently`], then runs `edit_mode_stop`
/// lifecycle hooks.
///
/// `git_backend`/`connect`/`build_run_one` are injected (mirroring
/// `up::run_start`'s `KubectlFactory` and `ca`/`deploy`'s `Connector`
/// parameter) so tests can exercise this pipeline without a live
/// cluster, real git clones, or real reloader subprocesses.
///
/// # Errors
///
/// Returns [`RunError::NoEditModeReloaders`] if no reloader has
/// `edit_mode: true`; [`RunError::RepositoryNotDefined`] if a reloader
/// references a repository absent from `repositories:`; an error if
/// configuration loading, cloning, creating the cluster manager,
/// resolving the DNS domain, or getting the kubeconfig fails; and,
/// once every reloader has finished, whatever
/// [`collect_reloader_errors`] reports.
pub fn run_run(
    ctx: &Context,
    executor: &Arc<dyn CommandRunner + Send + Sync>,
    git_backend: &Arc<dyn GitBackend + Send + Sync>,
    connect: clusterdomain::resolve::Connector,
    tui_logging_config: Option<xianyun_executor::logging::Config>,
    build_run_one: &RunOneBuilder,
) -> anyhow::Result<()> {
    crate::drivers::ensure_registered();

    let container = load_from_context(ctx).context("loading configuration")?;

    tracing::info!(config_file = %container.global.config_file, "loading configuration");

    let edit_mode_reloaders = find_edit_mode_reloaders(&container.caproni);

    if edit_mode_reloaders.is_empty() {
        return Err(RunError::NoEditModeReloaders.into());
    }

    tracing::info!(
        count = edit_mode_reloaders.len(),
        "found reloaders with edit mode enabled"
    );

    clone_repos(
        &container.caproni,
        &edit_mode_reloaders,
        git_backend.as_ref(),
    )?;

    let mut container = container;

    let manager = xianyun_cluster::registry::new_manager(
        container.caproni.cluster.clone(),
        container.global.profile_name.clone(),
    )
    .context("creating cluster manager")?;

    let (skipped, err) =
        clusterdomain::resolve::resolve_running(&mut container, manager.as_ref(), connect);

    if let Some(err) = err {
        return Err(anyhow::Error::new(err).context("resolving cluster DNS domain"));
    }

    let container = if skipped.is_none() {
        // `${CAPRONI_CLUSTER_DOMAIN}` expands with the discovered domain.
        let ctx = xianyun_core::config::global::with_context(
            Context::default(),
            container.global.clone(),
        );
        load_from_context(&ctx).context("reloading configuration")?
    } else {
        container
    };

    let edit_mode_reloaders = find_edit_mode_reloaders(&container.caproni);

    let kubeconfig = manager
        .get_kubeconfig()
        .context("getting kubeconfig from cluster")?;

    let static_vars = container.resolver.all();

    run_edit_mode_hooks(
        executor.as_ref(),
        &container.caproni,
        &static_vars,
        &edit_mode_reloaders,
        HookType::EditModeStart,
    )
    .context("running edit-mode-start hooks")?;

    // Ported from `setupTUI`: only started when stderr is a real terminal
    // (checked by the caller -- `tui_logging_config` is `Some` exactly
    // when that was true), matching Go's "if running in a terminal, start
    // the TUI and route the default logger through it" behavior. A
    // piped/non-interactive invocation runs with no TUI at all, logging
    // on whatever dispatcher `dispatch()` already installed as the
    // process-wide default.
    let tui_dispatch = tui_logging_config
        .map(|config| {
            let tui = std::sync::Arc::new(xianyun_executor::logfilter::Tui::new());
            let writer: Box<dyn std::io::Write + Send> = Box::new(tui.new_writer());
            let dispatch = xianyun_executor::logging::new_logger_with_writer(&config, writer)
                .map_err(|err| anyhow::anyhow!("creating TUI logger: {err}"))?;

            let tui_thread = tui.clone();
            let tui_handle = std::thread::spawn(move || {
                let out: Box<dyn std::io::Write + Send> = Box::new(std::io::stdout());
                if let Err(err) = tui_thread.start(out) {
                    tracing::warn!(error = %err, "log filter TUI exited with an error");
                }
            });

            Ok::<_, anyhow::Error>((tui, tui_handle, dispatch))
        })
        .transpose()?;

    let run_one = build_run_one(
        &container.caproni,
        &static_vars,
        &kubeconfig,
        tui_dispatch
            .as_ref()
            .map(|(_, _, dispatch)| dispatch.clone()),
    );

    let runtime = tokio::runtime::Runtime::new().context("creating async runtime")?;
    let result = runtime.block_on(execute_reloaders_concurrently(
        edit_mode_reloaders.clone(),
        run_one,
        Box::pin(wait_for_shutdown_signal()),
    ));

    if let Some((tui, tui_handle, _)) = tui_dispatch {
        tui.quit();
        let _ = tui_handle.join();
    }

    if let Err(err) = run_edit_mode_hooks(
        executor.as_ref(),
        &container.caproni,
        &static_vars,
        &edit_mode_reloaders,
        HookType::EditModeStop,
    ) {
        tracing::error!(error = %err, "running edit-mode-stop hooks");
    }

    result
}

/// Waits for `SIGINT` (Ctrl-C) or, on Unix, `SIGTERM`. Port of
/// `runAndWait`'s signal-handling goroutine (`signal.Notify` +
/// `syscall.SIGINT`/`SIGTERM`).
async fn wait_for_shutdown_signal() {
    #[cfg(unix)]
    {
        let mut sigterm =
            match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
                Ok(s) => s,
                Err(err) => {
                    tracing::warn!(error = %err, "failed to install SIGTERM handler");
                    let _ = tokio::signal::ctrl_c().await;
                    return;
                }
            };

        tokio::select! {
            _ = tokio::signal::ctrl_c() => {}
            _ = sigterm.recv() => {}
        }
    }

    #[cfg(not(unix))]
    {
        let _ = tokio::signal::ctrl_c().await;
    }
}

/// Port of `runAndWait`'s concurrent core: launches every edit-mode
/// reloader's `run_one` concurrently, waits for all of them to finish
/// (whether they succeed, fail, or a shutdown signal arrives first),
/// and collects any real failures.
///
/// Uses [`tokio::task::JoinSet`] for structured concurrency: every
/// spawned task is owned by the `JoinSet` and is guaranteed to be
/// awaited (via `join_next`) before this function returns, unlike Go's
/// bare goroutines-plus-`sync.WaitGroup`, where nothing at the type
/// level stops a goroutine from leaking past `wg.Done()`.
///
/// **Structural gap, not a stub** (see `PORT_NOTES.md`): Go's
/// `runReloader` runs inside a context that's cancelled on shutdown,
/// and its `reloader.Manager.Run(ctx, ...)` implementations react to
/// `ctx.Done()` to actually stop early. This port's
/// `xianyun_deploy::reloader::Manager`-shaped `run` calls (`mirror`/
/// `skaffold`'s concrete managers) take no context/cancellation
/// parameter at all -- they block until the underlying subprocess
/// exits on its own. So when `shutdown` resolves while reloaders are
/// still running, this function has no way to make them stop early
/// either; it can only stop *waiting new work* and, like Go's
/// `wg.Wait()` after `cancelRun()`, still block until every already
/// -running task finishes naturally. In practice (a real terminal
/// delivering `SIGINT` to the whole foreground process group) the
/// underlying `mirrord`/local child processes receive the signal
/// directly regardless of this crate's own cancellation plumbing, which
/// is the same mechanism Go's version ultimately relies on for a
/// process that ignores `ctx.Done()`.
///
/// # Errors
///
/// Returns whatever [`collect_reloader_errors`] reports once every
/// reloader has finished.
pub async fn execute_reloaders_concurrently(
    reloaders: HashMap<String, ReloaderConfig>,
    run_one: ReloaderRunFn,
    shutdown: Pin<Box<dyn Future<Output = ()> + Send>>,
) -> anyhow::Result<()> {
    let mut join_set: tokio::task::JoinSet<(String, anyhow::Result<()>)> =
        tokio::task::JoinSet::new();

    for (name, reloader_config) in reloaders {
        let run_one = run_one.clone();
        let task_name = name.clone();
        join_set.spawn(async move {
            let result = run_one(task_name.clone(), reloader_config).await;
            (task_name, result)
        });
    }

    tracing::info!(
        reloaders = join_set.len(),
        "all edit mode processes started, press Ctrl-C to stop"
    );

    let mut errors = Vec::new();
    let mut shutdown_requested = false;
    let mut shutdown = shutdown;

    loop {
        tokio::select! {
            () = &mut shutdown, if !shutdown_requested => {
                tracing::info!("shutting down edit mode, reverting cluster to production mode");
                shutdown_requested = true;
            }
            joined = join_set.join_next() => {
                match joined {
                    None => break,
                    Some(Ok((name, Ok(())))) => {
                        tracing::info!(reloader = %name, "reloader stopped");
                    }
                    Some(Ok((name, Err(err)))) => {
                        tracing::error!(reloader = %name, error = %err, "reloader error");
                        errors.push(err);
                    }
                    Some(Err(join_err)) => {
                        errors.push(anyhow::anyhow!("reloader task panicked: {join_err}"));
                    }
                }
            }
        }
    }

    tracing::info!("edit mode processes stopped");

    collect_reloader_errors(errors)
}

/// Production [`ReloaderRunFn`]: dispatches to the concrete `mirror`/
/// `skaffold` reloader manager for `reloader_config.reloader_type`,
/// running it (via [`tokio::task::spawn_blocking`], since both
/// managers' `run` blocks synchronously until their subprocess exits)
/// against a fresh [`kubectl::Client`] built from the cluster's
/// kubeconfig.
///
/// **Not ported** (see `PORT_NOTES.md`'s "fifth pass" entry): precondition
/// checks (`reloader.Manager.CheckPreconditions`) and mirror
/// deployment-copy preparation/cleanup
/// (`mirror.Manager.{PrepareFromReloaderConfigs,
/// CleanupCopyDeployments,CleanupStaleMirrordAgents}`). Neither is part
/// of the concurrent execution engine this pass was scoped to port;
/// each is a self-contained follow-up.
///
/// The log filter TUI (`logfilter.NewTUI`) *is* now wired, in
/// [`run_run`]: a [`xianyun_executor::logfilter::Tui`] runs its
/// interactive event loop on a dedicated OS thread (it blocks on
/// terminal I/O, so it can't run on an async executor's worker thread)
/// for the lifetime of [`execute_reloaders_concurrently`], and is
/// told to quit once that finishes -- mirroring Go's `tui.Quit()`
/// deferred at the end of `runRun`. Each reloader's own output is also
/// now routed through it, via `tui_dispatch` below.
///
/// `tui_dispatch`, when `Some` (the log-filter TUI is active), is
/// installed as the `tracing` default for the duration of this
/// reloader's `spawn_blocking` work via
/// `tracing::dispatcher::with_default` -- scoped to that one worker
/// thread, not process-wide. This is the Rust equivalent of Go's
/// `setupTUI` swapping `slog.Default()` for the whole process:
/// `tracing::dispatcher::set_global_default` can only succeed once per
/// process (see `hooks::resolve_logging_config`'s doc comment), so
/// routing `run`'s reloader output through the TUI for `run`'s lifetime
/// only -- while every *other* command keeps logging on the dispatcher
/// `commands::dispatch` already installed -- has to be a scoped
/// override per task rather than a second global swap. Go's reloaders
/// log via the ambient `slog` package-level functions with no awareness
/// of the swap; this port's reloader code (`mirror`/`skaffold` managers,
/// `run_single_reloader`) is unchanged for the same reason -- it logs
/// via plain `tracing::info!`/`tracing::debug!` macros, which resolve to
/// whatever dispatcher is active on the calling thread.
#[must_use]
pub fn build_reloader_run_fn(
    executor: Arc<dyn CommandRunner + Send + Sync>,
    caproni_cfg: &CaproniConfig,
    static_vars: &HashMap<String, String>,
    kubeconfig: &str,
    tui_dispatch: Option<tracing::Dispatch>,
) -> ReloaderRunFn {
    let caproni_cfg = caproni_cfg.clone();
    let static_vars = static_vars.clone();
    let kubeconfig = kubeconfig.to_string();

    Arc::new(move |name, reloader_config| {
        let executor = executor.clone();
        let caproni_cfg = caproni_cfg.clone();
        let static_vars = static_vars.clone();
        let kubeconfig = kubeconfig.clone();
        let tui_dispatch = tui_dispatch.clone();

        Box::pin(async move {
            tokio::task::spawn_blocking(move || {
                let run = || {
                    run_single_reloader(
                        &executor,
                        &kubeconfig,
                        &name,
                        &reloader_config,
                        &caproni_cfg,
                        &static_vars,
                    )
                };

                match &tui_dispatch {
                    Some(dispatch) => tracing::dispatcher::with_default(dispatch, run),
                    None => run(),
                }
                // Port of `runReloader`'s outer `fmt.Errorf("running
                // reloader %s: %w", reloaderName, err)` wrap, applied to
                // *any* failure from `runWithHooks` (hook failure or
                // reloader failure alike) -- `run_single_reloader` only
                // adds its own, more specific "running hooks
                // before/after reloader"/no wrap (reloader failure)
                // context; this is the second, outer layer Go's
                // `runReloader` always adds on top.
                .with_context(|| format!("running reloader {name}"))
            })
            .await
            .context("reloader task panicked")?
        }) as ReloaderFuture
    })
}

/// Runs a single reloader to completion, wrapped with its
/// `edit_mode_start`/`edit_mode_stop` hooks. Port of Go's
/// `runWithHooks`.
fn run_single_reloader(
    executor: &Arc<dyn CommandRunner + Send + Sync>,
    kubeconfig: &str,
    name: &str,
    reloader_config: &ReloaderConfig,
    caproni_cfg: &CaproniConfig,
    static_vars: &HashMap<String, String>,
) -> anyhow::Result<()> {
    tracing::info!(
        reloader = name,
        kind = %reloader_config.reloader_type,
        "starting reloader in edit mode"
    );

    let repo_config = caproni_cfg
        .repositories
        .as_ref()
        .and_then(|repos| repos.get(&reloader_config.repository))
        .ok_or_else(|| RunError::RepositoryNotDefined(reloader_config.repository.clone()))?;

    let resolver = StaticResolver(static_vars.clone());

    let start_hooks =
        lifecyclehooks::resolve_reloader_hooks(reloader_config, HookType::EditModeStart);
    lifecyclehooks::run_hooks(
        executor.as_ref(),
        caproni_cfg,
        &resolver,
        repo_config,
        &reloader_config.repository,
        &start_hooks,
        None,
    )
    .with_context(|| format!("running hooks before reloader {name}"))?;

    let kubectl_backend = kubectl::Client::with_config(executor.clone(), kubeconfig);

    let run_result =
        ReloaderManager::new(&reloader_config.reloader_type, Box::new(kubectl_backend))
            .and_then(|mgr| mgr.run(executor.as_ref(), reloader_config, caproni_cfg, ""))
            .with_context(|| format!("running reloader {name}"));

    let stop_hooks =
        lifecyclehooks::resolve_reloader_hooks(reloader_config, HookType::EditModeStop);
    let hook_result = lifecyclehooks::run_hooks(
        executor.as_ref(),
        caproni_cfg,
        &resolver,
        repo_config,
        &reloader_config.repository,
        &stop_hooks,
        None,
    )
    .with_context(|| format!("running hooks after reloader {name}"));

    match (run_result, hook_result) {
        (Ok(()), Ok(())) => Ok(()),
        (Err(err), Ok(())) => Err(err),
        (Ok(()), Err(hook_err)) => Err(hook_err),
        (Err(err), Err(hook_err)) => Err(err.context(format!("also: {hook_err}"))),
    }
}

/// A [`xianyun_core::variables::resolver::Resolver`] backed by a plain,
/// owned `HashMap`, so lifecycle hooks can be run inside a
/// `spawn_blocking` task without moving `Container::resolver`'s
/// `Box<dyn Resolver>` across a thread boundary (that trait object has
/// no `Send`/`Sync` bound, unlike this crate's own concrete type here).
struct StaticResolver(HashMap<String, String>);

impl xianyun_core::variables::resolver::Resolver for StaticResolver {
    fn expand(&self, name: &str) -> String {
        self.0
            .get(name)
            .cloned()
            .unwrap_or_else(|| std::env::var(name).unwrap_or_default())
    }

    fn all(&self) -> HashMap<String, String> {
        self.0.clone()
    }

    fn lookup(&self, name: &str) -> Option<String> {
        self.0
            .get(name)
            .cloned()
            .or_else(|| std::env::var(name).ok())
    }
}

/// Dispatches to the concrete `mirror`/`skaffold` reloader manager for a
/// given `reloader_type`. Both managers share the same `run`/
/// `check_preconditions` method shapes but have no common trait to
/// dispatch through generically in this workspace (see
/// `PORT_NOTES.md`), so this small enum stands in for one.
enum ReloaderManager {
    Mirror(xianyun_deploy::reloader::mirror::manager::Manager),
    Skaffold(xianyun_deploy::reloader::skaffold::manager::Manager),
}

impl ReloaderManager {
    fn new(
        reloader_type: &str,
        kubectl_backend: Box<dyn kubectl::Backend + Send + Sync>,
    ) -> anyhow::Result<Self> {
        match reloader_type {
            "mirror" => Ok(Self::Mirror(
                xianyun_deploy::reloader::mirror::manager::Manager::new(kubectl_backend),
            )),
            "skaffold" => Ok(Self::Skaffold(
                xianyun_deploy::reloader::skaffold::manager::Manager::new(kubectl_backend),
            )),
            other => Err(RunError::UnsupportedReloaderType(other.to_string()).into()),
        }
    }

    fn run(
        &self,
        executor: &dyn CommandRunner,
        reloader_config: &ReloaderConfig,
        caproni_config: &CaproniConfig,
        kubeconfig_path: &str,
    ) -> anyhow::Result<()> {
        match self {
            Self::Mirror(mgr) => mgr
                .run(executor, reloader_config, caproni_config, kubeconfig_path)
                .map_err(anyhow::Error::from),
            Self::Skaffold(mgr) => mgr
                .run(executor, reloader_config, caproni_config, kubeconfig_path)
                .map_err(anyhow::Error::from),
        }
    }
}

/// Port of `run.runEditModeStartHooks`/the `edit_mode_stop` half of
/// `runAndWait`'s cleanup goroutine: runs the given hook type for every
/// repository that has an active edit-mode reloader.
fn run_edit_mode_hooks(
    executor: &dyn CommandRunner,
    caproni_cfg: &CaproniConfig,
    static_vars: &HashMap<String, String>,
    edit_mode_reloaders: &HashMap<String, ReloaderConfig>,
    hook_type: HookType,
) -> anyhow::Result<()> {
    let Some(repositories) = &caproni_cfg.repositories else {
        return Ok(());
    };

    let resolver = StaticResolver(static_vars.clone());

    for (name, repo) in repositories.iter() {
        if !has_repository_active_reloader(name, edit_mode_reloaders) {
            continue;
        }

        let hooks = lifecyclehooks::filter_hooks(&repo.lifecycle_hooks, hook_type);

        lifecyclehooks::run_hooks(executor, caproni_cfg, &resolver, repo, name, &hooks, None)
            .with_context(|| format!("running hooks in repository '{name}'"))?;
    }

    Ok(())
}

/// Port of `run.cloneRepos`: ensures git repositories are cloned for
/// `edit_mode` reloaders.
fn clone_repos(
    caproni_cfg: &CaproniConfig,
    edit_mode_reloaders: &HashMap<String, ReloaderConfig>,
    git_backend: &dyn GitBackend,
) -> anyhow::Result<()> {
    let mut repos_to_clone: std::collections::HashSet<&str> = std::collections::HashSet::new();

    for reloader_config in edit_mode_reloaders.values() {
        repos_to_clone.insert(&reloader_config.repository);
    }

    for repo_name in repos_to_clone {
        let Some(repo_config) = caproni_cfg
            .repositories
            .as_ref()
            .and_then(|repos| repos.get(&repo_name.to_string()))
        else {
            return Err(RunError::RepositoryNotDefined(repo_name.to_string()).into());
        };

        let repo_dir = caproni_cfg.resolve_path(&repo_config.directory);

        tracing::info!(name = repo_name, directory = %repo_dir, "ensuring repository is cloned");

        git_backend
            .ensure_cloned(&repo_config.repository, &repo_dir, "on-demand")
            .with_context(|| format!("cloning repository {repo_name}"))?;
    }

    Ok(())
}

/// Port of `findEditModeReloaders`: returns every reloader with
/// `edit_mode: true`.
#[must_use]
pub fn find_edit_mode_reloaders(caproni_cfg: &CaproniConfig) -> HashMap<String, ReloaderConfig> {
    let mut reloaders = HashMap::new();

    if let Some(configured) = &caproni_cfg.reloaders {
        for (name, reloader_config) in configured.iter() {
            if reloader_config.edit_mode {
                reloaders.insert(name.clone(), reloader_config.clone());
            }
        }
    }

    reloaders
}

/// Port of `reloaderNames`: the sorted names of a reloader map, for
/// deterministic logging.
#[must_use]
pub fn reloader_names(reloaders: &HashMap<String, ReloaderConfig>) -> Vec<String> {
    let mut names: Vec<String> = reloaders.keys().cloned().collect();
    names.sort();
    names
}

/// Port of `hasRepositoryActiveReloader`: whether any edit-mode reloader
/// uses the given repository.
#[must_use]
pub fn has_repository_active_reloader(
    repo_name: &str,
    edit_mode_reloaders: &HashMap<String, ReloaderConfig>,
) -> bool {
    edit_mode_reloaders
        .values()
        .any(|r| r.edit_mode && r.repository == repo_name)
}

/// Port of `collectMirrorNamespaces`: the deduplicated set of target
/// namespaces from all `mirror` reloaders, in first-seen order.
#[must_use]
pub fn collect_mirror_namespaces(reloaders: &HashMap<String, ReloaderConfig>) -> Vec<String> {
    let mut namespaces = Vec::new();

    for reloader in reloaders.values() {
        if reloader.reloader_type != "mirror" {
            continue;
        }

        let Some(mirror) = &reloader.mirror else {
            continue;
        };
        let Some(target) = &mirror.target else {
            continue;
        };

        if !namespaces.contains(&target.namespace) {
            namespaces.push(target.namespace.clone());
        }
    }

    namespaces
}

/// Port of `collectReloaderErrors`: filters out cancellation errors
/// (expected during graceful shutdown) and returns any real failures.
///
/// Deviation from Go: `errors.Join` supports unwrapping into multiple
/// underlying errors via `Unwrap() []error`, which `anyhow::Error` has no
/// equivalent for. When more than one real (non-cancellation) error
/// remains, they're joined into a single `anyhow::Error` whose message
/// contains every original error's `Display` output, but only the single
/// error case preserves the original error's type for `downcast_ref`
/// (matching every test that exercises this: they only ever inject one
/// real failure alongside cancellations).
///
/// # Errors
///
/// Returns an error if any input error is not a cancellation
/// ([`xianyun_executor::exec::ExecError::Cancelled`]).
pub fn collect_reloader_errors(errors: Vec<anyhow::Error>) -> anyhow::Result<()> {
    let mut real_errors: Vec<anyhow::Error> = errors
        .into_iter()
        .filter(|err| !is_cancelled(err))
        .collect();

    match real_errors.len() {
        0 => Ok(()),
        1 => Err(real_errors.remove(0)),
        _ => {
            let combined = real_errors
                .iter()
                .map(std::string::ToString::to_string)
                .collect::<Vec<_>>()
                .join("\n");
            Err(anyhow::anyhow!(combined))
        }
    }
}

/// Whether `err`'s chain contains
/// [`xianyun_executor::exec::ExecError::Cancelled`], the closest
/// equivalent this workspace has to Go's `context.Canceled`.
fn is_cancelled(err: &anyhow::Error) -> bool {
    err.chain().any(|cause| {
        matches!(
            cause.downcast_ref::<xianyun_executor::exec::ExecError>(),
            Some(xianyun_executor::exec::ExecError::Cancelled)
        )
    })
}