zlayer-agent 0.13.0

//! Job execution engine - run-to-completion container lifecycle
//!
//! This module provides the `JobExecutor` which handles run-to-completion
//! workloads (jobs and cron jobs). Unlike services which run indefinitely,
//! jobs run to completion and track their exit status.

use crate::error::{AgentError, Result};
use crate::init::InitOrchestrator;
use crate::overlay_manager::OverlayManager;
use crate::runtime::{ContainerId, Runtime};
// Only the non-Windows attach path matches on the attach kind; on Windows the
// overlay is wired at container-create time inside overlayd.
#[cfg(not(target_os = "windows"))]
use crate::runtime::OverlayAttachKind;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};
use uuid::Uuid;
use zlayer_spec::ServiceSpec;

/// How a job container was attached to the overlay, so it can be detached when
/// the job exits (otherwise every execution leaks a veth + overlay IP).
#[cfg_attr(target_os = "windows", allow(dead_code))]
#[derive(Default)]
enum OverlayAttachment {
    /// Not attached (no overlay manager, no PID, attach failed, or Windows —
    /// where the overlay is wired at container-create time inside overlayd).
    #[default]
    None,
    /// Linux host-process (youki): detach by the PID recorded at attach.
    Pid(u32),
    /// macOS VZ guest: detach by the container id used at attach.
    Guest(String),
    /// macOS host-shared (Seatbelt/native-VZ/libkrun): detach by the container
    /// id used at attach (overlayd allocated a distinct overlay /32 + utun alias).
    HostShared(String),
}

/// Unique identifier for a job execution
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct JobExecutionId(pub String);

impl JobExecutionId {
    /// Create a new random execution ID
    #[must_use]
    pub fn new() -> Self {
        Self(Uuid::new_v4().to_string())
    }
}

impl Default for JobExecutionId {
    fn default() -> Self {
        Self::new()
    }
}

impl std::fmt::Display for JobExecutionId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// Status of a job execution
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum JobStatus {
    /// Job is queued, waiting to start
    Pending,
    /// Init steps are running
    Initializing,
    /// Main container is running
    Running,
    /// Job completed successfully
    Completed { exit_code: i32, duration: Duration },
    /// Job failed
    Failed {
        reason: String,
        exit_code: Option<i32>,
    },
    /// Job was cancelled
    Cancelled,
}

impl std::fmt::Display for JobStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            JobStatus::Pending => write!(f, "pending"),
            JobStatus::Initializing => write!(f, "initializing"),
            JobStatus::Running => write!(f, "running"),
            JobStatus::Completed { exit_code, .. } => write!(f, "completed({exit_code})"),
            JobStatus::Failed { exit_code, .. } => {
                if let Some(code) = exit_code {
                    write!(f, "failed({code})")
                } else {
                    write!(f, "failed")
                }
            }
            JobStatus::Cancelled => write!(f, "cancelled"),
        }
    }
}

/// How the job was triggered
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum JobTrigger {
    /// Triggered via HTTP endpoint
    Endpoint { remote_addr: Option<String> },
    /// Triggered via CLI
    Cli,
    /// Triggered by cron scheduler
    Scheduler,
    /// Triggered by internal system (dependency, etc.)
    Internal { reason: String },
}

impl std::fmt::Display for JobTrigger {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            JobTrigger::Endpoint { remote_addr } => {
                if let Some(addr) = remote_addr {
                    write!(f, "endpoint({addr})")
                } else {
                    write!(f, "endpoint")
                }
            }
            JobTrigger::Cli => write!(f, "cli"),
            JobTrigger::Scheduler => write!(f, "scheduler"),
            JobTrigger::Internal { reason } => write!(f, "internal({reason})"),
        }
    }
}

/// A single job execution record
#[derive(Debug, Clone)]
pub struct JobExecution {
    pub id: JobExecutionId,
    pub job_name: String,
    pub status: JobStatus,
    pub started_at: Option<Instant>,
    pub completed_at: Option<Instant>,
    pub container_id: Option<ContainerId>,
    /// Captured stdout/stderr (limited to last N bytes)
    pub logs: Option<String>,
    /// Trigger source (endpoint, cli, scheduler, etc.)
    pub trigger: JobTrigger,
}

/// Configuration for the job executor
#[derive(Debug, Clone)]
pub struct JobExecutorConfig {
    /// Maximum concurrent job executions per job name
    pub max_concurrent: usize,
    /// How long to retain completed job records
    pub retention: Duration,
    /// Maximum log size to capture (in bytes)
    pub max_log_size: usize,
}

impl Default for JobExecutorConfig {
    fn default() -> Self {
        Self {
            max_concurrent: 10,
            retention: Duration::from_secs(3600), // 1 hour
            max_log_size: 1024 * 1024,            // 1 MB
        }
    }
}

