feat: Add dynamic workflow support

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

@@ -34,6 +34,7 @@
 from ._duration import parse_duration as _parse_duration
 from ._elixir import elixir
 from ._envelope import dump_registry_json
+from ._env import env
 from ._go import go
 from ._js import JsProject, js, ts
 from ._pipeline import pipeline as _pipeline_factory

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

@@ -0,0 +1,10 @@
+"""Environment access shared by static and dynamic pipeline evaluation."""
+
+from __future__ import annotations
+
+import os
+
+
+def env(name: str, default: str | None = None) -> str | None:
+    """Return an environment variable, or ``default`` when it is unset."""
+    return os.environ.get(name, default)

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

@@ -117,5 +117,6 @@ 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)
+        key_material = s.cmd or s.dynamic_target_name or ""
+        keys[sid] = hash_key(parent_key, key_material, position)
     return keys

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

@@ -78,12 +78,16 @@ 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]
-    keys = resolve_keys(command_steps)
+    emitted_steps = [
+        s
+        for s in ordered
+        if (s.cmd is not None or s.dynamic_target_name is not None) and not s.is_wait
+    ]
+    keys = resolve_keys(emitted_steps)
 
     # Assign integer node indices (dense, in emission order).
     idx_by_id: dict[int, int] = {}
-    for i, s in enumerate(command_steps):
+    for i, s in enumerate(emitted_steps):
         idx_by_id[id(s)] = i
 
     # Track which node indices have a builds_in parent (for default_image).
@@ -107,18 +111,26 @@ def _lower_to_graph(
             pre_wait_indices = []
             continue
 
-        if s.cmd is None:
+        if s.cmd is None and s.dynamic_target_name is None:
             # 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,
-        }
+        step_eval: dict[str, str]
+        if s.dynamic_target_name is not None:
+            step_eval = {
+                "type": "dynamic",
+                "target_name": s.dynamic_target_name,
+            }
+        else:
+            assert s.cmd is not None
+            step_eval = {
+                "type": "cmd",
+                "cmd": s.cmd,
+            }
+        step_dict: dict[str, Any] = {"key": step_key, "eval": step_eval}
         if s.label is not None:
             step_dict["label"] = s.label
         if s.cache is not None:
@@ -147,7 +159,7 @@ def _lower_to_graph(
         # builds_in edge from parent.
         parent_key = _resolved_parent_key(s, keys)
         if parent_key is not None:
-            parent_idx = _find_idx_by_key(parent_key, command_steps, keys, idx_by_id)
+            parent_idx = _find_idx_by_key(parent_key, emitted_steps, keys, idx_by_id)
             edges.append([parent_idx, node_idx, "builds_in"])
             has_builds_in_parent.add(node_idx)
 
@@ -216,7 +228,9 @@ 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.cmd is not None or node.dynamic_target_name is not None
+        ) and not node.is_wait:
             return keys[id(node)]
         node = node.parent
     return None

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

@@ -24,6 +24,8 @@ class Step:
     """
 
     cmd: str | None = None
+    dynamic_target_name: str | None = None
+    """Registered target to evaluate when execution reaches this node."""
     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
@@ -107,7 +109,13 @@ def sh(
         # 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)
+            image
+            if image is not None
+            else (
+                self.image
+                if self.cmd is None and self.dynamic_target_name is None
+                else None
+            )
         )
         return Step(
             cmd=effective_cmd,

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

@@ -49,6 +49,7 @@ def venv() -> hm.Step:
 
 
 _TARGET_CACHE: dict[Callable[..., Any], Any] = {}
+_DYNAMIC_TARGETS_BY_NAME: dict[str, Callable[..., Any]] = {}
 
 
 def clear_target_memo() -> None:
@@ -71,12 +72,45 @@ def clear_target_cache() -> None:
     is wiped via ``clear_target_memo()``.
     """
     _TARGET_CACHE.clear()
+    _DYNAMIC_TARGETS_BY_NAME.clear()
     clear_target_names()
 
 
