refactor(hm-pipeline-ir): centralize duration-ms conversion in DurationMs

crates/hm-common/src/format.rs

@@ -5,11 +5,12 @@ use core::time::Duration;
 
 mod sealed {
     pub trait Sealed {}
-    impl Sealed for core::time::Duration {}
+    impl<T: Into<core::time::Duration>> Sealed for T {}
 }
 
-/// Extension trait adding a compact, stopwatch-style [`Display`] rendering to
-/// [`std::time::Duration`].
+/// Adds a compact, stopwatch-style [`Display`] rendering to any duration.
+///
+/// Implemented for any type convertible into a [`std::time::Duration`].
 ///
 /// ```
 /// # use hm_common::format::CompactDuration;
@@ -21,10 +22,11 @@ pub trait CompactDuration: sealed::Sealed {
     fn compact(self) -> StopwatchDurationDisplay;
 }
 
-impl CompactDuration for Duration {
+impl<T: Into<Duration>> CompactDuration for T {
     fn compact(self) -> StopwatchDurationDisplay {
+        let d: Duration = self.into();
         StopwatchDurationDisplay {
-            total_ms: u64::try_from(self.as_millis()).unwrap_or(u64::MAX),
+            total_ms: u64::try_from(d.as_millis()).unwrap_or(u64::MAX),
         }
     }
 }

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

@@ -18,6 +18,7 @@ use harmont_cloud::{
     logs::{LogEvent, StreamKind},
     models::{HarmontJob, OpenJobState, build_is_terminal, job_is_terminal},
 };
+use hm_pipeline_ir::DurationMs;
 use hm_plugin_protocol::events::{BuildEvent, PlanSummary, StdStream};
 use uuid::Uuid;
 
@@ -47,15 +48,6 @@ pub(crate) fn ts_or_now(ts_unix_ns: Option<i64>) -> DateTime<Utc> {
     ts_unix_ns.map_or_else(Utc::now, DateTime::<Utc>::from_timestamp_nanos)
 }
 
-/// Duration between two optional timestamps, in milliseconds (0 if either is
-/// missing or the interval is negative).
-fn duration_ms(start: Option<DateTime<Utc>>, end: Option<DateTime<Utc>>) -> u64 {
-    match (start, end) {
-        (Some(s), Some(e)) => (e - s).num_milliseconds().max(0).cast_unsigned(),
-        _ => 0,
-    }
-}
-
 /// Parse an optional RFC 3339 timestamp string (the form the v1 API serializes
 /// `started_at` / `finished_at` as) into a UTC datetime, dropping unparseable
 /// or absent values to `None`.
@@ -260,8 +252,7 @@ pub async fn watch_build(
     let _ = tx
         .send(BuildEvent::BuildEnd {
             exit_code: code,
-            // Saturate at u64::MAX (~584 million years) rather than panic.
-            duration_ms: u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
+            duration_ms: DurationMs::from(started.elapsed()),
         })
         .await;
     Ok(code)
@@ -276,13 +267,19 @@ fn step_end(job: &HarmontJob, step_id: Uuid) -> BuildEvent {
         .exit_code
         // Saturate exit codes outside [i32::MIN, i32::MAX] rather than panic.
         .map_or_else(|| i32::from(!passed), |c| i32::try_from(c).unwrap_or(1));
+    // Duration between the job's recorded start/finish timestamps, in
+    // milliseconds (0 if either is missing or the interval is negative).
+    let duration_ms = match (
+        parse_rfc3339(job.started_at.as_deref()),
+        parse_rfc3339(job.finished_at.as_deref()),
+    ) {
+        (Some(s), Some(e)) => DurationMs((e - s).num_milliseconds().max(0).cast_unsigned()),
+        _ => DurationMs(0),
+    };
     BuildEvent::StepEnd {
         step_id,
         exit_code,
-        duration_ms: duration_ms(
-            parse_rfc3339(job.started_at.as_deref()),
-            parse_rfc3339(job.finished_at.as_deref()),
-        ),
+        duration_ms,
         snapshot: None,
     }
 }

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

@@ -56,7 +56,7 @@ mod tests {
         let mut rx = bus.subscribe();
         bus.emit(BuildEvent::BuildEnd {
             exit_code: 0,
-            duration_ms: 1,
+            duration_ms: hm_pipeline_ir::DurationMs(1),
         });
         let ev = rx.recv().await.unwrap();
         matches!(ev, BuildEvent::BuildEnd { exit_code: 0, .. });
@@ -67,7 +67,7 @@ mod tests {
         let bus = EventBus::new();
         bus.emit(BuildEvent::BuildEnd {
             exit_code: 0,
-            duration_ms: 0,
+            duration_ms: hm_pipeline_ir::DurationMs(0),
         });
         // Should not panic.
     }

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

@@ -39,7 +39,7 @@ use hm_plugin_protocol::{
 };
 use uuid::Uuid;
 
-use hm_pipeline_ir::{EdgeKind, PipelineGraph, Transition};
+use hm_pipeline_ir::{DurationMs, EdgeKind, PipelineGraph, Transition};
 
 use crate::local::runner::{RunnerRegistry, StepContext};
 use crate::local::source::build_archive_bytes;
@@ -211,7 +211,7 @@ pub(crate) async fn run(
                         key: node_key,
                         status,
                         exit_code: None,
-                        duration_ms: 0,
+                        duration_ms: DurationMs(0),
                     }),
                     failed_or_skipped: true,
                 };
@@ -258,7 +258,7 @@ pub(crate) async fn run(
                             key: node_key,
                             status: StepStatus::Failed,
                             exit_code: Some(1),
-                            duration_ms: 0,
+                            duration_ms: DurationMs(0),
                         }),
                         failed_or_skipped: true,
                     }