/// Job executor handles run-to-completion workloads
pub struct JobExecutor {
    runtime: Arc<dyn Runtime + Send + Sync>,
    /// Active and recent job executions
    executions: Arc<RwLock<HashMap<JobExecutionId, JobExecution>>>,
    /// Job specs (for jobs that need to be stored)
    job_specs: Arc<RwLock<HashMap<String, ServiceSpec>>>,
    /// Overlay manager used to attach each job container to the overlay network
    /// (so the job can reach the daemon API / peer services) and detach it on
    /// exit. `None` when the daemon runs without overlay networking.
    overlay_manager: Option<Arc<RwLock<OverlayManager>>>,
    /// Sink for persisting/revoking per-container scoped tokens. `None`
    /// disables persistence (token minted without a `jti`, not revocable).
    token_sink: Option<Arc<dyn crate::auth::ContainerTokenSink>>,
    /// Configuration
    config: JobExecutorConfig,
    /// Shutdown flag
    shutdown: AtomicBool,
}

impl JobExecutor {
    /// Create a new job executor with default configuration
    pub fn new(runtime: Arc<dyn Runtime + Send + Sync>) -> Self {
        Self::with_config(runtime, JobExecutorConfig::default())
    }

    /// Create a new job executor with custom configuration
    pub fn with_config(runtime: Arc<dyn Runtime + Send + Sync>, config: JobExecutorConfig) -> Self {
        Self {
            runtime,
            executions: Arc::new(RwLock::new(HashMap::new())),
            job_specs: Arc::new(RwLock::new(HashMap::new())),
            overlay_manager: None,
            token_sink: None,
            config,
            shutdown: AtomicBool::new(false),
        }
    }

    /// Attach an overlay manager so job (and cron) containers join the overlay
    /// network and can reach the daemon API / peer services. Without this, a job
    /// container comes up with only a loopback netns and any call to the daemon
    /// node IP fails with `ENETUNREACH`.
    pub fn set_overlay_manager(&mut self, overlay_manager: Arc<RwLock<OverlayManager>>) {
        self.overlay_manager = Some(overlay_manager);
    }

    /// Set the sink used to persist/revoke per-container scoped tokens.
    pub fn set_token_sink(&mut self, sink: Arc<dyn crate::auth::ContainerTokenSink>) {
        self.token_sink = Some(sink);
    }

    /// Register a job spec (for later triggering)
    pub async fn register_job(&self, name: &str, spec: ServiceSpec) {
        let mut specs = self.job_specs.write().await;
        specs.insert(name.to_string(), spec);
        info!(job = %name, "Registered job spec");
    }

    /// Unregister a job spec
    pub async fn unregister_job(&self, name: &str) {
        let mut specs = self.job_specs.write().await;
        specs.remove(name);
        info!(job = %name, "Unregistered job spec");
    }

    /// Get a registered job spec
    pub async fn get_job_spec(&self, name: &str) -> Option<ServiceSpec> {
        let specs = self.job_specs.read().await;
        specs.get(name).cloned()
    }

    /// Names of all registered jobs, sorted.
    pub async fn registered_job_names(&self) -> Vec<String> {
        let specs = self.job_specs.read().await;
        let mut names: Vec<String> = specs.keys().cloned().collect();
        names.sort();
        names
    }

    /// The most recent execution for a job name, if any (the one that started
    /// last; a not-yet-started execution sorts before any started one).
    pub async fn latest_execution(&self, job_name: &str) -> Option<JobExecution> {
        let executions = self.executions.read().await;
        executions
            .values()
            .filter(|e| e.job_name == job_name)
            .max_by_key(|e| e.started_at)
            .cloned()
    }

    /// Trigger a job execution
    ///
    /// # Errors
    /// Returns an error if the job container cannot be created or started.
    pub async fn trigger(
        &self,
        job_name: &str,
        spec: &ServiceSpec,
        trigger: JobTrigger,
    ) -> Result<JobExecutionId> {
        if self.shutdown.load(Ordering::Relaxed) {
            return Err(AgentError::Internal("Job executor is shutting down".into()));
        }

        let exec_id = JobExecutionId::new();

        info!(
            job = %job_name,
            execution_id = %exec_id,
            trigger = %trigger,
            "Triggering job execution"
        );

        // Create execution record
        let execution = JobExecution {
            id: exec_id.clone(),
            job_name: job_name.to_string(),
            status: JobStatus::Pending,
            started_at: None,
            completed_at: None,
            container_id: None,
            logs: None,
            trigger,
        };

        // Store execution record
        {
            let mut executions = self.executions.write().await;
            executions.insert(exec_id.clone(), execution);
        }

        // Spawn the job execution task
        let runtime = self.runtime.clone();
        let spec = spec.clone();
        let exec_id_clone = exec_id.clone();
        let executions = self.executions.clone();
        let job_name = job_name.to_string();
        let max_log_size = self.config.max_log_size;
        let overlay_manager = self.overlay_manager.clone();
        let token_sink = self.token_sink.clone();

        tokio::spawn(async move {
            Box::pin(Self::run_job(
                runtime,
                executions,
                exec_id_clone,
                &job_name,
                spec,
                max_log_size,
                overlay_manager,
                token_sink,
            ))
            .await;
        });

        Ok(exec_id)
    }

