feat!: mount function for injecting aditional files into the workspace

crates/hm-dsl-engine/harmont-py/harmont/__init__.py

@@ -40,7 +40,7 @@
 from ._pipeline import pipeline_to_json
 from ._python import python
 from ._rust import RustProject, rust
-from ._step import Step, scratch, wait
+from ._step import Step, StepAction, scratch, wait
 from ._target import clear_target_cache, target  # noqa: F401  clear_target_cache used by tests
 from ._toolchain import apt_base
 from ._typing import BaseImage, Target
@@ -276,6 +276,46 @@ def sh(
     )
 
 
+def mount(
+    *,
+    from_: str,
+    to: str,
+    label: str = ":ext: dir",
+    image: str | None = None,
+    strict: bool = True,
+) -> Step:
+    """Declare a workspace-directory bind mount.
+
+    Shorthand for ``scratch().mount(from_=from_, to=to, ...)``.
+    All keyword arguments are forwarded to ``Step.mount``.
+
+    Args:
+        from_: Source directory relative to the workspace root.
+        to: Destination directory inside the step container,
+            relative to the workspace root.
+        label: Human-facing label shown in the UI.
+        image: Local-mode Docker base image override for this step.
+        strict: Aditional check which validates if the source file
+            exists during the function's call. Defaults to ``True``.
+            It's not included in the IR.
+
+    Returns:
+        A new root ``Step`` with the mount set.
+
+    Examples:
+        >>> import harmont as hm
+        >>> step = hm.mount(from_=".cache/npm", to="node_modules")
+    """
+    return scratch().mount(
+        from_=from_,
+        to=to,
+        cache=CacheNone(),
+        label=label,
+        image=image,
+        strict=strict,
+    )
+
+
 def group(steps: list[Step] | tuple[Step, ...]) -> tuple[Step, ...]:
     """Collect a list of steps into a tuple for use as a target return value.
 
@@ -310,6 +350,7 @@ def group(steps: list[Step] | tuple[Step, ...]) -> tuple[Step, ...]:
     "Pipeline",
     "RustProject",
     "Step",
+    "StepAction",
     "Target",
     "apt_base",
     "cmake",
@@ -320,6 +361,7 @@ def group(steps: list[Step] | tuple[Step, ...]) -> tuple[Step, ...]:
     "go",
     "group",
     "js",
+    "mount"
     "on_change",
     "pipeline",
     "pipeline_to_json",

crates/hm-dsl-engine/harmont-py/harmont/_keys.py

@@ -17,10 +17,12 @@
 import re
 from typing import TYPE_CHECKING
 
+from ._step import Command, Mount
+
 if TYPE_CHECKING:
     from collections.abc import Iterable
 
-    from ._step import Step
+    from ._step import Step, StepAction
 
 _EMOJI_SHORTCODE_RE = re.compile(r":[a-z0-9_+-]+:")
 _NON_ALNUM_RE = re.compile(r"[^a-z0-9]+")
@@ -43,15 +45,21 @@ def slugify_label(label: str) -> str:
     return s.strip("-")
 
 
-def hash_key(parent_key: str, cmd: str, position: int) -> str:
+def hash_key(parent_key: str, action: StepAction | None, position: int) -> str:
     """Stable 12-char SHA-256 prefix over (parent_key, cmd, position).
 
     Used as the fallback key when no usable slug is available."""
     h = hashlib.sha256()
     h.update(parent_key.encode("utf-8"))
     h.update(b"\x00")
-    h.update(cmd.encode("utf-8"))
-    h.update(b"\x00")
+    if isinstance(action, Command):
+        h.update(action.cmd.encode("utf-8"))
+        h.update(b"\x00")
+    elif isinstance(action, Mount):
+        h.update(action.from_.encode("utf-8"))
+        h.update(b"\x00")
+        h.update(action.to.encode("utf-8"))
+        h.update(b"\x00")
     h.update(str(position).encode("utf-8"))
     return h.hexdigest()[:12]
 
@@ -117,5 +125,5 @@ def resolve_keys(steps: Iterable[Step]) -> dict[int, str]:
         parent_key = ""
         if s.parent is not None and id(s.parent) in keys:
             parent_key = keys[id(s.parent)]
-        keys[sid] = hash_key(parent_key, s.cmd or "", position)
+        keys[sid] = hash_key(parent_key, s.action, position)
     return keys

crates/hm-dsl-engine/harmont-py/harmont/_pipeline.py

@@ -15,6 +15,7 @@
 
 from ._duration import parse_duration
 from ._keys import resolve_keys
+from ._step import Command, Mount
 from .cache import (
     CacheCompose,
     CacheForever,
@@ -25,7 +26,7 @@
 )
 
 if TYPE_CHECKING:
-    from ._step import Step
+    from ._step import Step, StepAction
 
 # Across-the-board default image for imageless root steps. The SDK's
 # toolchains assume an apt-capable base (apt-get), so ubuntu:24.04 is the
@@ -34,6 +35,19 @@
 DEFAULT_IMAGE = "ubuntu:24.04"
 
 
+def _action_to_dict(action: StepAction) -> dict[str, Any]:
+    action_dict = {}
+    if isinstance(action, Command):
+        action_dict["cmd"] = action.cmd
+        if action.env:
+            action_dict["env"] = action.env
+    elif isinstance(action, Mount):
+        action_dict["from"] = action.from_
+        action_dict["to"] = action.to
+
+    return action_dict
+
+
 def pipeline(
     leaves: list[Step] | tuple[Step, ...],
     *,
@@ -76,7 +90,7 @@ def _lower_to_graph(
     explicit ``depends_on`` edges.
     """
     ordered = _topo_collect(leaves)