@@ -334,7 +334,7 @@ pub(crate) async fn run(
 
     let steps: Vec<StepResultSummary> = outcomes.iter().filter_map(|o| o.summary.clone()).collect();
 
-    let dur = started_total.elapsed().as_millis() as u64;
+    let dur = DurationMs::from(started_total.elapsed());
 
     bus.emit(BuildEvent::BuildEnd {
         exit_code: status.exit_code(),
@@ -479,7 +479,7 @@ async fn execute_step(
                     // Per-step wall-clock budget exceeded. Emit a step-end with the
                     // conventional timeout exit code (124), fail the chain, and
                     // cancel siblings — same shape as a non-zero exit below.
-                    let dur_ms = started.elapsed().as_millis() as u64;
+                    let dur_ms = DurationMs::from(started.elapsed());
                     bus.emit(BuildEvent::StepEnd {
                         step_id,
                         exit_code: 124,
@@ -515,7 +515,7 @@ async fn execute_step(
         _ => exec.await,
     };
 
-    let dur_ms = started.elapsed().as_millis() as u64;
+    let dur_ms = DurationMs::from(started.elapsed());
     match result {
         Ok(sr) => {
             bus.emit(BuildEvent::StepEnd {

crates/hm-exec/src/outcome.rs

@@ -1,6 +1,7 @@
 //! The typed result of a backend run.
 
 use chrono::{DateTime, Utc};
+use hm_pipeline_ir::DurationMs;
 use hm_plugin_protocol::events::BuildRef;
 use uuid::Uuid;
 
@@ -48,7 +49,7 @@ pub struct StepResultSummary {
     pub key: String,
     pub status: StepStatus,
     pub exit_code: Option<i32>,
-    pub duration_ms: u64,
+    pub duration_ms: DurationMs,
 }
 
 /// The terminal result of a build.

crates/hm-exec/tests/backend_contract.rs

@@ -11,6 +11,7 @@ use futures::StreamExt;
 use rstest::rstest;
 use hm_exec::*;
 use hm_plugin_protocol::events::{BuildEvent, BuildRef};
+use hm_plugin_protocol::ir::DurationMs;
 use tokio_util::sync::CancellationToken;
 
 #[derive(Debug)]
@@ -37,7 +38,7 @@ impl ExecutionBackend for FakeBackend {
             let _ = tx
                 .send(BuildEvent::BuildEnd {
                     exit_code: 0,
-                    duration_ms: 5,
+                    duration_ms: DurationMs(5),
                 })
                 .await;
             Ok(BuildOutcome {

crates/hm-pipeline-ir/Cargo.toml

@@ -8,6 +8,7 @@ description = "Pipeline IR — the v0 wire-format schema consumed by hm."
 
 [dependencies]
 daggy      = { workspace = true }
+derive_more = { workspace = true }
 serde      = { workspace = true }
 serde_json = { workspace = true }
 schemars   = { workspace = true }

crates/hm-pipeline-ir/src/duration.rs

@@ -0,0 +1,93 @@
+//! A wire-friendly duration scalar.
+
+use std::time::Duration;
+
+use schemars::JsonSchema as DeriveJsonSchema;
+use serde::{Deserialize, Serialize};
+
+/// A duration in whole milliseconds, carried on the wire as a bare JSON number.
+///
+/// `Display` renders the bare millisecond count (no unit suffix), matching the
+/// wire form.
+#[derive(
+    Debug,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Serialize,
+    Deserialize,
+    DeriveJsonSchema,
+    derive_more::Display,
+)]
+#[serde(transparent)]
+pub struct DurationMs(pub u64);
+
+impl DurationMs {
+    /// The millisecond count.
+    #[must_use]
+    pub const fn get(self) -> u64 {
+        self.0
+    }
+}
+
+impl From<Duration> for DurationMs {
+    fn from(d: Duration) -> Self {
+        Self(u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
+    }
+}
+
+impl From<DurationMs> for Duration {
+    /// Exact: a millisecond count always widens losslessly into a [`Duration`].
+    fn from(d: DurationMs) -> Self {
+        Self::from_millis(d.0)
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used, reason = "test setup and assertions")]
+mod tests {
+    use super::*;
+    use rstest::rstest;
+
+    #[rstest]
+    #[case::zero(0, 0)]
+    #[case::one(1, 1)]
+    #[case::typical(1_500, 1_500)]
+    fn from_duration_preserves_millis(#[case] millis: u64, #[case] expected: u64) {
+        assert_eq!(DurationMs::from(Duration::from_millis(millis)).get(), expected);
+    }
+
+    #[rstest]
+    fn from_duration_saturates_at_u64_max() {
+        // A duration whose millisecond count exceeds u64::MAX clamps rather than
+        // wrapping or panicking. `Duration::MAX` is ~5.85e12 seconds * 1000 ms,
+        // well past u64::MAX ms.
+        assert_eq!(DurationMs::from(Duration::MAX), DurationMs(u64::MAX));
+    }
+
+    #[rstest]
+    fn converts_to_duration_losslessly() {
+        let d = DurationMs(1_234);
+        assert_eq!(Duration::from(d), Duration::from_millis(1_234));
+    }
+
+    #[rstest]
+    #[case::zero(DurationMs(0), "0")]
+    #[case::typical(DurationMs(1_500), "1500")]
+    fn display_is_bare_integer(#[case] d: DurationMs, #[case] expected: &str) {
+        assert_eq!(d.to_string(), expected);
+    }
+
+    #[rstest]
+    fn serializes_as_bare_json_number() {
+        // `#[serde(transparent)]` means the wire form is a plain number, not an
+        // object — the whole point of the newtype for cross-language consumers.
+        let json = serde_json::to_string(&DurationMs(1_500)).unwrap();
+        assert_eq!(json, "1500");
+        let back: DurationMs = serde_json::from_str(&json).unwrap();
+        assert_eq!(back, DurationMs(1_500));
+    }
+}

crates/hm-pipeline-ir/src/lib.rs

@@ -7,6 +7,8 @@
 #![forbid(unsafe_code)]
 #![allow(clippy::multiple_crate_versions, clippy::cargo_common_metadata)]
 
+mod duration;
 mod graph;
 
+pub use duration::DurationMs;
 pub use graph::{Cache, CommandStep, EdgeKind, PipelineGraph, Transition};

crates/hm-plugin-cloud/src/verbs/job.rs

@@ -6,6 +6,7 @@ use anyhow::Result;
 use chrono::Utc;
 use harmont_cloud::HarmontClient;
 use hm_plugin_protocol::events::{BuildEvent, PlanSummary};
+use hm_plugin_protocol::ir::DurationMs;
 use uuid::Uuid;
 
 use crate::cli::JobCommand;
@@ -117,14 +118,14 @@ async fn log_cmd(
         .send(BuildEvent::StepEnd {
             step_id: job_id,
             exit_code: 0,
-            duration_ms: 0,
+            duration_ms: DurationMs(0),
             snapshot: None,
         })
         .await;
     let _ = tx
         .send(BuildEvent::BuildEnd {
             exit_code: 0,
-            duration_ms: 0,
+            duration_ms: DurationMs(0),
         })
         .await;
     let _ = driver.await;

crates/hm-plugin-protocol/src/events.rs

@@ -8,6 +8,7 @@ use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
 use crate::executor::SnapshotRef;
+use crate::ir::DurationMs;
 
 #[derive(
     Debug,
@@ -73,7 +74,7 @@ pub enum BuildEvent {
     StepEnd {
         step_id: Uuid,
         exit_code: i32,
-        duration_ms: u64,
+        duration_ms: DurationMs,
         snapshot: Option<SnapshotRef>,
     },
     /// Emitted when any step in a chain returns non-zero. Carries the
@@ -90,7 +91,7 @@ pub enum BuildEvent {
     },
     BuildEnd {
         exit_code: i32,
-        duration_ms: u64,
+        duration_ms: DurationMs,
     },
 }
 

crates/hm-plugin-protocol/src/ir.rs

@@ -2,4 +2,4 @@
 //! live in the `hm-pipeline-ir` crate; this module keeps the
 //! `hm_plugin_protocol::ir::*` import path working.
 
-pub use hm_pipeline_ir::{Cache, CommandStep};
+pub use hm_pipeline_ir::{Cache, CommandStep, DurationMs};

crates/hm-plugin-protocol/tests/round_trip.rs

@@ -11,6 +11,7 @@
     reason = "serde round-trip tests: helpers unwrap and assert on JSON"
 )]
 
+use hm_plugin_protocol::ir::DurationMs;
 use hm_plugin_protocol::*;
 use rstest::rstest;
 use uuid::Uuid;
@@ -86,7 +87,7 @@ fn executor_input_round_trip() {
 #[case::step_end(BuildEvent::StepEnd {
     step_id: Uuid::nil(),
     exit_code: 0,
-    duration_ms: 1,
+    duration_ms: DurationMs(1),
     snapshot: None,
 })]
 #[case::chain_failed(BuildEvent::ChainFailed {
@@ -99,7 +100,7 @@ fn executor_input_round_trip() {
 })]
 #[case::build_end(BuildEvent::BuildEnd {
     exit_code: 0,
-    duration_ms: 2,
+    duration_ms: DurationMs(2),
 })]
 fn build_event_round_trips(#[case] event: BuildEvent) {
     rt(&event);