    /// Internal: Run a job to completion
    #[allow(clippy::too_many_lines, clippy::too_many_arguments)]
    async fn run_job(
        runtime: Arc<dyn Runtime + Send + Sync>,
        executions: Arc<RwLock<HashMap<JobExecutionId, JobExecution>>>,
        exec_id: JobExecutionId,
        job_name: &str,
        spec: ServiceSpec,
        max_log_size: usize,
        overlay_manager: Option<Arc<RwLock<OverlayManager>>>,
        token_sink: Option<Arc<dyn crate::auth::ContainerTokenSink>>,
    ) {
        let started = Instant::now();

        // Update status to Initializing
        Self::update_status(&executions, &exec_id, |exec| {
            exec.status = JobStatus::Initializing;
            exec.started_at = Some(started);
        })
        .await;

        // Create container ID for this execution
        // Use a unique replica number based on execution ID hash
        let replica = exec_id.0.chars().take(8).collect::<String>();
        let replica_num = u32::from_str_radix(&replica, 16).unwrap_or(0) % 10000;
        let container_id = ContainerId::new(format!("job-{job_name}"), replica_num);

        // Store container ID
        Self::update_status(&executions, &exec_id, |exec| {
            exec.container_id = Some(container_id.clone());
        })
        .await;

        debug!(
            job = %job_name,
            execution_id = %exec_id,
            container_id = %container_id,
            "Creating job container"
        );

        // Pull image
        let image_str = spec.image.name.to_string();
        if let Err(e) = runtime
            .pull_image_with_policy(
                &image_str,
                spec.image.pull_policy,
                None,
                spec.image.source_policy.unwrap_or_default(),
            )
            .await
        {
            error!(
                job = %job_name,
                execution_id = %exec_id,
                error = %e,
                "Image pull failed"
            );
            Self::update_status(&executions, &exec_id, |exec| {
                exec.status = JobStatus::Failed {
                    reason: format!("Image pull failed: {e}"),
                    exit_code: None,
                };
                exec.completed_at = Some(Instant::now());
            })
            .await;
            return;
        }

        // Create container
        if let Err(e) = runtime.create_container(&container_id, &spec).await {
            let error_msg = e.to_string();
            error!(
                job = %job_name,
                execution_id = %exec_id,
                error = %error_msg,
                "Container create failed"
            );
            Self::update_status(&executions, &exec_id, |exec| {
                exec.status = JobStatus::Failed {
                    reason: format!("Container create failed: {error_msg}"),
                    exit_code: None,
                };
                exec.completed_at = Some(Instant::now());
            })
            .await;
            return;
        }

        // Attach the job container to the overlay network BEFORE it starts, so
        // its routes (eth0 service bridge + eth1 global overlay → the daemon node
        // IP) and resolv.conf are in place before init actions and the job's own
        // command run. Without this a job container has only loopback and any call
        // to the daemon node IP fails with `ENETUNREACH`. youki records the init
        // PID + netns at create (paused on the start fifo), so the attach is
        // race-free. Best-effort: attach failure is logged, not fatal.
        let overlay_attachment = Self::attach_overlay(
            overlay_manager.as_ref(),
            runtime.as_ref(),
            &container_id,
            job_name,
            &spec,
        )
        .await;

        // Run init steps
        let init_orchestrator = InitOrchestrator::new(container_id.clone(), spec.init.clone());
        if let Err(e) = init_orchestrator.run().await {
            let error_msg = e.to_string();
            error!(
                job = %job_name,
                execution_id = %exec_id,
                error = %error_msg,
                "Init failed"
            );
            Self::update_status(&executions, &exec_id, |exec| {
                exec.status = JobStatus::Failed {
                    reason: format!("Init failed: {error_msg}"),
                    exit_code: None,
                };
                exec.completed_at = Some(Instant::now());
            })
            .await;
            // Cleanup: detach overlay (reclaim veth/IP) then remove container.
            Self::detach_overlay(
                overlay_manager.as_ref(),
                runtime.as_ref(),
                &container_id,
                &overlay_attachment,
            )
            .await;
            Self::maybe_teardown_job_segment(
                overlay_manager.as_ref(),
                &executions,
                &exec_id,
                job_name,
            )
            .await;
            let _ = runtime.remove_container(&container_id).await;
            Self::revoke_token(token_sink.as_ref(), &container_id).await;
            return;
        }

        // Update status to Running
        Self::update_status(&executions, &exec_id, |exec| {
            exec.status = JobStatus::Running;
        })
        .await;

        debug!(
            job = %job_name,
            execution_id = %exec_id,
            "Starting job container"
        );

        // Start container
        if let Err(e) = runtime.start_container(&container_id).await {
            let error_msg = e.to_string();
            error!(
                job = %job_name,
                execution_id = %exec_id,
                error = %error_msg,
                "Container start failed"
            );
            Self::update_status(&executions, &exec_id, |exec| {
                exec.status = JobStatus::Failed {
                    reason: format!("Container start failed: {error_msg}"),
                    exit_code: None,
                };
                exec.completed_at = Some(Instant::now());
            })
            .await;
            // Cleanup: detach overlay (reclaim veth/IP) then remove container.
            Self::detach_overlay(
                overlay_manager.as_ref(),
                runtime.as_ref(),
                &container_id,
                &overlay_attachment,
            )
            .await;
            Self::maybe_teardown_job_segment(
                overlay_manager.as_ref(),
                &executions,
                &exec_id,
                job_name,
            )
            .await;
            let _ = runtime.remove_container(&container_id).await;
            Self::revoke_token(token_sink.as_ref(), &container_id).await;
            return;
        }

        // Wait for container to exit using the runtime's wait_container method
        let exit_code = runtime.wait_container(&container_id).await;
        let duration = started.elapsed();

        // Collect logs before cleanup using the runtime's get_logs method
        let logs = match runtime.get_logs(&container_id).await {
            Ok(entries) => Some(
                entries
                    .iter()
                    .map(ToString::to_string)
                    .collect::<Vec<_>>()
                    .join("\n"),
            ),
            Err(e) => {
                // Fallback to container_logs if get_logs fails
                match runtime.container_logs(&container_id, max_log_size).await {
                    Ok(entries) => Some(
                        entries
                            .iter()
                            .map(ToString::to_string)
                            .collect::<Vec<_>>()
                            .join("\n"),
                    ),
                    Err(e2) => {
                        warn!(
                            job = %job_name,
                            execution_id = %exec_id,
                            error = %e,
                            fallback_error = %e2,
                            "Failed to collect logs"
                        );
                        None
                    }
                }
            }
        };

        // Update final status
        Self::update_status(&executions, &exec_id, |exec| {
            exec.logs = logs;
            exec.completed_at = Some(Instant::now());

            match exit_code {
                Ok(code) => {
                    if code == 0 {
                        info!(
                            job = exec.job_name,
                            execution_id = %exec.id,
                            duration_ms = duration.as_millis(),
                            "Job completed successfully"
                        );
                        exec.status = JobStatus::Completed {
                            exit_code: code,
                            duration,
                        };
                    } else {
                        warn!(
                            job = exec.job_name,
                            execution_id = %exec.id,
                            exit_code = code,
                            duration_ms = duration.as_millis(),
                            "Job failed with non-zero exit code"
                        );
                        exec.status = JobStatus::Failed {
                            reason: format!("Non-zero exit code: {code}"),
                            exit_code: Some(code),
                        };
                    }
                }
                Err(err) => {
                    error!(
                        job = exec.job_name,
                        execution_id = %exec.id,
                        error = %err,
                        "Job execution error"
                    );
                    exec.status = JobStatus::Failed {
                        reason: err.to_string(),
                        exit_code: None,
                    };
                }
            }
        })
        .await;

        // Detach from the overlay (reclaim the veth + overlay IP) BEFORE removing
        // the container, so each job execution doesn't leak network resources.
        Self::detach_overlay(
            overlay_manager.as_ref(),
            runtime.as_ref(),
            &container_id,
            &overlay_attachment,
        )
        .await;
        Self::maybe_teardown_job_segment(overlay_manager.as_ref(), &executions, &exec_id, job_name)
            .await;

        // Cleanup container
        if let Err(e) = runtime.remove_container(&container_id).await {
            warn!(
                job = %job_name,
                execution_id = %exec_id,
                error = %e,
                "Failed to remove job container"
            );
        }

        // Revoke the job container's scoped token (best-effort).
        Self::revoke_token(token_sink.as_ref(), &container_id).await;
    }