-    command_steps = [s for s in ordered if s.cmd is not None and not s.is_wait]
+    command_steps = [s for s in ordered if s.action and not s.is_wait]
     keys = resolve_keys(command_steps)
 
     # Assign integer node indices (dense, in emission order).
@@ -105,18 +119,16 @@ def _lower_to_graph(
             pre_wait_indices = []
             continue
 
-        if s.cmd is None:
+        if not s.action:
             # scratch or fork — passthrough, not emitted.
             continue
 
         node_idx = idx_by_id[id(s)]
         step_key = keys[id(s)]
 
-        # Build the CommandStep dict (no "type" or "builds_in" fields).
-        step_dict: dict[str, Any] = {
-            "key": step_key,
-            "cmd": s.cmd,
-        }
+        # Build the Step dict (no "type" or "builds_in" fields).
+        step_dict: dict[str, Any] = {"key": step_key, "action": _action_to_dict(s.action)}
+
         if s.label is not None:
             step_dict["label"] = s.label
         if s.cache is not None:
@@ -137,8 +149,8 @@ def _lower_to_graph(
         }
         if env:
             merged_env.update(env)
-        if s.env:
-            merged_env.update(s.env)
+        if isinstance(s.action, Command) and s.action.env:
+            merged_env.update(s.action.env)
 
         nodes.append({"step": step_dict, "env": merged_env})
 
@@ -172,12 +184,12 @@ def _lower_to_graph(
 
 def _find_idx_by_key(
     key: str,
-    command_steps: list[Step],
+    steps: list[Step],
     keys: dict[int, str],
     idx_by_id: dict[int, int],
 ) -> int:
     """Return the node index for the step with the given resolved key."""
-    for s in command_steps:
+    for s in steps:
         if keys[id(s)] == key:
             return idx_by_id[id(s)]
     msg = f"BUG: no step with key {key!r}"
@@ -216,7 +228,7 @@ def _resolved_parent_key(s: Step, keys: dict[int, str]) -> str | None:
     """Walk back through scratch/fork nodes to the nearest emitted ancestor."""
     node = s.parent
     while node is not None:
-        if node.cmd is not None and not node.is_wait:
+        if node.action and not node.is_wait:
             return keys[id(node)]
         node = node.parent
     return None

crates/hm-dsl-engine/harmont-py/harmont/_step.py

@@ -8,22 +8,40 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Union
 
 if TYPE_CHECKING:
     from .cache import CachePolicy
 
 
+@dataclass(frozen=True)
+class Command:
+    cmd: str
+    env: dict[str, str] | None = None
+
+
+@dataclass(frozen=True)
+class Mount:
+    from_: str
+    to: str
+
+
+StepAction = Union["Command", "Mount"]
+
+
 @dataclass(frozen=True)
 class Step:
     """Immutable chain node — the primitive the DSL is built on.
 
     Steps are constructed via `scratch()` or `wait()` and extended by
-    calling ``.sh()`` or ``.fork()`` on the result. Every mutating method
+    calling ``.sh()``, ``mount()`` or ``.fork()`` on the result. Every mutating method
     returns a new ``Step``; the receiver is unchanged.
     """
 
-    cmd: str | None = None
+    action: StepAction | None = None
+    """Action realized by the Step. Can be either a Mount or a Commmand"""
+
     parent: Step | None = None
     """In-tree pointer used by the lowering pass to walk back to the
     nearest emitted ancestor. Distinct from the wire-format
@@ -33,7 +51,6 @@ class Step:
     continue_on_failure: bool = False
     label: str | None = None
     cache: CachePolicy | None = None
-    env: dict[str, str] | None = None
     timeout_seconds: int | None = None
     image: str | None = None
     """Local-mode Docker base image override for this step. Ignored when
@@ -53,6 +70,86 @@ class Step:
     The field is renamed so it doesn't shadow the runtime-derived key
     the lowering pass produces in pipeline.py."""
 
+    def mount(
+        self,
+        *,
+        from_: str,
+        to: str,
+        label: str | None = None,
+        cache: CachePolicy | None = None,
+        image: str | None = None,
+        runner: str | None = None,
+        runner_args: dict[str, Any] | None = None,
+        key: str | None = None,
+        strict: bool = True,
+    ) -> Step:
+        """Injects an external directory into the workspace during the pipeline execution.
+
+        Returns a new ``Step``; the receiver is unchanged (steps are immutable).
+
+        The mount is a hint to the executor to bind-mount the file/directory at
+        ``from_`` onto the container's path ``to``.
+        Both paths must be relative to the workspace root and ``to``
+        should always point to a directory inside the workspace.
+
+        Args:
+            from_: Source directory relative to the workspace root.
+            to: Destination directory inside the step container, relative
+                to the workspace root. Must exist as a directory or not
+                exist yet; must be inside the workspace.
+            label: Human-facing label shown in the UI.
+            cache: Cache policy controlling result reuse across builds.
+            image: Local-mode Docker base image for this step.
+            runner: Executor plugin runner name. ``None`` selects the
+                default Docker runner.
+            runner_args: Plugin-specific arguments validated by the
+                runner's schema.
+            key: Manual key override for this step in the v0 IR.
+            strict: Aditional check which validates if the source file
+                exists during the function's call. Defaults to ``True``.
+                It's not included in the IR.
+
+        Returns:
+            A new ``Step`` with this mount appended to the chain.
+
+        Raises:
+            ValueError: If ``from_`` or ``to`` is an absolute path, if
+                ``to`` resolves to an existing file (non-directory), or
+                outside the workspace root.
+        """
+
+        effective_image = image or (self.image if self.action is None else None)
+        source_path = Path(from_)
+        dest_path = Path(to)
+        workspace = Path(".").resolve()
+
+        if strict and not source_path.exists():
+            msg = (
+                "Source path does not exists."
+                "If you are generating it dinamically"
+                "declare the ``strict`` argument as ``False``"
+            )
+            raise ValueError(msg)
+        if dest_path.is_absolute() or source_path.is_absolute():
+            msg = "Both paths should be relative to the workspace"
+            raise ValueError(msg)
+        if dest_path.exists() and not dest_path.is_dir():
+            msg = "Destination path of a mount should be a directory"
+            raise ValueError(msg)
+        if workspace == (r_dest := dest_path.resolve()) or workspace in r_dest.parents:
+            return Step(
+                action=Mount(from_=from_, to=to),
+                parent=self,
+                label=label,
+                cache=cache,
+                image=effective_image,
+                runner=runner,
+                runner_args=runner_args,
+                key_override=key,
+            )
+        msg = f"Destination path of a mount should be inside the workspace, but points to {to}"
+        raise ValueError(msg)
+
     def sh(
         self,
         cmd: str,
@@ -107,15 +204,12 @@ def sh(
         # passes it down to the first emitted command step. Once the
         # chain has a real cmd, inheritance stops — keeps wire format
         # identical for normal chains.
-        effective_image = (
-            image if image is not None else (self.image if self.cmd is None else None)
-        )
+        effective_image = image or (self.image if self.action is None else None)
         return Step(
-            cmd=effective_cmd,
+            action=Command(cmd=effective_cmd, env=env),
             parent=self,
             label=label,
             cache=cache,
-            env=env,
             image=effective_image,
             runner=runner,
             runner_args=runner_args,
@@ -135,7 +229,7 @@ def fork(self, label: str | None = None) -> Step:
         Returns:
             A new ``Step`` branching from this one.
         """
-        return Step(cmd=None, parent=self, label=label)
+        return Step(parent=self, label=label)
 
 
 def scratch() -> Step: