Merge remote-tracking branch 'origin/main' into feat/dynamic-workflow-support

# Conflicts:
#	crates/hm-exec/src/local/scheduler.rs

Author: markovejnovic

crates/hm-exec/src/cloud/watch.rs

@@ -17,6 +17,7 @@ use harmont_cloud::{
     HarmontClient, HarmontError,
     logs::{LogEvent, StreamKind},
     models::{build_is_terminal, job_is_terminal},
+    types::JobState,
 };
 use hm_plugin_protocol::events::{BuildEvent, PlanSummary, StdStream};
 use uuid::Uuid;
@@ -56,6 +57,27 @@ fn duration_ms(start: Option<DateTime<Utc>>, end: Option<DateTime<Utc>>) -> u64
     }
 }
 
+/// Whether a job has reached a state where its logs exist (running or already
+/// terminal), and so a log stream should be started for it.
+///
+/// Matching the typed [`JobState`] enum (rather than `to_string()`/`as_str()`
+/// against string literals) makes the set of states exhaustive: when the cloud
+/// adds a new `JobState` variant the compiler forces this decision to be
+/// revisited, and a misspelled state can no longer silently drop a job's logs.
+const fn job_logs_available(state: JobState) -> bool {
+    match state {
+        JobState::Running
+        | JobState::Passed
+        | JobState::Failed
+        | JobState::TimedOut
+        | JobState::Canceling
+        | JobState::Canceled
+        | JobState::TimingOut => true,
+        // No logs yet (not started) or never produced (skipped).
+        JobState::Pending | JobState::Scheduled | JobState::Assigned | JobState::Skipped => false,
+    }
+}
+
 /// Map a terminal build state to the process exit code the renderer and the
 /// `hm run` driver use. `passed` → 0, `canceled` → 130 (SIGINT-cancel, mirrors
 /// [`crate::BuildStatus::Canceled`]), everything else (`failed`, and any