    /// Attach a job container to the overlay network, mirroring the service
    /// path (`ServiceInstance` create flow): host-process runtimes (Linux youki)
    /// plumb a veth by PID; the macOS VZ guest gets an allocated overlay config
    /// pushed over vsock. `join_global = true` so the container gets the global
    /// overlay interface (and thus a route to the daemon node IP). Returns how it
    /// was attached so the caller can detach on exit. Best-effort: any failure is
    /// logged and yields [`OverlayAttachment::None`].
    #[cfg(not(target_os = "windows"))]
    #[allow(clippy::too_many_lines)]
    async fn attach_overlay(
        overlay_manager: Option<&Arc<RwLock<OverlayManager>>>,
        runtime: &(dyn Runtime + Send + Sync),
        container_id: &ContainerId,
        job_name: &str,
        spec: &ServiceSpec,
    ) -> OverlayAttachment {
        let Some(overlay) = overlay_manager else {
            return OverlayAttachment::None;
        };
        let guard = overlay.read().await;

        // Stand up the per-service overlay segment (bridge / dedicated WG / shared
        // bridge) BEFORE attaching — `attach_container` errors if the service
        // bridge doesn't exist. Idempotent: a repeat job execution reuses the
        // existing bridge. Mirrors the service restore path in the daemon.
        let mode = spec.overlay.as_ref().map(|o| o.mode).unwrap_or_default();
        if let Err(e) = guard.setup_service_overlay(job_name, mode).await {
            warn!(service = %job_name, error = %e, "failed to set up job overlay segment; job will have no overlay network");
            return OverlayAttachment::None;
        }

        // Per-deployment resolv.conf search domain (`<deployment>.<zone> <zone>`)
        // so the job's bare `<svc>` resolves to ITS deployment, matching the
        // service path's `dns_search_domain`. Falls back to the global zone inside
        // `attach_container` when this is `None`.
        let dns_override = guard.dns_domain().and_then(|zone| {
            spec.deployment.as_deref().map(|d| {
                let zone = zone.trim_end_matches('.');
                format!("{d}.{zone} {zone}")
            })
        });

        // Auto-fence isolation-scope modes (`Isolated`/`Dedicated`) to the job
        // name when no explicit `com.zlayer.isolation_network` label is set,
        // mirroring `ServiceInstance::isolation_network`.
        let isolation_network = {
            let explicit = spec
                .labels
                .get(zlayer_types::overlay::ISOLATION_NETWORK_LABEL)
                .cloned();
            crate::overlay_manager::resolve_isolation_network(mode, job_name, explicit)
        };

        match runtime.overlay_attach_kind_for(container_id).await {
            // Host-shared native runtime (macOS Seatbelt / native-VZ / libkrun):
            // overlayd allocates a distinct overlay /32 + utun alias for this job
            // container so it is a first-class overlay member like every other
            // runtime. Detach by the container id used at attach.
            OverlayAttachKind::HostProxy => {
                let cid = container_id.to_string();
                match guard
                    .attach_container_host_shared(
                        &cid,
                        job_name,
                        true,
                        isolation_network.clone(),
                        dns_override,
                    )
                    .await
                {
                    Ok(ip) => {
                        info!(container = %container_id, overlay_ip = %ip, "attached host-shared job container to overlay");
                        if let Err(e) = runtime.attach_overlay_ip(container_id, ip).await {
                            warn!(container = %container_id, error = %e, "failed to start host-shared overlay forwarders for job");
                        }
                        OverlayAttachment::HostShared(cid)
                    }
                    Err(e) => {
                        warn!(container = %container_id, error = %e, "failed to attach host-shared job container to overlay");
                        OverlayAttachment::None
                    }
                }
            }
            // macOS VZ-Linux guest: overlayd allocates the identity, we push it
            // into the guest over vsock.
            OverlayAttachKind::GuestManaged => {
                let cid = container_id.to_string();
                match guard
                    .attach_container_guest(
                        &cid,
                        job_name,
                        true,
                        isolation_network.clone(),
                        dns_override,
                    )
                    .await
                {
                    Ok(cfg) => match runtime.push_overlay_config(container_id, &cfg).await {
                        Ok(()) => {
                            info!(container = %container_id, overlay_ip = %cfg.overlay_ip, "attached job container to overlay (guest)");
                            OverlayAttachment::Guest(cid)
                        }
                        Err(e) => {
                            warn!(container = %container_id, error = %e, "failed to push overlay config into job guest; rolling back");
                            let _ = guard.detach_container_guest(&cid).await;
                            OverlayAttachment::None
                        }
                    },
                    Err(e) => {
                        warn!(container = %container_id, error = %e, "failed to allocate guest overlay config for job");
                        OverlayAttachment::None
                    }
                }
            }
            // Host-process runtimes (Linux youki): plumb a veth by PID.
            _ => match runtime.get_container_pid(container_id).await {
                Ok(Some(pid)) => match guard
                    .attach_container(pid, job_name, true, true, isolation_network, dns_override)
                    .await
                {
                    Ok(ip) => {
                        info!(container = %container_id, overlay_ip = %ip, "attached job container to overlay");
                        OverlayAttachment::Pid(pid)
                    }
                    Err(e) => {
                        warn!(container = %container_id, error = %e, "failed to attach job container to overlay network");
                        OverlayAttachment::None
                    }
                },
                Ok(None) => {
                    debug!(container = %container_id, "skipping job overlay attach - no PID available");
                    OverlayAttachment::None
                }
                Err(e) => {
                    warn!(container = %container_id, error = %e, "failed to read job container PID for overlay attach");
                    OverlayAttachment::None
                }
            },
        }
    }