+def evaluate_dynamic_target(name: str) -> Any:
+    """Evaluate a registered dynamic target for runtime graph expansion."""
+    try:
+        fn = _DYNAMIC_TARGETS_BY_NAME[name]
+    except KeyError as e:
+        raise KeyError(f"hm: dynamic target {name!r} not found") from e
+
+    from ._unwrap import as_leaves
+
+    return as_leaves(call_with_deps(fn))
+
+
+def render_dynamic_target_json(
+    name: str,
+    *,
+    env: dict[str, str] | None = None,
+) -> str:
+    """Evaluate one dynamic target and serialize its concrete graph fragment."""
+    from ._pipeline import pipeline, pipeline_to_json
+
+    clear_target_memo()
+    runtime_env = env if env is not None else {}
+    leaves = evaluate_dynamic_target(name)
+    fragment = pipeline(leaves, env=runtime_env)
+    return pipeline_to_json(
+        fragment,
+        pipeline_slug=name,
+        env=runtime_env,
+    )
+
+
 def target(
     *,
     name: str | None = None,
+    dynamic: bool = False,
 ) -> Callable[[Callable[..., Any]], Callable[[], Any]]:
     """Mark a function as a reusable, memoized pipeline building block.
 
@@ -89,6 +123,8 @@ def target(
         name: Registry key for this target. Defaults to the decorated
             function's name. Override when the name collides with another
             target or a more human-readable key is preferred.
+        dynamic: Defer this target's evaluation until pipeline execution.
+            The initial bake emits a dynamic placeholder referencing its name.
 
     Returns:
         A decorator that registers and memoizes the wrapped function.
@@ -106,11 +142,20 @@ def decorator(fn: Callable[..., Any]) -> Callable[[], Any]:
 
         @wraps(fn)
         def wrapper() -> Any:
+            if dynamic:
+                from ._step import Step
+
+                return Step(
+                    dynamic_target_name=target_name,
+                    key_override=target_name,
+                )
             if fn not in _TARGET_CACHE:
                 _TARGET_CACHE[fn] = call_with_deps(fn)
             return _TARGET_CACHE[fn]
 
         register_named_target(target_name, wrapper)
+        if dynamic:
+            _DYNAMIC_TARGETS_BY_NAME[target_name] = fn
         return wrapper
 
     return decorator

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

@@ -61,7 +61,8 @@ def resolve_pipeline_keys(
         cache = step.get("cache")
         if not cache or cache["policy"] == "none":
             continue
-        cmd = step.get("cmd", "")
+        evaluation = step["eval"]
+        cmd = evaluation["cmd"] if evaluation["type"] == "cmd" else ""
         parent = parent_key_map.get(step["key"])
         parent_resolved = _lookup_parent(parent, resolved)
         policy_res = _resolve_policy(cache, cmd, now, base_path, env)

crates/hm-dsl-engine/harmont-py/tests/test_cmake.py

@@ -8,7 +8,7 @@
 
 
 def _cmds(p: dict) -> list[str]:
-    return [n["step"]["cmd"] for n in p["graph"]["nodes"]]
+    return [n["step"]["eval"]["cmd"] for n in p["graph"]["nodes"]]
 
 
 # ---------------------------------------------------------------------------
@@ -226,7 +226,7 @@ def test_vcpkg_step_has_on_change_cache_policy(self):
         proj = hm.cmake(path="svc", deps="vcpkg")
         p = hm.pipeline([proj.built], default_image="ubuntu:24.04")
         nodes = p["graph"]["nodes"]
-        vcpkg_node = next(n for n in nodes if "bootstrap-vcpkg" in n["step"]["cmd"])
+        vcpkg_node = next(n for n in nodes if "bootstrap-vcpkg" in n["step"]["eval"]["cmd"])
         assert vcpkg_node["step"]["cache"]["policy"] == "on_change"
 
 

crates/hm-dsl-engine/harmont-py/tests/test_e2e_fixtures.py

@@ -160,7 +160,8 @@ def test_e2e_fixture(name: str) -> None:
 
     for node in ir["graph"]["nodes"]:
         assert "key" in node["step"]
-        assert "cmd" in node["step"]
+        assert node["step"]["eval"]["type"] == "cmd"
+        assert "cmd" in node["step"]["eval"]
         assert isinstance(node["env"], dict)
 
     for src, dst, kind in ir["graph"]["edges"]:

crates/hm-dsl-engine/harmont-py/tests/test_elixir.py

@@ -8,12 +8,12 @@
 
 
 def _cmds(p: dict) -> list[str]:
-    return [n["step"]["cmd"] for n in p["graph"]["nodes"]]
+    return [n["step"]["eval"]["cmd"] for n in p["graph"]["nodes"]]
 
 
 def _step_by_substring(p: dict, needle: str) -> dict:
     for n in p["graph"]["nodes"]:
-        if needle in (n["step"].get("cmd") or ""):
+        if needle in (n["step"].get("eval", {}).get("cmd") or ""):
             return n["step"]
     msg = f"no command step containing {needle!r}"
     raise AssertionError(msg)
@@ -56,8 +56,8 @@ def test_elixir_version_in_install_cmd():
     ex = hm.elixir(elixir_version="1.18.3", otp_version="27.3.3")
     p = hm.pipeline([ex.compile()])
     elixir_step = _step_by_substring(p, "elixir-otp")
-    assert "1.18.3" in elixir_step["cmd"]
-    assert "27" in elixir_step["cmd"]
+    assert "1.18.3" in elixir_step["eval"]["cmd"]
+    assert "27" in elixir_step["eval"]["cmd"]
 
 
 def test_elixir_invalid_version_rejected():
@@ -98,7 +98,7 @@ def test_elixir_plt_cached_on_lock():
     assert step.label == ":ex: plt"
     p = hm.pipeline([step])
     plt_ir = next(
-        n["step"] for n in p["graph"]["nodes"] if "dialyzer --plt" in (n["step"].get("cmd") or "")
+        n["step"] for n in p["graph"]["nodes"] if "dialyzer --plt" in (n["step"].get("eval", {}).get("cmd") or "")
     )
     assert plt_ir["cache"]["policy"] == "on_change"
     assert "./mix.lock" in plt_ir["cache"]["paths"]

crates/hm-dsl-engine/harmont-py/tests/test_env.py

@@ -0,0 +1,16 @@
+import harmont as hm
+
+
+def test_env_returns_value(monkeypatch):
+    monkeypatch.setenv("HARMONT_TEST_ENV", "configured")
+    assert hm.env("HARMONT_TEST_ENV") == "configured"
+
+
+def test_env_returns_default_when_unset(monkeypatch):
+    monkeypatch.delenv("HARMONT_TEST_ENV", raising=False)
+    assert hm.env("HARMONT_TEST_ENV", "fallback") == "fallback"
+
+
+def test_env_returns_none_without_default(monkeypatch):
+    monkeypatch.delenv("HARMONT_TEST_ENV", raising=False)
+    assert hm.env("HARMONT_TEST_ENV") is None

crates/hm-dsl-engine/harmont-py/tests/test_envelope.py

@@ -30,7 +30,7 @@ def _graph_edges(definition):
 
 
 def _step_cmds(definition):
-    return [n["step"].get("cmd") for n in _graph_nodes(definition)]
+    return [n["step"].get("eval", {}).get("cmd") for n in _graph_nodes(definition)]
 
 
 def _builds_in_children(definition, parent_key):
@@ -72,7 +72,7 @@ def ci() -> hm.Step:
     assert definition["version"] == "0"
     nodes = _graph_nodes(definition)
     assert len(nodes) == 1
-    assert nodes[0]["step"]["cmd"] == "echo hi"
+    assert nodes[0]["step"]["eval"]["cmd"] == "echo hi"
     assert nodes[0]["step"]["label"] == "hi"
 
 
@@ -107,7 +107,7 @@ def ci() -> hm.Pipeline:
 
     out = json.loads(hm.dump_registry_json())
     p = out["pipelines"][0]
-    cmds = sorted(n["step"]["cmd"] for n in _graph_nodes(p["definition"]))
+    cmds = sorted(n["step"]["eval"]["cmd"] for n in _graph_nodes(p["definition"]))
     assert cmds == ["a", "b"]
 
 
@@ -152,7 +152,7 @@ def ci():
 
     out = json.loads(hm.dump_registry_json())
     nodes = _graph_nodes(out["pipelines"][0]["definition"])
-    cmds = [n["step"].get("cmd") for n in nodes]
+    cmds = [n["step"].get("eval", {}).get("cmd") for n in nodes]
     assert any("go build" in (c or "") for c in cmds)
 
 
@@ -176,11 +176,11 @@ def ci() -> tuple[hm.Step, ...]:
     out = json.loads(hm.dump_registry_json())
     definition = out["pipelines"][0]["definition"]
     nodes = _graph_nodes(definition)
-    apt_nodes = [n for n in nodes if n["step"].get("cmd") == "apt-get update"]
+    apt_nodes = [n for n in nodes if n["step"].get("eval", {}).get("cmd") == "apt-get update"]
     assert len(apt_nodes) == 1  # deduplicated via target memoization
     children = _builds_in_children(definition, apt_nodes[0]["step"]["key"])
     assert len(children) == 2
-    child_cmds = sorted(n["step"]["cmd"] for n in children)
+    child_cmds = sorted(n["step"]["eval"]["cmd"] for n in children)
     assert child_cmds == ["cabal build", "pytest"]
 
 

crates/hm-dsl-engine/harmont-py/tests/test_go.py

@@ -8,12 +8,12 @@
 
 
 def _cmds(p: dict) -> list[str]:
-    return [n["step"]["cmd"] for n in p["graph"]["nodes"]]
+    return [n["step"]["eval"]["cmd"] for n in p["graph"]["nodes"]]
 
 
 def _step_by_substring(p: dict, needle: str) -> dict:
     for n in p["graph"]["nodes"]:
-        if needle in (n["step"].get("cmd") or ""):
+        if needle in (n["step"].get("eval", {}).get("cmd") or ""):
             return n["step"]
     msg = f"no command step containing {needle!r}"
     raise AssertionError(msg)
@@ -50,7 +50,7 @@ def test_go_version_in_install_cmd():
     go = hm.go(path=".", version="1.23.2")
     p = hm.pipeline([go.build()])
     install = _step_by_substring(p, "go.dev/dl/")
-    assert "go1.23.2" in install["cmd"]
+    assert "go1.23.2" in install["eval"]["cmd"]
 
 
 def test_go_invalid_version_rejected():

crates/hm-dsl-engine/harmont-py/tests/test_har_28_example.py

@@ -58,13 +58,13 @@ def ci():
     p = out["pipelines"][0]
     nodes = _graph_nodes(p["definition"])
 
-    cmds = [n["step"].get("cmd") for n in nodes]
+    cmds = [n["step"].get("eval", {}).get("cmd") for n in nodes]
     assert any("pytest -v" in (c or "") for c in cmds)
     assert any("go build" in (c or "") for c in cmds)
     assert any("npm" in (c or "") for c in cmds)
 
     # apt-base used by the venv chain appears exactly once (memoized).
-    apt_update_nodes = [n for n in nodes if n["step"].get("cmd") == "apt-get update"]
+    apt_update_nodes = [n for n in nodes if n["step"].get("eval", {}).get("cmd") == "apt-get update"]
     assert len(apt_update_nodes) == 1
 
 
@@ -75,5 +75,5 @@ def ci():
 
     out = json.loads(hm.dump_registry_json())
     nodes = _graph_nodes(out["pipelines"][0]["definition"])
-    cmds = [n["step"]["cmd"] for n in nodes]
+    cmds = [n["step"]["eval"]["cmd"] for n in nodes]
     assert "cd cidsl/py && pytest -v" in cmds