xbp-deploy 10.57.0

use std::sync::Arc;

use async_trait::async_trait;
use futures::future::join_all;
use serde::{Deserialize, Serialize};
use xbp_k8s::{K8sAdapter, K8sApplyOptions, KubeTarget};
use xbp_oci::OciPromoter;

use crate::bootstrap::bootstrap_workload_documents;
use crate::cloudflare::{CloudflareDeployAdapter, CloudflareServiceDeployRequest};
use crate::context::{DeployContext, DeployMode};
use crate::error::{DeployError, Result};
use crate::history::{record_from_plan_with_version, DeployHistoryStore};
use crate::lock::{lock_from_plan, write_lock};
use crate::oci_build::{OciBuildAdapter, OciBuildRequest};
use crate::providers::{
    is_cloudflare_provider, is_kubernetes_provider, is_railway_provider, provider_to_destination,
};
use crate::types::{DeployPlan, ServicePlan};
use std::path::PathBuf;
use std::process::Command;
use std::sync::OnceLock;

/// Prefer local Docker for CF container pre-build when the engine is available.
fn docker_available_for_local_cf_build() -> bool {
    static AVAILABLE: OnceLock<bool> = OnceLock::new();
    *AVAILABLE.get_or_init(|| {
        Command::new("docker")
            .args(["info", "--format", "{{.ServerVersion}}"])
            .output()
            .map(|o| o.status.success())
            .unwrap_or(false)
    })
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeployResult {
    pub ok: bool,
    pub lines: Vec<String>,
    pub failed_services: Vec<String>,
    pub error: Option<String>,
    /// Secrets extracted when writing redacted deploy history (for D1 upload).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub history_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub history_path: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub history_secrets: Vec<crate::history::DeployHistorySecret>,
}

#[async_trait]
pub trait DeployRunner: Send + Sync {
    async fn run(&self, ctx: &DeployContext, plan: &DeployPlan) -> Result<DeployResult>;
}

pub struct DefaultDeployRunner {
    pub k8s: Arc<dyn K8sAdapter>,
    pub promoter: Option<Arc<dyn OciPromoter>>,
    /// Optional confirmation callback: returns true to proceed.
    pub confirm: Option<Arc<dyn Fn(String) -> bool + Send + Sync>>,
    /// Optional Cloudflare adapter. Required only when the plan includes CF providers.
    pub cloudflare: Option<Arc<dyn CloudflareDeployAdapter>>,
    /// Optional docker build/push before k8s apply when services declare oci.dockerfile.
    pub oci_build: Option<Arc<dyn OciBuildAdapter>>,
}

#[async_trait]
impl DeployRunner for DefaultDeployRunner {
    async fn run(&self, ctx: &DeployContext, plan: &DeployPlan) -> Result<DeployResult> {
        let mut lines = Vec::new();
        let mut failed = Vec::new();

        let k8s_services: Vec<&ServicePlan> = plan
            .services
            .iter()
            .filter(|s| is_kubernetes_provider(&s.provider))
            .collect();
        let cf_services: Vec<&ServicePlan> = plan
            .services
            .iter()
            .filter(|s| is_cloudflare_provider(&s.provider))
            .collect();

        lines.push(format!(
            "run target={} env={} dry_run={} services={} k8s={} cloudflare={}",
            plan.target.label(),
            plan.env,
            ctx.flags.dry_run,
            plan.order.len(),
            k8s_services.len(),
            cf_services.len()
        ));
        if !k8s_services.is_empty() {
            let ns_parts: Vec<String> = k8s_services
                .iter()
                .map(|s| {
                    format!(
                        "{}={}",
                        s.name,
                        s.deploy
                            .namespace
                            .as_deref()
                            .or(plan.k8s_plan.default_namespace.as_deref())
                            .unwrap_or("?")
                    )
                })
                .collect();
            lines.push(format!(
                "context={:?} namespaces=[{}] plan_default={:?}",
                plan.k8s_plan.context,
                ns_parts.join(", "),
                plan.k8s_plan.default_namespace
            ));
        }

        // Promote mode: OCI only (unchanged).
        if matches!(ctx.mode, DeployMode::Promote) {
            return self.promote(ctx, plan).await;
        }

        // Mutable service snapshots so build can pin digests before k8s apply/bootstrap.
        let mut k8s_work: Vec<ServicePlan> =
            k8s_services.iter().map(|s| (*s).clone()).collect();

        // May flip to true if registry push fails and the operator opts into local-only.
        let mut local_image = ctx.flags.local_image;

        // ------------------------------------------------------------------
        // Phase 1 — build + push OCI images FIRST (before cluster confirm/apply).
        // Requires services[].oci.dockerfile + image. Use --skip-build to skip.
        // ------------------------------------------------------------------
        match self
            .run_oci_build_phase(
                ctx,
                plan,
                &mut k8s_work,
                &mut lines,
                &mut failed,
                &mut local_image,
            )
            .await
        {
            Ok(()) => {}
            Err(e) => return Err(e),
        }
        // Drop k8s work items whose image build/push failed (keep CF destinations alive).
        // Multi-dest services share `name` (e.g. athena-auth); fail keys use `name@destination`.
        if !failed.is_empty() {
            let failed_set: std::collections::HashSet<&str> =
                failed.iter().map(String::as_str).collect();
            k8s_work.retain(|s| !failed_set.contains(service_step_key(s).as_str()));
            let has_remaining_cf = plan
                .services
                .iter()
                .any(|s| is_cloudflare_provider(&s.provider));
            if k8s_work.is_empty() && !has_remaining_cf {
                lines.push(
                    "deploy aborted: image build/push failed — fix Docker/registry, then re-run (cluster was not modified for failed services)"
                        .into(),
                );
                lines.push(
                    "hint: re-run with --local-image to build into local Docker without registry push (docker-desktop / kind / minikube / CF --local-build)"
                        .into(),
                );
                return Ok(finish_result(plan, ctx, lines, failed));
            }
            if has_remaining_cf {
                lines.push(
                    "image build/push failed for some service(s) — continuing Cloudflare destinations (k8s steps for failed images skipped)"
                        .into(),
                );
                lines.push(
                    "hint: for k8s use --local-image or `docker login ghcr.io`; for CF-only re-run with --to cloudflare"
                        .into(),
                );
            }
        }

        // Context confirmation only when Kubernetes services are actually applied.
        if !k8s_work.is_empty() && ctx.require_confirmation() {
            let msg = format!(
                "Apply deploy `{}` to env `{}` (context={:?})? Images already built/pushed.",
                plan.target.label(),
                plan.env,
                plan.k8s_plan.context
            );
            let ok = self
                .confirm
                .as_ref()
                .map(|f| f(msg))
                .unwrap_or(false);
            if !ok {
                return Err(DeployError::Cancelled);
            }
            lines.push("context confirmed".into());
        }

        // --- Kubernetes path (only when k8s services present; identical apply semantics) ---
        if !k8s_work.is_empty() {
            let target = KubeTarget {
                context: plan.k8s_plan.context.clone(),
                namespace: plan.k8s_plan.default_namespace.clone(),
                kubeconfig: ctx
                    .config
                    .kubernetes
                    .as_ref()
                    .and_then(|k| k.kubeconfig.clone()),
            };

            // Fail fast with structured diagnosis when the API is down (e.g. Docker Desktop
            // Kubernetes stopped / kindest node bootstrap failed) instead of dumping apply spam.
            if !ctx.flags.dry_run {
                if let Err(e) = self.k8s.ensure_ready(&target).await {
                    failed.push("kubernetes".into());
                    lines.push(format!("ERROR: cluster not ready: {e}"));
                    if let Some(report) =
                        xbp_k8s::kubectl_failure_debug_report(&e.to_string(), target.context.as_deref())
                    {
                        for line in report.lines().filter(|l| !l.is_empty()) {
                            lines.push(line.to_string());
                        }
                    }
                    return Ok(finish_result(plan, ctx, lines, failed));
                }
                lines.push(format!(
                    "cluster ready (context={:?})",
                    target.context.as_deref().unwrap_or("(default)")
                ));
            }

            // Ensure per-service namespaces exist before apply/rollout.
            if !ctx.flags.skip_apply && !ctx.flags.dry_run {
                let opts = K8sApplyOptions {
                    dry_run: false,
                    server_side: false,
                    prune: false,
                };
                match ensure_plan_namespaces(self.k8s.as_ref(), &target, &k8s_work, &opts).await {
                    Ok(ns_lines) => lines.extend(ns_lines),
                    Err(e) => {
                        failed.push("kubernetes".into());
                        lines.push(format!("ERROR: ensure namespaces: {e}"));
                        return Ok(finish_result(plan, ctx, lines, failed));
                    }
                }
            }

            if !ctx.flags.skip_apply && failed.is_empty() {
                let opts = K8sApplyOptions {
                    dry_run: ctx.flags.dry_run,
                    server_side: false,
                    prune: false,
                };
                let k8s_plan = plan.to_k8s_plan();
                let has_manifests = k8s_plan.services.iter().any(|s| {
                    !s.manifest_paths.is_empty()
                        || s.crds_path.is_some()
                        || s.install_path.is_some()
                });

                // Default: bootstrap when no manifests (first-time deploy).
                if !has_manifests && ctx.flags.bootstrap_workload {
                    for svc in &k8s_work {
                        if !svc.deploy.manifest_paths.is_empty() {
                            continue;
                        }
                        if svc.deploy.workload.is_none() {
                            failed.push(svc.name.clone());
                            lines.push(format!(
                                "[{}] bootstrap failed: services[].deploy.envs.*.kubernetes.workload is unset (e.g. deployment/{})",
                                svc.name, svc.name
                            ));
                            continue;
                        }
                        // Compose may set container_port; default 8080 for generic services.
                        let container_port = svc.container_port.unwrap_or(8080);
                        // Local-only builds are not on a registry — Never avoids ImagePullBackOff.
                        let pull_policy = if local_image {
                            "Never"
                        } else {
                            "IfNotPresent"
                        };
                        let Some(docs) = bootstrap_workload_documents(
                            svc,
                            container_port,
                            pull_policy,
                        ) else {
                            failed.push(svc.name.clone());
                            lines.push(format!(
                                "[{}] bootstrap failed: need image_ref + workload (image_ref={:?}, workload={:?}, ns={:?})",
                                svc.name,
                                svc.image_ref,
                                svc.deploy.workload,
                                svc.deploy.namespace
                            ));
                            continue;
                        };
                        let yaml = docs.join("---\n");
                        let t = target.with_service_namespace(svc.deploy.namespace.as_deref());
                        lines.push(format!(
                            "[{}] bootstrap workload ns={} image={} port={} env_keys={} config_files={}",
                            svc.name,
                            t.namespace.as_deref().unwrap_or("default"),
                            svc.image_ref.as_deref().unwrap_or("?"),
                            container_port,
                            svc.runtime_env.len(),
                            svc.config_mounts.len()
                        ));
                        match write_temp_and_apply(self.k8s.as_ref(), &t, &opts, &svc.name, &yaml)
                            .await
                        {
                            Ok(msg) => lines.push(format!("[{}] {msg}", svc.name)),
                            Err(e) => {
                                failed.push(svc.name.clone());
                                lines.push(format!("[{}] bootstrap apply failed: {e}", svc.name));
                            }
                        }
                    }
                } else if !has_manifests {
                    lines.push(
                        "ERROR: no kubernetes manifest paths — bootstrap is disabled (--no-bootstrap-workload); re-enable default bootstrap or set kubernetes.manifests"
                            .into(),
                    );
                    for svc in &k8s_work {
                        if svc.deploy.workload.is_some() && svc.deploy.manifest_paths.is_empty() {
                            failed.push(svc.name.clone());
                        }
                    }
                }

                if has_manifests && failed.is_empty() {
                    match self.k8s.apply(&target, &k8s_plan, &opts).await {
                        Ok(apply_lines) => {
                            if apply_lines.is_empty() {
                                lines.push("kubernetes apply: (no output)".into());
                            }
                            lines.extend(apply_lines);
                        }
                        Err(e) => {
                            failed.push("kubernetes".into());
                            lines.push(format!("ERROR: {e}"));
                        }
                    }
                }
            } else if ctx.flags.skip_apply {
                lines.push("skip-apply (kubernetes)".into());
            }

            if !ctx.flags.skip_rollout && !ctx.flags.dry_run && failed.is_empty() {
                for svc in &k8s_work {
                    if let Some(workload) = &svc.deploy.workload {
                        let t = target.with_service_namespace(svc.deploy.namespace.as_deref());
                        let ns_label = t
                            .namespace
                            .as_deref()
                            .unwrap_or("(cluster-default)");
                        match self.k8s.rollout(&t, workload, "180s").await {
                            Ok(out) => lines.push(format!(
                                "[{}] ns={ns_label} {}",
                                svc.name,
                                out.lines().next().unwrap_or("ok")
                            )),
                            Err(e) => {
                                failed.push(svc.name.clone());
                                let mut msg = format!(
                                    "[{}] rollout failed (ns={ns_label}, workload={workload}): {e}",
                                    svc.name
                                );
                                if svc.deploy.manifest_paths.is_empty()
                                    && !ctx.flags.bootstrap_workload
                                {
                                    msg.push_str(
                                        " — no manifests; use default bootstrap or set kubernetes.manifests",
                                    );
                                }
                                lines.push(msg);
                            }
                        }
                    }
                }
            }

            // Local routing: port-forward + hosts DNS (after successful rollout or dry-run plan).
            // With --local-image (or registry-fail fallback), auto port-forward even without expose config.
            if failed.is_empty() && !ctx.flags.skip_expose {
                for svc in &k8s_work {
                    let auto_local = local_image;
                    if crate::expose::effective_expose(svc, auto_local).is_some() {
                        let result = crate::expose::expose_service_with_auto(
                            svc,
                            ctx.flags.context.as_deref(),
                            ctx.flags.dry_run || ctx.flags.skip_rollout,
                            auto_local,
                        );
                        lines.extend(result.lines);
                    }
                }
            }
        } else {
            lines.push("kubernetes: no k8s services in plan (skipped)".into());
        }

        // --- Explicit unsupported destinations (fail clear, no kubectl) ---
        for svc in plan.services.iter().filter(|s| is_railway_provider(&s.provider)) {
            if !ctx.flags.skip_apply {
                failed.push(svc.name.clone());
                lines.push(format!(
                    "[{}] destination `railway` is reserved but not wired yet (use --to cloudflare or --to kubernetes)",
                    svc.name
                ));
            }
        }
        for svc in plan.services.iter().filter(|s| {
            let d = s
                .destination
                .clone()
                .unwrap_or_else(|| provider_to_destination(&s.provider));
            !is_kubernetes_provider(&s.provider)
                && !is_cloudflare_provider(&s.provider)
                && !is_railway_provider(&s.provider)
                && d != "local"
                && d != "oci"
        }) {
            if !ctx.flags.skip_apply {
                lines.push(format!(
                    "[{}] destination `{}` has no runner yet (provider={})",
                    svc.name,
                    svc.destination
                        .as_deref()
                        .unwrap_or("?"),
                    svc.provider
                ));
            }
        }

        // --- Cloudflare path (additive; never calls kubectl) ---
        //
        // Build ownership:
        // - OpenNext / Wrangler always builds Worker assets during CF deploy (unless skip_apply).
        // - `local_image` → pre-build CF Container Dockerfile with local Docker (`--local-build`).
        // - `skip_apply` + `local_image` → local container build only (no Wrangler).
        // - GHCR/docker push is intentionally not used for CF-only plans.
        if !cf_services.is_empty() {
            let local_build_only = ctx.flags.skip_apply && local_image;
            let skip_cf_deploy = ctx.flags.skip_apply && !local_image;

            if skip_cf_deploy {
                lines.push(
                    "skip-apply (cloudflare): no Wrangler deploy (use full CF build strategy without skip-apply to ship)"
                        .into(),
                );
            } else {
                let Some(cf) = &self.cloudflare else {
                    for svc in &cf_services {
                        failed.push(svc.name.clone());
                        lines.push(format!(
                            "[{}] cloudflare provider `{}` requires a Cloudflare adapter (use `xbp deploy` from the CLI, or `xbp cloudflare deploy --app {}`)",
                            svc.name,
                            svc.provider,
                            svc.worker_app.as_deref().unwrap_or(svc.name.as_str())
                        ));
                    }
                    return Ok(finish_result(plan, ctx, lines, failed));
                };

                for svc in &cf_services {
                    let worker_app = svc
                        .worker_app
                        .clone()
                        .unwrap_or_else(|| svc.name.clone());
                    // Prefer local Docker pre-build for CF container workers unless dry-run.
                    // OpenNext plain Workers always build on-machine (WSL/native); this flag
                    // only affects container image pre-build. Operators can still use
                    // --local-image explicitly; we default prefer-local when Docker is usable.
                    let prefer_local_cf = !ctx.flags.dry_run
                        && (local_image
                            || local_build_only
                            || docker_available_for_local_cf_build());
                    // Config-only CF deploys are the default (image may not change).
                    // Operators who need a new image fail with --require-new-container-image.
                    // --skip-build also implies config-only is expected.
                    let allow_unchanged =
                        !ctx.flags.require_new_container_image || ctx.flags.skip_build;
                    let request = CloudflareServiceDeployRequest {
                        service_name: svc.name.clone(),
                        worker_app: worker_app.clone(),
                        dry_run: ctx.flags.dry_run,
                        rollout: svc.rollout.clone(),
                        skip_deploy: local_build_only,
                        allow_unchanged_container_image: allow_unchanged,
                        prune_old_images: false,
                        local_build: prefer_local_cf,
                        local_build_only,
                    };
                    if local_build_only {
                        lines.push(format!(
                            "[{}] cloudflare local container build only (no Wrangler) app={}",
                            svc.name, worker_app
                        ));
                    } else {
                        lines.push(format!(
                            "[{}] cloudflare deploy app={} dry_run={} local_build={} allow_unchanged_image={}",
                            svc.name,
                            worker_app,
                            ctx.flags.dry_run,
                            prefer_local_cf,
                            allow_unchanged
                        ));
                    }
                    match cf.deploy_service(request).await {
                        Ok(cf_lines) => {
                            for line in cf_lines {
                                lines.push(format!("[{}] {line}", svc.name));
                            }
                        }
                        Err(e) => {
                            failed.push(svc.name.clone());
                            lines.push(format!("[{}] cloudflare deploy failed: {e}", svc.name));
                        }
                    }
                }
            }
        }

        // Health probes for services that declare health URLs.
        // For pure CF, the cloudflare workflow already verifies containers/health;
        // env-level health still runs unless skip_health (same as k8s).
        if !ctx.flags.skip_health && !ctx.flags.dry_run && failed.is_empty() {
            let health_futs = plan.services.iter().flat_map(|svc| {
                svc.deploy.health.iter().map(move |url| {
                    let name = svc.name.clone();
                    let url = url.clone();
                    async move { (name, url.clone(), http_status(&url).await) }
                })
            });
            for (name, url, result) in join_all(health_futs).await {
                match result {
                    Ok(code) if (200..400).contains(&code) => {
                        lines.push(format!("[{name}] health {url} → {code}"));
                    }
                    Ok(code) => {
                        failed.push(name.clone());
                        lines.push(format!("[{name}] health {url} → {code}"));
                    }
                    Err(e) => {
                        failed.push(name.clone());
                        lines.push(format!("[{name}] health {url} failed: {e}"));
                    }
                }
            }
        }

        Ok(finish_result(plan, ctx, lines, failed))
    }
}

async fn write_temp_and_apply(
    k8s: &dyn xbp_k8s::K8sAdapter,
    target: &KubeTarget,
    opts: &K8sApplyOptions,
    service: &str,
    yaml: &str,
) -> std::result::Result<String, String> {
    let dir = std::env::temp_dir().join(format!("xbp-bootstrap-{service}"));
    let _ = std::fs::create_dir_all(&dir);
    let path = dir.join("manifest.yaml");
    std::fs::write(&path, yaml).map_err(|e| format!("write temp manifest: {e}"))?;
    // Ensure namespace object first when set.
    if let Some(ns) = target.namespace.as_deref().filter(|s| !s.is_empty()) {
        let ns_yaml = format!(
            "apiVersion: v1\nkind: Namespace\nmetadata:\n  name: {ns}\n  labels:\n    app.kubernetes.io/managed-by: xbp\n"
        );
        let ns_path = dir.join("namespace.yaml");
        let _ = std::fs::write(&ns_path, ns_yaml);
        let _ = k8s.apply_path(target, &ns_path, opts).await;
    }
    k8s.apply_path(target, &path, opts)
        .await
        .map(|out| {
            format!(
                "bootstrap apply {}: {}",
                path.display(),
                out.lines().next().unwrap_or("ok")
            )
        })
        .map_err(|e| e.to_string())
}

/// Create missing namespaces declared by the plan (idempotent apply of Namespace objects).
async fn ensure_plan_namespaces(
    k8s: &dyn xbp_k8s::K8sAdapter,
    target: &KubeTarget,
    services: &[ServicePlan],
    opts: &K8sApplyOptions,
) -> std::result::Result<Vec<String>, String> {
    let mut namespaces: Vec<String> = services
        .iter()
        .filter_map(|s| {
            s.deploy
                .namespace
                .as_deref()
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .map(str::to_string)
        })
        .collect();
    if let Some(ns) = target
        .namespace
        .as_deref()
        .map(str::trim)
        .filter(|s| !s.is_empty())
    {
        namespaces.push(ns.to_string());
    }
    namespaces.sort();
    namespaces.dedup();
    if namespaces.is_empty() {
        return Ok(Vec::new());
    }

    let dir = std::env::temp_dir().join(format!(
        "xbp-ensure-ns-{}",
        std::process::id()
    ));
    std::fs::create_dir_all(&dir).map_err(|e| format!("create temp dir: {e}"))?;

    let mut lines = Vec::new();
    for ns in namespaces {
        let ns_yaml = format!(
            "apiVersion: v1\nkind: Namespace\nmetadata:\n  name: {ns}\n  labels:\n    app.kubernetes.io/managed-by: xbp\n"
        );
        let ns_path = dir.join(format!("{ns}.yaml"));
        std::fs::write(&ns_path, ns_yaml).map_err(|e| format!("write namespace yaml: {e}"))?;
        // Apply without forcing --namespace (Namespace is cluster-scoped).
        let ns_target = KubeTarget {
            context: target.context.clone(),
            namespace: None,
            kubeconfig: target.kubeconfig.clone(),
        };
        match k8s.apply_path(&ns_target, &ns_path, opts).await {
            Ok(out) => {
                let detail = out.lines().next().unwrap_or("ok");
                lines.push(format!("ensure namespace `{ns}`: {detail}"));
            }
            Err(e) => {
                let msg = e.to_string();
                // Idempotent: already exists is success.
                if msg.to_ascii_lowercase().contains("alreadyexists")
                    || msg.to_ascii_lowercase().contains("already exists")
                {
                    lines.push(format!("ensure namespace `{ns}`: already exists"));
                } else {
                    return Err(format!("namespace `{ns}`: {msg}"));
                }
            }
        }
    }
    Ok(lines)
}

fn svc_root(project_root: &std::path::Path, svc: &ServicePlan) -> Option<PathBuf> {
    let rel = svc.root_directory.as_deref()?.trim();
    if rel.is_empty() {
        return None;
    }
    let p = PathBuf::from(rel);
    Some(if p.is_absolute() {
        p
    } else {
        project_root.join(p)
    })
}

fn resolve_build_context(project_root: &std::path::Path, svc: &ServicePlan) -> PathBuf {
    let service_root = svc_root(project_root, svc);
    if let Some(ctx) = svc.build_context.as_deref().map(str::trim).filter(|s| !s.is_empty()) {
        let p = PathBuf::from(ctx);
        if p.is_absolute() {
            return p;
        }
        if let Some(root) = &service_root {
            return root.join(p);
        }
        return project_root.join(p);
    }
    service_root.unwrap_or_else(|| project_root.to_path_buf())
}

/// Compact OPERATION-friendly error: service names plus first root-cause line.
fn format_deploy_failure(failed: &[String], lines: &[String]) -> String {
    let services = failed.join(", ");
    let reason = first_failure_reason(failed, lines);
    match reason {
        Some(r) => format!("failed: {services} — {r}"),
        None => format!("failed: {services}"),
    }
}

fn first_failure_reason(failed: &[String], lines: &[String]) -> Option<String> {
    // Prefer lines tagged with a failed service name.
    for name in failed {
        let prefix = format!("[{name}]");
        for line in lines.iter().rev() {
            let trimmed = line.trim();
            if !trimmed.starts_with(&prefix) {
                continue;
            }
            if let Some(rest) = extract_failure_detail(trimmed) {
                return Some(rest);
            }
        }
    }
    // Generic ERROR: lines from apply path.
    for line in lines.iter().rev() {
        let trimmed = line.trim();
        if let Some(rest) = trimmed.strip_prefix("ERROR:") {
            let rest = rest.trim();
            if !rest.is_empty() {
                return Some(rest.to_string());
            }
        }
    }
    None
}

fn extract_failure_detail(line: &str) -> Option<String> {
    // "[svc] rollout failed: …" / "[svc] apply …" / "[svc] health … failed: …"
    for marker in ["rollout failed: ", " failed: ", "cloudflare deploy failed: "] {
        if let Some(idx) = line.find(marker) {
            let rest = line[idx + marker.len()..].trim();
            if !rest.is_empty() {
                return Some(rest.to_string());
            }
        }
    }
    None
}

fn finish_result(
    plan: &DeployPlan,
    ctx: &DeployContext,
    lines: Vec<String>,
    failed: Vec<String>,
) -> DeployResult {
    let ok = failed.is_empty();
    let mut result = DeployResult {
        ok,
        lines: lines.clone(),
        failed_services: failed.clone(),
        error: if ok {
            None
        } else {
            Some(format_deploy_failure(&failed, &lines))
        },
        history_id: None,
        history_path: None,
        history_secrets: Vec::new(),
    };

    if matches!(ctx.mode, DeployMode::Run) && !ctx.flags.dry_run {
        let store = DeployHistoryStore::new(&ctx.config.history_dir);
        let xbp_version = std::env::var("CARGO_PKG_VERSION")
            .ok()
            .filter(|s| !s.is_empty())
            .or_else(|| Some(env!("CARGO_PKG_VERSION").to_string()));
        let sanitized = record_from_plan_with_version(
            plan,
            ok,
            if ok {
                "deploy succeeded".into()
            } else {
                result.error.clone().unwrap_or_default()
            },
            result.error.clone(),
            plan.k8s_plan.context.clone(),
            xbp_version,
        );
        if let Ok(path) = store.write_record(&sanitized.record) {
            result.history_id = Some(sanitized.record.id.clone());
            result.history_path = Some(path.display().to_string());
            if !sanitized.secrets.is_empty() {
                result.lines.push(format!(
                    "history: redacted {} secret(s) from runtime_env → placeholders (id={})",
                    sanitized.secrets.len(),
                    sanitized.record.id
                ));
            }
        }
        result.history_secrets = sanitized.secrets;
        if ok {
            let lock = lock_from_plan(plan);
            let _ = write_lock(&ctx.config.lock_file, &lock);
        }
    }

    result
}

impl DefaultDeployRunner {
    /// Phase 1 of run: build and push every service image that declares `oci.dockerfile`.
    ///
    /// Runs before Kubernetes context confirmation / apply so registry has digests
    /// available for bootstrap and pull. Updates `k8s_work` image_ref/digest in place.
    ///
    /// On registry auth/denied failures in a TTY (without `--yes`), offers to continue
    /// with local-only load (same as `--local-image`) and sets `local_image`.
    async fn run_oci_build_phase(
        &self,
        ctx: &DeployContext,
        plan: &DeployPlan,
        k8s_work: &mut [ServicePlan],
        lines: &mut Vec<String>,
        failed: &mut Vec<String>,
        local_image: &mut bool,
    ) -> Result<()> {
        if ctx.flags.skip_build {
            lines.push("phase build: skipped (--skip-build)".into());
            return Ok(());
        }

        // Only Kubernetes (and pure OCI destination) steps need GHCR/docker push here.
        // Cloudflare Workers / Containers own their own image path (OpenNext / wrangler
        // containers / --local-build). Building ghcr.io for CF-only plans is wrong and
        // blocks deploys that never touch a cluster (e.g. next-heroui-example).
        //
        // Defense in depth: never build for cloudflare providers even if they leak into
        // k8s_work or carry dockerfile + image_ref from services[].oci.
        let mut build_targets: Vec<usize> = Vec::new(); // indices into k8s_work
        for (i, svc) in k8s_work.iter().enumerate() {
            if is_cloudflare_provider(&svc.provider) {
                lines.push(format!(
                    "[{}] phase build: skip GHCR for cloudflare provider (use CF local_build / OpenNext)",
                    svc.name
                ));
                continue;
            }
            if !is_kubernetes_provider(&svc.provider) {
                continue;
            }
            if svc.dockerfile.is_some() && svc.image_ref.is_some() {
                build_targets.push(i);
            }
        }
        // Explicit `oci` destination only (not cloudflare / railway / local / k8s).
        let mut extra: Vec<ServicePlan> = plan
            .services
            .iter()
            .filter(|s| {
                crate::providers::is_oci_provider(&s.provider)
                    && !is_cloudflare_provider(&s.provider)
                    && !is_kubernetes_provider(&s.provider)
                    && s.dockerfile.is_some()
                    && s.image_ref.is_some()
            })
            .cloned()
            .collect();

        if build_targets.is_empty() && extra.is_empty() {
            let cf_with_dockerfile = plan.services.iter().any(|s| {
                is_cloudflare_provider(&s.provider)
                    && s.dockerfile.is_some()
                    && s.image_ref.is_some()
            });
            let cf_any = plan
                .services
                .iter()
                .any(|s| is_cloudflare_provider(&s.provider));
            if cf_with_dockerfile || cf_any {
                if *local_image {
                    lines.push(
                        "phase build: Cloudflare path — container pre-build runs in CF step (`local_build`); no GHCR push"
                            .into(),
                    );
                } else {
                    lines.push(
                        "phase build: skipped GHCR (Cloudflare owns image build — OpenNext/Wrangler/containers)"
                            .into(),
                    );
                }
            }
            return Ok(());
        }

        let n = build_targets.len() + extra.len();
        if *local_image {
            lines.push(format!(
                "phase build: building {n} image(s) into local Docker (no registry push)"
            ));
        } else {
            lines.push(format!(
                "phase build/push: building and pushing {n} image(s) before cluster apply"
            ));
        }

        let Some(builder) = &self.oci_build else {
            for i in &build_targets {
                lines.push(format!(
                    "[{}] oci.dockerfile set but no build adapter wired (skip build)",
                    k8s_work[*i].name
                ));
            }
            for svc in &extra {
                lines.push(format!(
                    "[{}] oci.dockerfile set but no build adapter wired (skip build)",
                    svc.name
                ));
            }
            return Ok(());
        };

        for i in &build_targets {
            let step_key = service_step_key(&k8s_work[*i]);
            let svc_name = k8s_work[*i].name.clone();
            let image_ref = k8s_work[*i].image_ref.clone().expect("checked");
            let dockerfile = k8s_work[*i].dockerfile.clone().expect("checked");
            let platforms = k8s_work[*i].platforms.clone();
            let context = resolve_build_context(&ctx.config.project_root, &k8s_work[*i]);
            let service_root = svc_root(&ctx.config.project_root, &k8s_work[*i]);

            match self
                .build_one_image(
                    builder.as_ref(),
                    ctx,
                    &svc_name,
                    &image_ref,
                    dockerfile,
                    context,
                    platforms,
                    service_root,
                    local_image,
                    lines,
                )
                .await
            {
                Ok(Some(result)) => {
                    let svc = &mut k8s_work[*i];
                    if let Some(pinned) = result.image_ref {
                        // Keep tag refs for local load (digest pins need registry pulls).
                        if *local_image && pinned.contains('@') {
                            svc.image_ref = Some(image_ref.clone());
                        } else {
                            svc.image_ref = Some(pinned);
                        }
                    }
                    if !*local_image {
                        if let Some(d) = result.digest {
                            svc.digest = Some(d);
                        }
                    }
                }
                Ok(None) => {}
                Err(_e) => {
                    failed.push(step_key);
                    // Fail fast: do not build remaining images after a failure.
                    // Error detail already logged by build_one_image.
                    return Ok(());
                }
            }
        }

        for svc in &mut extra {
            let step_key = service_step_key(svc);
            let svc_name = svc.name.clone();
            let image_ref = svc.image_ref.clone().expect("checked");
            let dockerfile = svc.dockerfile.clone().expect("checked");
            let platforms = svc.platforms.clone();
            let context = resolve_build_context(&ctx.config.project_root, svc);
            let service_root = svc_root(&ctx.config.project_root, svc);

            match self
                .build_one_image(
                    builder.as_ref(),
                    ctx,
                    &svc_name,
                    &image_ref,
                    dockerfile,
                    context,
                    platforms,
                    service_root,
                    local_image,
                    lines,
                )
                .await
            {
                Ok(_) => {}
                Err(_e) => {
                    failed.push(step_key);
                    // Error detail already logged by build_one_image.
                    return Ok(());
                }
            }
        }

        if failed.is_empty() && n > 0 {
            if *local_image {
                lines.push("phase build: complete (local Docker)".into());
            } else {
                lines.push("phase build/push: complete".into());
            }
        }
        Ok(())
    }

    /// Build/push one image. On registry auth failure, may prompt and retry local-only.
    #[allow(clippy::too_many_arguments)]
    async fn build_one_image(
        &self,
        builder: &dyn OciBuildAdapter,
        ctx: &DeployContext,
        svc_name: &str,
        image_ref: &str,
        dockerfile: String,
        context: PathBuf,
        platforms: Vec<String>,
        service_root: Option<PathBuf>,
        local_image: &mut bool,
        lines: &mut Vec<String>,
    ) -> std::result::Result<Option<crate::oci_build::OciBuildResult>, String> {
        let verb = if *local_image {
            "build/load"
        } else {
            "build/push"
        };
        lines.push(format!("[{svc_name}] {verb} start → {image_ref}"));

        let req = OciBuildRequest {
            service: svc_name.to_string(),
            image_ref: image_ref.to_string(),
            dockerfile: Some(dockerfile.clone()),
            context: context.clone(),
            platforms: platforms.clone(),
            dry_run: ctx.flags.dry_run,
            project_root: ctx.config.project_root.clone(),
            service_root: service_root.clone(),
            local_image: *local_image,
        };

        match builder.build_and_push(req).await {
            Ok(result) => {
                let used_local = result.local_only || *local_image;
                for line in &result.lines {
                    lines.push(format!("[{svc_name}] {line}"));
                }
                let done_verb = if used_local {
                    "build/load"
                } else {
                    "build/push"
                };
                lines.push(format!("[{svc_name}] {done_verb} ok"));
                Ok(Some(result))
            }
            Err(e) => {
                lines.push(format!("[{svc_name}] {verb} failed: {e}"));
                if !*local_image
                    && is_registry_auth_failure(&e)
                    && self.should_offer_local_image_fallback(ctx)
                {
                    let prompt = format!(
                        "[{svc_name}] Registry push failed. Continue with local Docker only (no push, like --local-image)?"
                    );
                    let accepted = self
                        .confirm
                        .as_ref()
                        .map(|f| f(prompt))
                        .unwrap_or(false);
                    if accepted {
                        *local_image = true;
                        lines.push(format!(
                            "[{svc_name}] opted into --local-image after registry failure; rebuilding into local Docker"
                        ));
                        let retry = OciBuildRequest {
                            service: svc_name.to_string(),
                            image_ref: image_ref.to_string(),
                            dockerfile: Some(dockerfile),
                            context,
                            platforms,
                            dry_run: ctx.flags.dry_run,
                            project_root: ctx.config.project_root.clone(),
                            service_root,
                            local_image: true,
                        };
                        match builder.build_and_push(retry).await {
                            Ok(result) => {
                                for line in &result.lines {
                                    lines.push(format!("[{svc_name}] {line}"));
                                }
                                lines.push(format!(
                                    "[{svc_name}] build/load ok (local fallback)"
                                ));
                                return Ok(Some(result));
                            }
                            Err(e2) => {
                                return Err(format!(
                                    "[{svc_name}] build/load failed after local fallback: {e2}"
                                ));
                            }
                        }
                    }
                    lines.push(format!(
                        "[{svc_name}] declined local-image fallback — fix `docker login` / GH_TOKEN / `xbp config oci`, or re-run with --local-image"
                    ));
                } else if !*local_image && is_registry_auth_failure(&e) {
                    lines.push(format!(
                        "[{svc_name}] hint: re-run with --local-image (or without --yes in a TTY to be prompted) to skip registry push"
                    ));
                }
                Err(format!("[{svc_name}] {verb} failed: {e}"))
            }
        }
    }

    fn should_offer_local_image_fallback(&self, ctx: &DeployContext) -> bool {
        if ctx.flags.yes || ctx.flags.json || ctx.flags.dry_run {
            return false;
        }
        self.confirm.is_some()
    }
}

/// Registry push denied / unauthorized (operator often wanted local load instead).
fn is_registry_auth_failure(err: &str) -> bool {
    let el = err.to_ascii_lowercase();
    el.contains("denied")
        || el.contains("unauthorized")
        || el.contains("authentication required")
        || el.contains("no basic auth credentials")
        || el.contains("requested access to the resource is denied")
        || el.contains("error from registry")
        || el.contains("registry auth")
}

/// Stable step id for multi-destination plans (`name@destination` when set).
fn service_step_key(svc: &ServicePlan) -> String {
    match svc.destination.as_deref().map(str::trim).filter(|s| !s.is_empty()) {
        Some(d) => format!("{}@{d}", svc.name),
        None => svc.name.clone(),
    }
}

impl DefaultDeployRunner {
    async fn promote(&self, ctx: &DeployContext, plan: &DeployPlan) -> Result<DeployResult> {
        let tag = ctx
            .flags
            .promote_tag
            .as_deref()
            .ok_or_else(|| DeployError::Validation("promote requires a target tag".into()))?;
        let Some(promoter) = &self.promoter else {
            return Err(DeployError::Oci("no OCI promoter configured".into()));
        };
        let mut lines = Vec::new();
        let mut failed = Vec::new();
        for svc in &plan.services {
            // CF destinations never use GHCR promote.
            if is_cloudflare_provider(&svc.provider) {
                lines.push(format!(
                    "[{}] skip promote (cloudflare owns image path)",
                    svc.name
                ));
                continue;
            }
            let Some(image) = &svc.image else {
                continue;
            };
            let target = image.with_tag(tag);
            if ctx.flags.dry_run {
                lines.push(format!(
                    "[{}] dry-run promote {} -> {}",
                    svc.name,
                    image.reference(),
                    target.reference()
                ));
                continue;
            }
            match promoter.promote(image, &target, true).await {
                Ok(()) => lines.push(format!(
                    "[{}] promoted {} -> {}",
                    svc.name,
                    image.reference(),
                    target.reference()
                )),
                Err(e) => {
                    failed.push(svc.name.clone());
                    lines.push(format!("[{}] promote failed: {e}", svc.name));
                }
            }
        }
        Ok(DeployResult {
            ok: failed.is_empty(),
            lines,
            failed_services: failed.clone(),
            error: if failed.is_empty() {
                None
            } else {
                Some(format!("promote failed: {}", failed.join(", ")))
            },
            history_id: None,
            history_path: None,
            history_secrets: Vec::new(),
        })
    }
}

async fn http_status(url: &str) -> std::result::Result<u16, String> {
    use tokio::process::Command;
    let output = Command::new("curl")
        .args(["-sS", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "15", url])
        .output()
        .await
        .map_err(|e| e.to_string())?;
    if !output.status.success() {
        return Err(String::from_utf8_lossy(&output.stderr).trim().to_string());
    }
    String::from_utf8_lossy(&output.stdout)
        .trim()
        .parse()
        .map_err(|_| "invalid status".into())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::context::{DeployFlags, DeployMode};
    use crate::types::*;
    use async_trait::async_trait;
    use std::collections::HashMap;
    use std::path::PathBuf;
    use std::sync::Mutex;
    use xbp_k8s::{K8sApplyOptions, K8sPlan, K8sServicePlan, KubeTarget};

    struct RecordingK8s {
        apply_calls: Mutex<usize>,
        rollout_calls: Mutex<usize>,
    }

    #[async_trait]
    impl K8sAdapter for RecordingK8s {
        async fn render(
            &self,
            _plan: &K8sPlan,
        ) -> xbp_k8s::Result<xbp_k8s::RenderedManifests> {
            Ok(xbp_k8s::RenderedManifests {
                sources: vec![],
                documents: vec![],
            })
        }

        async fn apply(
            &self,
            _target: &KubeTarget,
            plan: &K8sPlan,
            _opts: &K8sApplyOptions,
        ) -> xbp_k8s::Result<Vec<String>> {
            *self.apply_calls.lock().unwrap() += 1;
            Ok(vec![format!("applied {} services", plan.services.len())])
        }

        async fn diff(
            &self,
            _target: &KubeTarget,
            _plan: &K8sPlan,
        ) -> xbp_k8s::Result<String> {
            Ok(String::new())
        }

        async fn rollout(
            &self,
            _target: &KubeTarget,
            _workload: &str,
            _timeout: &str,
        ) -> xbp_k8s::Result<String> {
            *self.rollout_calls.lock().unwrap() += 1;
            Ok("ok".into())
        }

        async fn verify(
            &self,
            _target: &KubeTarget,
            _plan: &K8sPlan,
        ) -> xbp_k8s::Result<Vec<String>> {
            Ok(vec![])
        }

        async fn apply_path(
            &self,
            _target: &KubeTarget,
            _path: &std::path::Path,
            _opts: &K8sApplyOptions,
        ) -> xbp_k8s::Result<String> {
            Ok(String::new())
        }
    }

    struct RecordingCf {
        calls: Mutex<Vec<String>>,
        /// Each call: (worker_app, local_build)
        local_build_flags: Mutex<Vec<bool>>,
        allow_unchanged_flags: Mutex<Vec<bool>>,
    }

    impl RecordingCf {
        fn new() -> Self {
            Self {
                calls: Mutex::new(Vec::new()),
                local_build_flags: Mutex::new(Vec::new()),
                allow_unchanged_flags: Mutex::new(Vec::new()),
            }
        }
    }

    #[async_trait]
    impl CloudflareDeployAdapter for RecordingCf {
        async fn deploy_service(
            &self,
            request: CloudflareServiceDeployRequest,
        ) -> std::result::Result<Vec<String>, String> {
            self.calls.lock().unwrap().push(request.worker_app.clone());
            self.local_build_flags
                .lock()
                .unwrap()
                .push(request.local_build);
            self.allow_unchanged_flags
                .lock()
                .unwrap()
                .push(request.allow_unchanged_container_image);
            Ok(vec!["cf-ok".into()])
        }
    }

    fn base_ctx(plan_services_only: bool) -> (DeployContext, PathBuf) {
        let root = std::env::temp_dir().join(format!(
            "xbp-deploy-test-{}",
            std::process::id()
        ));
        let _ = std::fs::create_dir_all(&root);
        let ctx = DeployContext {
            env: "production".into(),
            target: DeployTarget::Service("svc".into()),
            config: ProjectConfig {
                project_name: "t".into(),
                version: "1.0.0".into(),
                project_root: root.clone(),
                services: vec![],
                groups: HashMap::new(),
                default_env: Some("production".into()),
                kubernetes: Some(KubernetesDefaults {
                    default_context: Some("kind".into()),
                    default_namespace: Some("default".into()),
                    require_context_confirmation_for_apply: false,
                    kubeconfig: None,
                }),
                history_dir: root.join("history"),
                lock_file: root.join("lock.json"),
                git_sha: None,
            },
            flags: DeployFlags {
                dry_run: true,
                yes: true,
                skip_health: true,
                skip_rollout: true,
                bootstrap_workload: false, // tests supply real apply paths / no bootstrap
                skip_doctor: true,
                ..DeployFlags::default()
            },
            mode: DeployMode::Run,
        };
        let _ = plan_services_only;
        (ctx, root)
    }

    #[tokio::test]
    async fn k8s_only_plan_still_calls_kubectl_apply() {
        let k8s = Arc::new(RecordingK8s {
            apply_calls: Mutex::new(0),
            rollout_calls: Mutex::new(0),
        });
        let (ctx, root) = base_ctx(true);
        let manifest = root.join("api.yaml");
        std::fs::write(&manifest, "apiVersion: v1\nkind: ConfigMap\n").unwrap();
        let plan = DeployPlan {
            target: DeployTarget::Service("api".into()),
            env: "production".into(),
            project: "t".into(),
            project_version: "1".into(),
            git_sha: None,
            services: vec![ServicePlan {
                name: "api".into(),
                provider: "kubernetes".into(),
                destination: Some("kubernetes".into()),
                root_directory: None,
                version: "1".into(),
                image: None,
                image_ref: None,
                digest: None,
                dockerfile: None,
                build_context: None,
                platforms: vec![],
                worker_app: None,
                rollout: None,
                runtime_env: Default::default(),
                container_port: None,
                config_mounts: vec![],
                expose: None,
                deploy: ServiceDeployPlan {
                    namespace: Some("default".into()),
                    workload: Some("deploy/api".into()),
                    service: None,
                    health: vec![],
                    manifest_paths: vec![manifest.clone()],
                    crds_path: None,
                    install_path: None,
                    selector: None,
                    actions: vec![],
                },
            }],
            order: vec!["api".into()],
            oci_plan: OciPlan::default(),
            k8s_plan: K8sPlanView {
                context: Some("kind".into()),
                default_namespace: Some("default".into()),
                services: vec![K8sServicePlan {
                    service: "api".into(),
                    provider: "kubernetes".into(),
                    namespace: Some("default".into()),
                    workload: Some("deploy/api".into()),
                    manifest_paths: vec![manifest],
                    crds_path: None,
                    install_path: None,
                    selector: None,
                    health: vec![],
                }],
            },
            hash: "h".into(),
        };
        let runner = DefaultDeployRunner {
            k8s: k8s.clone(),
            promoter: None,
            confirm: None,
            cloudflare: None,
            oci_build: None,
        };
        let result = runner.run(&ctx, &plan).await.unwrap();
        assert!(result.ok, "{:?}", result.lines);
        assert_eq!(*k8s.apply_calls.lock().unwrap(), 1);
    }

    #[tokio::test]
    async fn cloudflare_only_plan_does_not_call_kubectl() {
        let k8s = Arc::new(RecordingK8s {
            apply_calls: Mutex::new(0),
            rollout_calls: Mutex::new(0),
        });
        let cf = Arc::new(RecordingCf::new());
        let (ctx, _root) = base_ctx(true);
        let plan = DeployPlan {
            target: DeployTarget::Service("athena-auth".into()),
            env: "production".into(),
            project: "t".into(),
            project_version: "1".into(),
            git_sha: None,
            services: vec![ServicePlan {
                name: "athena-auth".into(),
                provider: "cloudflare-containers".into(),
                destination: Some("cloudflare".into()),
                root_directory: None,
                version: "1.14.2".into(),
                image: None,
                image_ref: None,
                digest: None,
                dockerfile: None,
                build_context: None,
                platforms: vec![],
                worker_app: Some("athena-auth".into()),
                rollout: Some("immediate".into()),
                runtime_env: Default::default(),
                container_port: None,
                config_mounts: vec![],
                expose: None,
                deploy: ServiceDeployPlan {
                    namespace: None,
                    workload: None,
                    service: None,
                    health: vec![],
                    manifest_paths: vec![],
                    crds_path: None,
                    install_path: None,
                    selector: None,
                    actions: vec![],
                },
            }],
            order: vec!["athena-auth".into()],
            oci_plan: OciPlan::default(),
            k8s_plan: K8sPlanView {
                context: None,
                default_namespace: None,
                services: vec![],
            },
            hash: "h".into(),
        };
        let runner = DefaultDeployRunner {
            k8s: k8s.clone(),
            promoter: None,
            confirm: None,
            cloudflare: Some(cf.clone()),
            oci_build: None,
        };
        let result = runner.run(&ctx, &plan).await.unwrap();
        assert!(result.ok, "{:?}", result.lines);
        assert_eq!(*k8s.apply_calls.lock().unwrap(), 0);
        assert_eq!(*k8s.rollout_calls.lock().unwrap(), 0);
        assert_eq!(cf.calls.lock().unwrap().as_slice(), ["athena-auth"]);
    }

    #[tokio::test]
    async fn mixed_plan_runs_k8s_then_cf_without_mixing_adapters() {
        let k8s = Arc::new(RecordingK8s {
            apply_calls: Mutex::new(0),
            rollout_calls: Mutex::new(0),
        });
        let cf = Arc::new(RecordingCf::new());
        let (ctx, root) = base_ctx(true);
        let manifest = root.join("api.yaml");
        std::fs::write(&manifest, "apiVersion: v1\nkind: ConfigMap\n").unwrap();
        let plan = DeployPlan {
            target: DeployTarget::Group("edge".into()),
            env: "production".into(),
            project: "t".into(),
            project_version: "1".into(),
            git_sha: None,
            services: vec![
                ServicePlan {
                    name: "api".into(),
                    provider: "kubernetes".into(),
                    destination: Some("kubernetes".into()),
                    root_directory: None,
                    version: "1".into(),
                    image: None,
                    image_ref: None,
                    digest: None,
                    dockerfile: None,
                    build_context: None,
                    platforms: vec![],
                    worker_app: None,
                    rollout: None,
                    runtime_env: Default::default(),
                    container_port: None,
                    config_mounts: vec![],
                    expose: None,
                    deploy: ServiceDeployPlan {
                        namespace: Some("default".into()),
                        workload: None,
                        service: None,
                        health: vec![],
                        manifest_paths: vec![manifest.clone()],
                        crds_path: None,
                        install_path: None,
                        selector: None,
                        actions: vec![],
                    },
                },
                ServicePlan {
                    name: "auth".into(),
                    provider: "cloudflare-containers".into(),
                    destination: Some("cloudflare".into()),
                    root_directory: None,
                    version: "1".into(),
                    image: None,
                    image_ref: None,
                    digest: None,
                    dockerfile: None,
                    build_context: None,
                    platforms: vec![],
                    worker_app: Some("auth".into()),
                    rollout: None,
                    runtime_env: Default::default(),
                    container_port: None,
                    config_mounts: vec![],
                    expose: None,
                    deploy: ServiceDeployPlan {
                        namespace: None,
                        workload: None,
                        service: None,
                        health: vec![],
                        manifest_paths: vec![],
                        crds_path: None,
                        install_path: None,
                        selector: None,
                        actions: vec![],
                    },
                },
            ],
            order: vec!["api@kubernetes".into(), "auth@cloudflare".into()],
            oci_plan: OciPlan::default(),
            k8s_plan: K8sPlanView {
                context: Some("kind".into()),
                default_namespace: Some("default".into()),
                services: vec![K8sServicePlan {
                    service: "api".into(),
                    provider: "kubernetes".into(),
                    namespace: Some("default".into()),
                    workload: None,
                    manifest_paths: vec![manifest],
                    crds_path: None,
                    install_path: None,
                    selector: None,
                    health: vec![],
                }],
            },
            hash: "h".into(),
        };
        let runner = DefaultDeployRunner {
            k8s: k8s.clone(),
            promoter: None,
            confirm: None,
            cloudflare: Some(cf.clone()),
            oci_build: None,
        };
        let result = runner.run(&ctx, &plan).await.unwrap();
        assert!(result.ok, "{:?}", result.lines);
        assert_eq!(*k8s.apply_calls.lock().unwrap(), 1);
        assert_eq!(cf.calls.lock().unwrap().as_slice(), ["auth"]);
    }

    #[test]
    fn format_deploy_failure_includes_rollout_reason() {
        let failed = vec!["athena".into()];
        let lines = vec![
            "run target=athena env=production dry_run=false services=1".into(),
            "[athena] rollout failed: kubectl failed: cannot reach cluster API at 127.0.0.1:6443 (connection refused)".into(),
        ];
        let msg = format_deploy_failure(&failed, &lines);
        assert!(msg.starts_with("failed: athena"), "got: {msg}");
        assert!(msg.contains("cannot reach cluster API"), "got: {msg}");
        assert!(msg.contains("connection refused"), "got: {msg}");
    }

    #[test]
    fn format_deploy_failure_includes_notfound_with_ns() {
        let failed = vec!["athena".into()];
        let lines = vec![
            "[athena] rollout failed (ns=athena, workload=deploy/athena): kubectl failed: Error from server (NotFound): deployments.apps \"athena\" not found".into(),
        ];
        let msg = format_deploy_failure(&failed, &lines);
        assert!(msg.contains("NotFound") || msg.contains("not found"), "got: {msg}");
        assert!(msg.contains("ns=athena") || msg.contains("athena"), "got: {msg}");
    }

    #[test]
    fn registry_auth_failure_detects_denied_and_unauthorized() {
        assert!(is_registry_auth_failure(
            "docker push ghcr.io/x/y:latest failed: error from registry: denied"
        ));
        assert!(is_registry_auth_failure("unauthorized: authentication required"));
        assert!(is_registry_auth_failure("no basic auth credentials"));
        assert!(!is_registry_auth_failure("Dockerfile not found: ./Dockerfile"));
        assert!(!is_registry_auth_failure("docker CLI not found"));
    }

    struct FailPushThenLocalOci {
        calls: Mutex<Vec<bool>>,
    }

    #[async_trait]
    impl OciBuildAdapter for FailPushThenLocalOci {
        async fn build_and_push(
            &self,
            request: OciBuildRequest,
        ) -> std::result::Result<crate::oci_build::OciBuildResult, String> {
            self.calls.lock().unwrap().push(request.local_image);
            if request.local_image {
                Ok(crate::oci_build::OciBuildResult {
                    lines: vec!["local ok".into()],
                    digest: None,
                    image_ref: Some(request.image_ref),
                    local_only: true,
                })
            } else {
                Err(
                    "docker push ghcr.io/xylex-group/athena-auth:latest failed: error from registry: denied\ndenied"
                        .into(),
                )
            }
        }
    }

    #[tokio::test]
    async fn registry_denied_offers_local_fallback_when_confirm_yes() {
        let k8s = Arc::new(RecordingK8s {
            apply_calls: Mutex::new(0),
            rollout_calls: Mutex::new(0),
        });
        let oci = Arc::new(FailPushThenLocalOci {
            calls: Mutex::new(Vec::new()),
        });
        let (mut ctx, root) = base_ctx(true);
        ctx.flags.dry_run = false;
        ctx.flags.yes = false; // allow fallback prompt
        ctx.flags.skip_apply = true;
        ctx.flags.skip_rollout = true;
        ctx.flags.skip_health = true;
        ctx.flags.bootstrap_workload = false;

        let dockerfile = root.join("Dockerfile");
        std::fs::write(&dockerfile, "FROM scratch\n").unwrap();
        let manifest = root.join("api.yaml");
        std::fs::write(&manifest, "apiVersion: v1\nkind: ConfigMap\n").unwrap();

        let plan = DeployPlan {
            target: DeployTarget::Service("api".into()),
            env: "production".into(),
            project: "t".into(),
            project_version: "1".into(),
            git_sha: None,
            services: vec![ServicePlan {
                name: "api".into(),
                provider: "kubernetes".into(),
                destination: Some("kubernetes".into()),
                root_directory: None,
                version: "1".into(),
                image: None,
                image_ref: Some("ghcr.io/xylex-group/api:latest".into()),
                digest: None,
                dockerfile: Some("Dockerfile".into()),
                build_context: Some(".".into()),
                platforms: vec![],
                worker_app: None,
                rollout: None,
                runtime_env: Default::default(),
                container_port: None,
                config_mounts: vec![],
                expose: None,
                deploy: ServiceDeployPlan {
                    namespace: Some("default".into()),
                    workload: Some("deployment/api".into()),
                    service: None,
                    health: vec![],
                    manifest_paths: vec![manifest],
                    crds_path: None,
                    install_path: None,
                    selector: None,
                    actions: vec![],
                },
            }],
            order: vec!["api".into()],
            oci_plan: OciPlan::default(),
            k8s_plan: K8sPlanView {
                context: Some("kind".into()),
                default_namespace: Some("default".into()),
                services: vec![],
            },
            hash: "h".into(),
        };

        let runner = DefaultDeployRunner {
            k8s,
            promoter: None,
            confirm: Some(Arc::new(|msg: String| {
                assert!(msg.contains("local Docker only"), "prompt: {msg}");
                true
            })),
            cloudflare: None,
            oci_build: Some(oci.clone()),
        };
        let result = runner.run(&ctx, &plan).await.unwrap();
        assert!(result.ok, "lines={:?}", result.lines);
        assert!(
            result
                .lines
                .iter()
                .any(|l| l.contains("opted into --local-image")),
            "lines={:?}",
            result.lines
        );
        assert_eq!(oci.calls.lock().unwrap().as_slice(), &[false, true]);
    }

    struct CountingOci {
        calls: Mutex<usize>,
    }

    #[async_trait]
    impl OciBuildAdapter for CountingOci {
        async fn build_and_push(
            &self,
            _request: OciBuildRequest,
        ) -> std::result::Result<crate::oci_build::OciBuildResult, String> {
            *self.calls.lock().unwrap() += 1;
            Ok(crate::oci_build::OciBuildResult {
                lines: vec!["oci-ok".into()],
                digest: None,
                image_ref: Some("ghcr.io/x/app:latest".into()),
                local_only: false,
            })
        }
    }

    #[tokio::test]
    async fn cloudflare_only_with_dockerfile_skips_oci_build() {
        let k8s = Arc::new(RecordingK8s {
            apply_calls: Mutex::new(0),
            rollout_calls: Mutex::new(0),
        });
        let oci = Arc::new(CountingOci {
            calls: Mutex::new(0),
        });
        let cf = Arc::new(RecordingCf::new());
        let (mut ctx, _root) = base_ctx(true);
        ctx.flags.dry_run = false;
        ctx.flags.yes = true;
        ctx.flags.skip_health = true;

        let plan = DeployPlan {
            target: DeployTarget::Service("next-heroui-example".into()),
            env: "production".into(),
            project: "t".into(),
            project_version: "1".into(),
            git_sha: None,
            services: vec![ServicePlan {
                name: "next-heroui-example".into(),
                provider: "cloudflare-worker".into(),
                destination: Some("cloudflare".into()),
                root_directory: None,
                version: "4.0.0".into(),
                image: None,
                image_ref: Some("ghcr.io/xylex-group/next-heroui-example:latest".into()),
                digest: None,
                dockerfile: Some("Dockerfile".into()),
                build_context: Some(".".into()),
                platforms: vec![],
                worker_app: Some("next-heroui-example".into()),
                rollout: None,
                runtime_env: Default::default(),
                container_port: None,
                config_mounts: vec![],
                expose: None,
                deploy: ServiceDeployPlan {
                    namespace: Some("default".into()),
                    workload: None,
                    service: None,
                    health: vec![],
                    manifest_paths: vec![],
                    crds_path: None,
                    install_path: None,
                    selector: None,
                    actions: vec![],
                },
            }],
            order: vec!["next-heroui-example@cloudflare".into()],
            oci_plan: OciPlan::default(),
            k8s_plan: K8sPlanView {
                context: None,
                default_namespace: None,
                services: vec![],
            },
            hash: "h".into(),
        };

        let runner = DefaultDeployRunner {
            k8s: k8s.clone(),
            promoter: None,
            confirm: None,
            cloudflare: Some(cf.clone()),
            oci_build: Some(oci.clone()),
        };
        let result = runner.run(&ctx, &plan).await.unwrap();
        assert!(result.ok, "lines={:?}", result.lines);
        assert_eq!(
            *oci.calls.lock().unwrap(),
            0,
            "CF-only must not call GHCR/docker OCI adapter; lines={:?}",
            result.lines
        );
        assert!(
            result
                .lines
                .iter()
                .any(|l| l.contains("Cloudflare owns image build")),
            "lines={:?}",
            result.lines
        );
        assert_eq!(cf.calls.lock().unwrap().as_slice(), ["next-heroui-example"]);
        // Default policy: allow unchanged container image (config-only deploys).
        assert_eq!(
            cf.allow_unchanged_flags.lock().unwrap().as_slice(),
            &[true],
            "CF deploy should allow_unchanged by default"
        );
    }

    #[tokio::test]
    async fn require_new_container_image_disables_allow_unchanged() {
        let k8s = Arc::new(RecordingK8s {
            apply_calls: Mutex::new(0),
            rollout_calls: Mutex::new(0),
        });
        let cf = Arc::new(RecordingCf::new());
        let (mut ctx, _root) = base_ctx(true);
        ctx.flags.dry_run = false;
        ctx.flags.yes = true;
        ctx.flags.skip_health = true;
        ctx.flags.require_new_container_image = true;

        let plan = DeployPlan {
            target: DeployTarget::Service("athena-auth".into()),
            env: "production".into(),
            project: "t".into(),
            project_version: "1".into(),
            git_sha: None,
            services: vec![ServicePlan {
                name: "athena-auth".into(),
                provider: "cloudflare-containers".into(),
                destination: Some("cloudflare".into()),
                root_directory: None,
                version: "1".into(),
                image: None,
                image_ref: None,
                digest: None,
                dockerfile: None,
                build_context: None,
                platforms: vec![],
                worker_app: Some("athena-auth".into()),
                rollout: Some("immediate".into()),
                runtime_env: Default::default(),
                container_port: None,
                config_mounts: vec![],
                expose: None,
                deploy: ServiceDeployPlan {
                    namespace: None,
                    workload: None,
                    service: None,
                    health: vec![],
                    manifest_paths: vec![],
                    crds_path: None,
                    install_path: None,
                    selector: None,
                    actions: vec![],
                },
            }],
            order: vec!["athena-auth@cloudflare".into()],
            oci_plan: OciPlan::default(),
            k8s_plan: K8sPlanView {
                context: None,
                default_namespace: None,
                services: vec![],
            },
            hash: "h".into(),
        };
        let runner = DefaultDeployRunner {
            k8s: k8s.clone(),
            promoter: None,
            confirm: None,
            cloudflare: Some(cf.clone()),
            oci_build: None,
        };
        let result = runner.run(&ctx, &plan).await.unwrap();
        assert!(result.ok, "{:?}", result.lines);
        assert_eq!(
            cf.allow_unchanged_flags.lock().unwrap().as_slice(),
            &[false],
            "require_new_container_image must set allow_unchanged=false; lines={:?}",
            result.lines
        );
        assert_eq!(*k8s.apply_calls.lock().unwrap(), 0);
    }

    #[tokio::test]
    async fn registry_denied_aborts_when_confirm_no() {
        let k8s = Arc::new(RecordingK8s {
            apply_calls: Mutex::new(0),
            rollout_calls: Mutex::new(0),
        });
        let oci = Arc::new(FailPushThenLocalOci {
            calls: Mutex::new(Vec::new()),
        });
        let (mut ctx, root) = base_ctx(true);
        ctx.flags.dry_run = false;
        ctx.flags.yes = false;
        ctx.flags.skip_apply = true;

        std::fs::write(root.join("Dockerfile"), "FROM scratch\n").unwrap();
        let manifest = root.join("api.yaml");
        std::fs::write(&manifest, "apiVersion: v1\nkind: ConfigMap\n").unwrap();

        let plan = DeployPlan {
            target: DeployTarget::Service("api".into()),
            env: "production".into(),
            project: "t".into(),
            project_version: "1".into(),
            git_sha: None,
            services: vec![ServicePlan {
                name: "api".into(),
                provider: "kubernetes".into(),
                destination: Some("kubernetes".into()),
                root_directory: None,
                version: "1".into(),
                image: None,
                image_ref: Some("ghcr.io/xylex-group/api:latest".into()),
                digest: None,
                dockerfile: Some("Dockerfile".into()),
                build_context: Some(".".into()),
                platforms: vec![],
                worker_app: None,
                rollout: None,
                runtime_env: Default::default(),
                container_port: None,
                config_mounts: vec![],
                expose: None,
                deploy: ServiceDeployPlan {
                    namespace: Some("default".into()),
                    workload: Some("deployment/api".into()),
                    service: None,
                    health: vec![],
                    manifest_paths: vec![manifest],
                    crds_path: None,
                    install_path: None,
                    selector: None,
                    actions: vec![],
                },
            }],
            order: vec!["api".into()],
            oci_plan: OciPlan::default(),
            k8s_plan: K8sPlanView {
                context: Some("kind".into()),
                default_namespace: Some("default".into()),
                services: vec![],
            },
            hash: "h".into(),
        };

        let runner = DefaultDeployRunner {
            k8s,
            promoter: None,
            confirm: Some(Arc::new(|_msg: String| false)),
            cloudflare: None,
            oci_build: Some(oci.clone()),
        };
        let result = runner.run(&ctx, &plan).await.unwrap();
        assert!(!result.ok);
        assert!(
            result.failed_services.iter().any(|s| s == "api" || s == "api@kubernetes"),
            "failed={:?}",
            result.failed_services
        );
        assert_eq!(oci.calls.lock().unwrap().as_slice(), &[false]);
        assert!(
            result
                .lines
                .iter()
                .any(|l| l.contains("declined local-image fallback")),
            "lines={:?}",
            result.lines
        );
    }

    #[tokio::test]
    async fn dual_dest_oci_fail_still_runs_cloudflare() {
        let k8s = Arc::new(RecordingK8s {
            apply_calls: Mutex::new(0),
            rollout_calls: Mutex::new(0),
        });
        let oci = Arc::new(FailPushThenLocalOci {
            calls: Mutex::new(Vec::new()),
        });
        let cf = Arc::new(RecordingCf::new());
        let (mut ctx, root) = base_ctx(true);
        ctx.flags.dry_run = false;
        ctx.flags.yes = true; // no local fallback prompt
        ctx.flags.skip_health = true;
        ctx.flags.bootstrap_workload = false;

        std::fs::write(root.join("Dockerfile"), "FROM scratch\n").unwrap();
        let manifest = root.join("api.yaml");
        std::fs::write(&manifest, "apiVersion: v1\nkind: ConfigMap\n").unwrap();

        let plan = DeployPlan {
            target: DeployTarget::Service("athena-auth".into()),
            env: "production".into(),
            project: "t".into(),
            project_version: "1".into(),
            git_sha: None,
            services: vec![
                ServicePlan {
                    name: "athena-auth".into(),
                    provider: "kubernetes".into(),
                    destination: Some("kubernetes".into()),
                    root_directory: None,
                    version: "1".into(),
                    image: None,
                    image_ref: Some("ghcr.io/xylex-group/athena-auth:latest".into()),
                    digest: None,
                    dockerfile: Some("Dockerfile".into()),
                    build_context: Some(".".into()),
                    platforms: vec![],
                    worker_app: None,
                    rollout: None,
                    runtime_env: Default::default(),
                    container_port: None,
                    config_mounts: vec![],
                    expose: None,
                    deploy: ServiceDeployPlan {
                        namespace: Some("athena".into()),
                        workload: Some("deployment/athena-auth".into()),
                        service: None,
                        health: vec![],
                        manifest_paths: vec![manifest],
                        crds_path: None,
                        install_path: None,
                        selector: None,
                        actions: vec![],
                    },
                },
                ServicePlan {
                    name: "athena-auth".into(),
                    provider: "cloudflare-containers".into(),
                    destination: Some("cloudflare".into()),
                    root_directory: None,
                    version: "1".into(),
                    image: None,
                    image_ref: Some("ghcr.io/xylex-group/athena-auth:latest".into()),
                    digest: None,
                    dockerfile: Some("Dockerfile".into()),
                    build_context: Some(".".into()),
                    platforms: vec![],
                    worker_app: Some("athena-auth".into()),
                    rollout: Some("immediate".into()),
                    runtime_env: Default::default(),
                    container_port: None,
                    config_mounts: vec![],
                    expose: None,
                    deploy: ServiceDeployPlan {
                        namespace: None,
                        workload: None,
                        service: None,
                        health: vec![],
                        manifest_paths: vec![],
                        crds_path: None,
                        install_path: None,
                        selector: None,
                        actions: vec![],
                    },
                },
            ],
            order: vec![
                "athena-auth@kubernetes".into(),
                "athena-auth@cloudflare".into(),
            ],
            oci_plan: OciPlan::default(),
            k8s_plan: K8sPlanView {
                context: Some("kind".into()),
                default_namespace: Some("athena".into()),
                services: vec![],
            },
            hash: "h".into(),
        };

        let runner = DefaultDeployRunner {
            k8s: k8s.clone(),
            promoter: None,
            confirm: None,
            cloudflare: Some(cf.clone()),
            oci_build: Some(oci.clone()),
        };
        let result = runner.run(&ctx, &plan).await.unwrap();
        // Overall not ok (k8s OCI failed) but CF still ran.
        assert!(!result.ok, "lines={:?}", result.lines);
        assert!(
            result
                .failed_services
                .iter()
                .any(|s| s.contains("kubernetes") || s.contains("athena-auth")),
            "failed={:?}",
            result.failed_services
        );
        assert_eq!(oci.calls.lock().unwrap().as_slice(), &[false]);
        assert_eq!(cf.calls.lock().unwrap().as_slice(), ["athena-auth"]);
        assert_eq!(*k8s.apply_calls.lock().unwrap(), 0);
        assert!(
            result
                .lines
                .iter()
                .any(|l| l.contains("continuing Cloudflare")),
            "lines={:?}",
            result.lines
        );
    }
}