    /// Windows containers receive their overlay (HCN) at container-create time
    /// inside overlayd, so there is no post-start attach step for jobs.
    #[cfg(target_os = "windows")]
    #[allow(clippy::unused_async)] // signature must match the Linux attach for the shared call site
    async fn attach_overlay(
        _overlay_manager: Option<&Arc<RwLock<OverlayManager>>>,
        _runtime: &(dyn Runtime + Send + Sync),
        _container_id: &ContainerId,
        _job_name: &str,
        _spec: &ServiceSpec,
    ) -> OverlayAttachment {
        OverlayAttachment::None
    }

    /// Release the overlay resources held by a job container. Mirrors the
    /// service path's detach-by-recorded-handle so an exited one-shot container
    /// never leaks its veth + overlay IP.
    #[cfg(not(target_os = "windows"))]
    async fn detach_overlay(
        overlay_manager: Option<&Arc<RwLock<OverlayManager>>>,
        runtime: &(dyn Runtime + Send + Sync),
        container_id: &ContainerId,
        attachment: &OverlayAttachment,
    ) {
        let Some(overlay) = overlay_manager else {
            return;
        };
        let guard = overlay.read().await;
        match attachment {
            OverlayAttachment::Pid(pid) => {
                if let Err(e) = guard.detach_container(*pid).await {
                    warn!(pid = pid, error = %e, "failed to detach job container from overlay (veth/IP may leak)");
                }
            }
            OverlayAttachment::Guest(id) => {
                if let Err(e) = guard.detach_container_guest(id).await {
                    warn!(id = %id, error = %e, "failed to detach job guest from overlay");
                }
            }
            OverlayAttachment::HostShared(id) => {
                if let Err(e) = runtime.detach_overlay_ip(container_id).await {
                    warn!(container = %container_id, error = %e, "failed to stop host-shared overlay forwarders for job");
                }
                if let Err(e) = guard.detach_container_host_shared(id).await {
                    warn!(id = %id, error = %e, "failed to detach host-shared job container from overlay");
                }
            }
            OverlayAttachment::None => {}
        }
    }