@@ -131,18 +153,7 @@ pub async fn watch_build(
         // state where logs exist (running or already terminal).
         let jobs = client.list_jobs(org, pipeline, number).await?;
         for job in &jobs {
-            let state = job.state.to_string();
-            let logs_available = matches!(
-                state.as_str(),
-                "running"
-                    | "passed"
-                    | "failed"
-                    | "timed_out"
-                    | "canceling"
-                    | "canceled"
-                    | "timing_out"
-            );
-            if logs_available && streaming.insert(job.id) {
+            if job_logs_available(job.state) && streaming.insert(job.id) {
                 let name = job.name.clone().unwrap_or_else(|| "job".to_string());
                 let idx = *chain_idx.entry(job.id).or_insert_with(|| {
                     let i = next_idx;
@@ -400,7 +411,34 @@ async fn emit(
 
 #[cfg(test)]
 mod tests {
-    use super::exit_code_for_state;
+    use super::{JobState, exit_code_for_state, job_logs_available};
+
+    #[test]
+    fn logs_available_for_running_and_terminal_states() {
+        for state in [
+            JobState::Running,
+            JobState::Passed,
+            JobState::Failed,
+            JobState::TimedOut,
+            JobState::Canceling,
+            JobState::Canceled,
+            JobState::TimingOut,
+        ] {
+            assert!(job_logs_available(state), "expected logs for {state}");
+        }
+    }
+
+    #[test]
+    fn no_logs_before_start_or_when_skipped() {
+        for state in [
+            JobState::Pending,
+            JobState::Scheduled,
+            JobState::Assigned,
+            JobState::Skipped,
+        ] {
+            assert!(!job_logs_available(state), "expected no logs for {state}");
+        }
+    }
 
     #[test]
     fn passed_is_zero_canceled_is_130_else_is_one() {

crates/hm-exec/src/local/backend.rs

@@ -4,6 +4,7 @@
 //! drives the [`hm_vm`] subsystem. The VM backend (Docker, etc.) is injected;
 //! snapshot caching is owned by `hm-vm`'s [`hm_vm::ImageRegistry`].
 
+use std::num::NonZeroU64;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 
@@ -17,7 +18,7 @@ use crate::{BackendError, BackendHandle, Capabilities, ExecutionBackend, Result,
 
 /// Number of cached snapshots the image registry retains before evicting
 /// least-recently-used entries.
-const REGISTRY_CAPACITY: u64 = 64;
+const REGISTRY_CAPACITY: NonZeroU64 = NonZeroU64::new(64).expect("64 is non-zero");
 
 /// Runs the build locally via the in-process DAG scheduler, executing each
 /// step inside a VM provided by the injected [`hm_vm::VmBackend`].

crates/hm-exec/src/local/runner/mod.rs

@@ -11,7 +11,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 
 use anyhow::Result;
-use hm_plugin_protocol::{ExecutorInput, StepResult};
+use hm_plugin_protocol::{ExecutorInput, SnapshotRef, StepResult};
 use tokio_util::sync::CancellationToken;
 
 use crate::local::archive::ArchiveStore;
@@ -55,6 +55,21 @@ pub trait StepRunner: Send + Sync + fmt::Debug {
         ctx: &StepContext,
         input: ExecutorInput,
     ) -> Pin<Box<dyn Future<Output = Result<StepResult>> + Send + '_>>;
+
+    /// Reap transient snapshots once a run has finished.
+    ///
+    /// Ephemeral (uncached) leaf steps commit a snapshot purely so a
+    /// downstream `BuildsIn` child can restore from it; nothing else holds a
+    /// reference and the cache registry never tracks them. The scheduler
+    /// collects every such snapshot and calls this at run end (best-effort)
+    /// so they don't leak in the backend store. The default is a no-op for
+    /// runners that produce no reapable snapshots.
+    fn reap_snapshots<'a>(
+        &'a self,
+        _snapshots: Vec<SnapshotRef>,
+    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
+        Box::pin(async {})
+    }
 }
 
 /// Maps runner names to [`StepRunner`] implementations.

crates/hm-exec/src/local/runner/vm.rs

@@ -49,6 +49,21 @@ impl StepRunner for VmRunner {
         let vm = Arc::clone(&self.vm);
         Box::pin(async move { run_step_vm(&vm, &ctx, input).await })
     }
+
+    fn reap_snapshots<'a>(
+        &'a self,
+        snapshots: Vec<SnapshotRef>,
+    ) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>> {
+        let vm = Arc::clone(&self.vm);
+        Box::pin(async move {
+            for snap in snapshots {
+                let id = SnapshotId::new(snap.0);
+                if let Err(e) = vm.remove_snapshot(&id).await {
+                    tracing::warn!(snapshot = %id, error = %e, "failed to reap ephemeral snapshot");
+                }
+            }
+        })
+    }
 }
 
 #[tracing::instrument(skip(vm, ctx), fields(step_key = %input.step.key))]

crates/hm-exec/src/local/scheduler.rs

@@ -58,6 +58,12 @@ struct StepOutcome {
     exit_code: i32,
     snapshot: Option<SnapshotRef>,
     summaries: Vec<StepResultSummary>,
+    /// Set when this step did not complete successfully — it failed, timed
+    /// out, was cancelled, or was itself skipped. Descendants gate on this
+    /// (not on `exit_code`) so a skip propagates transitively: a skipped
+    /// step reports `exit_code == 0`, so the exit code alone cannot
+    /// distinguish "passed" from "skipped" and the cascade would break.
+    failed_or_skipped: bool,
 }
 
 type StepFuture = futures::future::Shared<BoxFuture<'static, StepOutcome>>;
@@ -325,8 +331,12 @@ pub(crate) async fn run(
             let pred_outcomes: Vec<StepOutcome> =
                 join_all(preds.iter().map(|(_, f)| f.clone())).await;
 
-            // Early exit if any predecessor failed or the build was cancelled.
-            if cancel.is_cancelled() || pred_outcomes.iter().any(|o| o.exit_code != 0) {
+            // Early exit if any predecessor failed/was skipped, or the build
+            // was cancelled. Gating on `failed_or_skipped` (not `exit_code`)
+            // is what makes the skip propagate transitively: a skipped
+            // predecessor reports `exit_code == 0`, so an exit-code-only gate
+            // would let a skipped step's descendants run anyway.
+            if cancel.is_cancelled() || pred_outcomes.iter().any(|o| o.failed_or_skipped) {
                 let status = if cancel.is_cancelled() {
                     StepStatus::Canceled
                 } else {
@@ -342,6 +352,7 @@ pub(crate) async fn run(
                         exit_code: None,
                         duration_ms: 0,
                     }],
+                    failed_or_skipped: true,
                 };
             }
 
@@ -392,6 +403,7 @@ pub(crate) async fn run(
                             exit_code: Some(1),
                             duration_ms: 0,
                         }],
+                        failed_or_skipped: true,
                     }
                 }
             }
@@ -428,6 +440,22 @@ pub(crate) async fn run(
     let outcomes: Vec<StepOutcome> = join_all(pending).await;
     let any_failed = outcomes.iter().any(|o| o.exit_code != 0);
 
+    // Reap ephemeral leaf snapshots. Uncached steps commit an `ephemeral:*`
+    // image for downstream container lineage; the cache registry never tracks
+    // them, so once the run is over nothing else will. Collect every such
+    // snapshot the steps produced and ask the default runner to remove them
+    // (best-effort — failures are logged, not fatal).
+    let ephemeral: Vec<SnapshotRef> = outcomes
+        .iter()
+        .filter_map(|o| o.snapshot.clone())
+        .filter(|s| s.0.starts_with("ephemeral:"))
+        .collect();
+    if !ephemeral.is_empty()
+        && let Some(runner) = runner_registry.resolve(None)
+    {
+        runner.reap_snapshots(ephemeral).await;
+    }
+
     // Derive the overall verdict. Timeout wins (it also fired cancellation);
     // then cancellation; then any failed step; otherwise the build passed.
     let status = if timed_out {
@@ -552,6 +580,7 @@ async fn execute_step(
                 exit_code: 0,
                 snapshot: None,
                 summaries: vec![summary],
+                failed_or_skipped: true,
             });
             continue;
         }
@@ -612,10 +641,15 @@ async fn execute_step(
     } else {
         None
     };
+    let failed_or_skipped = outcomes
+        .iter()
+        .flatten()
+        .any(|outcome| outcome.failed_or_skipped);
     Ok(StepOutcome {
         exit_code: first_failure.unwrap_or(0),
         snapshot,
         summaries,
+        failed_or_skipped,
     })
 }
 
@@ -641,10 +675,13 @@ async fn execute_command(
     let step_key = step_wire.key.clone();
     let display_name = step_wire.label.clone().unwrap_or_else(|| {
         let cmd = step_wire.cmd.trim();
-        if cmd.len() <= 40 {
+        if cmd.chars().count() <= 40 {
             cmd.to_owned()
         } else {
-            format!("{}…", &cmd[..39])
+            // Truncate on a char boundary, not a byte offset: `&cmd[..39]`
+            // panics if byte 39 falls inside a multibyte UTF-8 sequence.
+            let truncated: String = cmd.chars().take(39).collect();
+            format!("{truncated}…")
         }
     });
     let env_map = transition.env;
@@ -754,6 +791,7 @@ async fn execute_command(
                             exit_code: Some(124),
                             duration_ms: dur_ms,
                         }],
+                        failed_or_skipped: true,
                     });
                 }
             }
@@ -800,6 +838,7 @@ async fn execute_command(
                     exit_code: Some(sr.exit_code),
                     duration_ms: dur_ms,
                 }],
+                failed_or_skipped: sr.exit_code != 0,
             })
         }
         Err(e) => {