    /// Windows: overlay teardown happens at container-remove time inside
    /// overlayd, so there is no explicit job detach step.
    #[cfg(target_os = "windows")]
    #[allow(clippy::unused_async)] // signature must match the Linux detach for the shared call site
    async fn detach_overlay(
        _overlay_manager: Option<&Arc<RwLock<OverlayManager>>>,
        _runtime: &(dyn Runtime + Send + Sync),
        _container_id: &ContainerId,
        _attachment: &OverlayAttachment,
    ) {
    }

    /// Tear down the shared per-job overlay bridge segment created by
    /// `attach_overlay`'s `setup_service_overlay(job_name, mode)`. `detach_overlay`
    /// only reclaims the container's veth/IP — without this the per-job bridge
    /// (`zl-<deployment>-<instance>-<jobname>-b`) leaks one segment per execution.
    #[cfg(not(target_os = "windows"))]
    async fn maybe_teardown_job_segment(
        overlay_manager: Option<&Arc<RwLock<OverlayManager>>>,
        executions: &Arc<RwLock<HashMap<JobExecutionId, JobExecution>>>,
        exec_id: &JobExecutionId,
        job_name: &str,
    ) {
        let Some(overlay) = overlay_manager else {
            return;
        };
        // Only tear down the shared per-job bridge when NO OTHER execution of the
        // same job is still active — concurrent/overlapping cron runs share the
        // bridge by job_name, so tearing it down under a live sibling would cut
        // its network. This execution's own record is already terminal here.
        if Self::other_active_execution_exists(executions, exec_id, job_name).await {
            return;
        }
        let guard = overlay.read().await;
        guard.teardown_service_overlay(job_name).await;
    }

    /// Windows: overlay teardown happens at container-remove time inside
    /// overlayd, so there is no explicit per-job segment teardown step.
    #[cfg(target_os = "windows")]
    #[allow(clippy::unused_async)] // signature parity with the Linux path; Windows tears down at container-remove inside overlayd
    async fn maybe_teardown_job_segment(
        _overlay_manager: Option<&Arc<RwLock<OverlayManager>>>,
        _executions: &Arc<RwLock<HashMap<JobExecutionId, JobExecution>>>,
        _exec_id: &JobExecutionId,
        _job_name: &str,
    ) {
    }

    /// Whether any execution OTHER than `exec_id` is still active (pending /
    /// initializing / running) for the same `job_name`. Used to gate per-job
    /// bridge teardown so an overlapping cron sibling isn't cut off.
    #[cfg(not(target_os = "windows"))]
    async fn other_active_execution_exists(
        executions: &Arc<RwLock<HashMap<JobExecutionId, JobExecution>>>,
        exec_id: &JobExecutionId,
        job_name: &str,
    ) -> bool {
        let execs = executions.read().await;
        execs.values().any(|e| {
            &e.id != exec_id
                && e.job_name == job_name
                && matches!(
                    e.status,
                    JobStatus::Pending | JobStatus::Initializing | JobStatus::Running
                )
        })
    }

    /// Revoke a job container's scoped token (best-effort). The jti matches the
    /// deterministic `container:<service>:<service>-<replica>` string the runtime
    /// minted under, where `service == container_id.service` (`job-<job_name>`).
    async fn revoke_token(
        token_sink: Option<&Arc<dyn crate::auth::ContainerTokenSink>>,
        container_id: &ContainerId,
    ) {
        if let Some(sink) = token_sink {
            sink.revoke(&format!(
                "container:{}:{}-{}",
                container_id.service, container_id.service, container_id.replica
            ))
            .await;
        }
    }

    async fn update_status<F>(
        executions: &RwLock<HashMap<JobExecutionId, JobExecution>>,
        exec_id: &JobExecutionId,
        f: F,
    ) where
        F: FnOnce(&mut JobExecution),
    {
        let mut execs = executions.write().await;
        if let Some(exec) = execs.get_mut(exec_id) {
            f(exec);
        }
    }

    /// Get the status of a job execution
    pub async fn get_execution(&self, exec_id: &JobExecutionId) -> Option<JobExecution> {
        let executions = self.executions.read().await;
        executions.get(exec_id).cloned()
    }

    /// List all executions for a job
    pub async fn list_executions(&self, job_name: &str) -> Vec<JobExecution> {
        let executions = self.executions.read().await;
        executions
            .values()
            .filter(|e| e.job_name == job_name)
            .cloned()
            .collect()
    }

    /// List all executions (across all jobs)
    pub async fn list_all_executions(&self) -> Vec<JobExecution> {
        let executions = self.executions.read().await;
        executions.values().cloned().collect()
    }

    /// Cancel a running job execution
    ///
    /// # Errors
    /// Returns an error if the execution is not found or not in a cancellable state.
    pub async fn cancel(&self, exec_id: &JobExecutionId) -> Result<()> {
        let mut executions = self.executions.write().await;
        if let Some(execution) = executions.get_mut(exec_id) {
            if matches!(
                execution.status,
                JobStatus::Pending | JobStatus::Initializing | JobStatus::Running
            ) {
                if let Some(ref container_id) = execution.container_id {
                    self.runtime
                        .stop_container(container_id, Duration::from_secs(10))
                        .await?;
                    self.runtime.remove_container(container_id).await?;
                    Self::revoke_token(self.token_sink.as_ref(), container_id).await;
                }
                execution.status = JobStatus::Cancelled;
                execution.completed_at = Some(Instant::now());
                info!(
                    job = %execution.job_name,
                    execution_id = %exec_id,
                    "Job execution cancelled"
                );
            }
        }
        Ok(())
    }

    /// Clean up old execution records
    pub async fn cleanup_old_executions(&self) {
        let now = Instant::now();
        let mut executions = self.executions.write().await;
        let before_count = executions.len();
        executions.retain(|_, exec| match exec.completed_at {
            Some(completed) => now.duration_since(completed) < self.config.retention,
            None => true, // Keep running executions
        });
        let removed = before_count - executions.len();
        if removed > 0 {
            debug!(removed = removed, "Cleaned up old job execution records");
        }
    }

    /// Signal shutdown
    pub fn shutdown(&self) {
        self.shutdown.store(true, Ordering::Relaxed);
    }

    /// Check if executor is shutting down
    pub fn is_shutting_down(&self) -> bool {
        self.shutdown.load(Ordering::Relaxed)
    }

    /// Get the number of active (non-completed) executions
    pub async fn active_execution_count(&self) -> usize {
        let executions = self.executions.read().await;
        executions
            .values()
            .filter(|e| {
                matches!(
                    e.status,
                    JobStatus::Pending | JobStatus::Initializing | JobStatus::Running
                )
            })
            .count()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::runtime::MockRuntime;

    fn mock_job_spec() -> ServiceSpec {
        use zlayer_spec::*;
        serde_yaml::from_str::<DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  backup:
    rtype: job
    image:
      name: backup:latest
",
        )
        .unwrap()
        .services
        .remove("backup")
        .unwrap()
    }

    #[tokio::test]
    async fn test_job_execution_id() {
        let id1 = JobExecutionId::new();
        let id2 = JobExecutionId::new();
        assert_ne!(id1, id2);
        assert!(!id1.0.is_empty());
    }

    #[tokio::test]
    async fn test_job_executor_trigger() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let executor = JobExecutor::new(runtime);

        let spec = mock_job_spec();
        let exec_id = executor
            .trigger("backup", &spec, JobTrigger::Cli)
            .await
            .unwrap();

        // Give the job a moment to start
        tokio::time::sleep(Duration::from_millis(50)).await;

        let execution = executor.get_execution(&exec_id).await;
        assert!(execution.is_some());

        let exec = execution.unwrap();
        assert_eq!(exec.job_name, "backup");
        assert!(matches!(exec.trigger, JobTrigger::Cli));
    }

    #[tokio::test]
    async fn test_job_executor_list_executions() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let executor = JobExecutor::new(runtime);

        let spec = mock_job_spec();

        // Trigger multiple executions
        executor
            .trigger("backup", &spec, JobTrigger::Cli)
            .await
            .unwrap();
        executor
            .trigger("backup", &spec, JobTrigger::Scheduler)
            .await
            .unwrap();

        tokio::time::sleep(Duration::from_millis(50)).await;

        let executions = executor.list_executions("backup").await;
        assert_eq!(executions.len(), 2);
    }

    #[tokio::test]
    async fn test_job_executor_register_spec() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let executor = JobExecutor::new(runtime);

        let spec = mock_job_spec();
        executor.register_job("backup", spec.clone()).await;

        let retrieved = executor.get_job_spec("backup").await;
        assert!(retrieved.is_some());
        assert_eq!(retrieved.unwrap().image.name, spec.image.name);
    }

    #[cfg(not(target_os = "windows"))]
    fn job_exec(id: &str, job_name: &str, status: JobStatus) -> JobExecution {
        JobExecution {
            id: JobExecutionId(id.to_string()),
            job_name: job_name.to_string(),
            status,
            started_at: None,
            completed_at: None,
            container_id: None,
            logs: None,
            trigger: JobTrigger::Cli,
        }
    }

    #[cfg(not(target_os = "windows"))]
    #[tokio::test]
    async fn test_other_active_execution_exists() {
        use std::collections::HashMap;

        let current = JobExecutionId("current".to_string());

        // (a) current terminal + sibling same-job Running -> true
        {
            let mut map: HashMap<JobExecutionId, JobExecution> = HashMap::new();
            map.insert(
                current.clone(),
                job_exec(
                    "current",
                    "backup",
                    JobStatus::Completed {
                        exit_code: 0,
                        duration: Duration::from_secs(1),
                    },
                ),
            );
            map.insert(
                JobExecutionId("sibling".to_string()),
                job_exec("sibling", "backup", JobStatus::Running),
            );
            let executions = Arc::new(RwLock::new(map));
            assert!(
                JobExecutor::other_active_execution_exists(&executions, &current, "backup").await
            );
        }

        // (b) current terminal + sibling same-job Completed -> false
        {
            let mut map: HashMap<JobExecutionId, JobExecution> = HashMap::new();
            map.insert(
                current.clone(),
                job_exec(
                    "current",
                    "backup",
                    JobStatus::Failed {
                        reason: "boom".into(),
                        exit_code: Some(1),
                    },
                ),
            );
            map.insert(
                JobExecutionId("sibling".to_string()),
                job_exec(
                    "sibling",
                    "backup",
                    JobStatus::Completed {
                        exit_code: 0,
                        duration: Duration::from_secs(1),
                    },
                ),
            );
            let executions = Arc::new(RwLock::new(map));
            assert!(
                !JobExecutor::other_active_execution_exists(&executions, &current, "backup").await
            );
        }

        // (c) a Running execution of a DIFFERENT job_name -> false
        {
            let mut map: HashMap<JobExecutionId, JobExecution> = HashMap::new();
            map.insert(
                current.clone(),
                job_exec(
                    "current",
                    "backup",
                    JobStatus::Completed {
                        exit_code: 0,
                        duration: Duration::from_secs(1),
                    },
                ),
            );
            map.insert(
                JobExecutionId("other".to_string()),
                job_exec("other", "restore", JobStatus::Running),
            );
            let executions = Arc::new(RwLock::new(map));
            assert!(
                !JobExecutor::other_active_execution_exists(&executions, &current, "backup").await
            );
        }
    }

    #[tokio::test]
    async fn test_job_status_display() {
        assert_eq!(format!("{}", JobStatus::Pending), "pending");
        assert_eq!(format!("{}", JobStatus::Running), "running");
        assert_eq!(
            format!(
                "{}",
                JobStatus::Completed {
                    exit_code: 0,
                    duration: Duration::from_secs(10)
                }
            ),
            "completed(0)"
        );
        assert_eq!(
            format!(
                "{}",
                JobStatus::Failed {
                    reason: "error".into(),
                    exit_code: Some(1)
                }
            ),
            "failed(1)"
        );
    }
}