fix(dsl): port corepack PM version pin to Python DSL; fail-fast on legacy pipeline(step) form (#134)

Author: markovejnovic

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

@@ -82,9 +82,28 @@ def my_pipeline() -> Step: ...
 
     Returns:
         A v0 IR ``dict`` in factory form, or a decorator in decorator form.
+
+    Raises:
+        TypeError: When called with the legacy variadic ``Step`` form
+            (``pipeline(step)`` / ``pipeline(a, b)``). The factory now takes
+            a single list of leaves.
     """
     if args and isinstance(args[0], (list, tuple)):
         return _pipeline_factory(args[0], **kwargs)
+    # Legacy form: leaves passed as positional Step args (pre-CLI-9
+    # `pipeline(step)` / `pipeline(a, b)`). Without this guard the call would
+    # fall through to the decorator and fail far downstream with a cryptic
+    # AttributeError. Fail fast with the migration hint instead.
+    if args and all(isinstance(a, Step) for a in args):
+        msg = (
+            "hm.pipeline() takes a single list of leaves, not variadic Step "
+            "arguments\n"
+            f"  observed: {len(args)} positional Step "
+            f"argument{'s' if len(args) != 1 else ''}\n"
+            "  → wrap the leaves in a list, e.g. "
+            "hm.pipeline([step]) or hm.pipeline([a, b])"
+        )
+        raise TypeError(msg)
     return _decorator.pipeline(*args, **kwargs)
 
 

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

@@ -16,11 +16,13 @@
 class DetectedToolchain:
     runtime: Runtime | None = None
     pm: Pm | None = None
+    pm_version: str | None = None
 
 
 def detect_from_package_json(package_json: dict) -> DetectedToolchain:
     runtime: Runtime | None = None
     pm: Pm | None = None
+    pm_version: str | None = None
 
     engines = package_json.get("engines")
     if isinstance(engines, dict):
@@ -35,23 +37,25 @@ def detect_from_package_json(package_json: dict) -> DetectedToolchain:
     if pm is None:
         pm_field = package_json.get("packageManager")
         if isinstance(pm_field, str):
-            name = pm_field.split("@")[0]
+            parts = pm_field.split("@")
+            name = parts[0]
+            ver = parts[1] if len(parts) > 1 else ""
             if name == "pnpm":
                 pm = "pnpm"
             elif name == "bun":
                 pm = "bun"
             elif name == "npm":
                 pm = "npm"
             elif name == "yarn":
-                parts = pm_field.split("@")
-                ver = parts[1] if len(parts) > 1 else ""
                 try:
                     major = int(ver.split(".")[0])
                 except (ValueError, IndexError):
                     major = 1
                 pm = "yarn-berry" if major >= 2 else "yarn-classic"
+            if pm is not None and ver:
+                pm_version = ver
 
-    return DetectedToolchain(runtime=runtime, pm=pm)
+    return DetectedToolchain(runtime=runtime, pm=pm, pm_version=pm_version)
 
 
 def detect_from_lockfiles(files: list[str]) -> DetectedToolchain:
@@ -86,4 +90,5 @@ def detect(path: str) -> DetectedToolchain:
     return DetectedToolchain(
         runtime=from_pkg.runtime if from_pkg.runtime is not None else from_lock.runtime,
         pm=from_pkg.pm if from_pkg.pm is not None else from_lock.pm,
+        pm_version=from_pkg.pm_version,
     )

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

@@ -69,21 +69,32 @@
 }
 
 
-def _pm_bootstrap(pm: PackageManager, runtime: Runtime) -> str | None:
+def _pm_bootstrap(pm: PackageManager, runtime: Runtime, version: str | None = None) -> str | None:
     """Command to bring ``pm`` onto the runtime image, or ``None`` when the PM
     already ships with the runtime (npm with node, bun/deno with their own
-    runtime)."""
+    runtime).
+
+    When ``version`` is set (from the project's ``packageManager`` field),
+    corepack-managed PMs pin that exact version with ``corepack install -g``."""
     if pm == "npm":
         return None  # bundled with node
     if pm == "bun":
         return None if runtime == "bun" else bun_install_cmd()
     if pm == "deno":
         return None  # bundled with deno
     if pm == "pnpm":
-        return "corepack enable pnpm"
+        return (
+            f"corepack enable pnpm && corepack install -g pnpm@{version}"
+            if version is not None
+            else "corepack enable pnpm"
+        )
     # corepack resolves the exact yarn from the `packageManager` field; its
     # bundled default is classic 1.x, which suits yarn-classic.
-    return "corepack enable"
+    return (
+        f"corepack enable yarn && corepack install -g yarn@{version}"
+        if version is not None
+        else "corepack enable"
+    )
 
 
 def _validate_version(runtime: Runtime, version: str) -> None:
@@ -171,6 +182,7 @@ def _make_js(
 
     # --- Node / Bun runtime ---
     detected_pm = detected.pm if detected else None
+    pm_version = detected.pm_version if detected else None
     resolved_pm: PackageManager = (
         pm
         if pm is not None
@@ -205,14 +217,22 @@ def _make_js(
     )
 
     # Layer the package manager onto the runtime image when it isn't bundled.
-    bootstrap = _pm_bootstrap(resolved_pm, runtime)
+    # When a pinned version is detected, the bootstrap step's snapshot must be
+    # rebuilt whenever package.json (hence the pin) changes; otherwise it is a
+    # deterministic, forever-cacheable install.
+    bootstrap = _pm_bootstrap(resolved_pm, runtime, pm_version)
+    bootstrap_cache = (
+        CacheOnChange(paths=(f"{path}/package.json",))
+        if pm_version is not None
+        else CacheForever(env_keys=())
+    )
     pm_ready = (
         runtime_installed
         if bootstrap is None
         else runtime_installed.sh(
             bootstrap,
             label=f":{lang_tag}: {resolved_pm}",
-            cache=CacheForever(env_keys=()),
+            cache=bootstrap_cache,
         )
     )
 

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

@@ -30,23 +30,31 @@ def test_engines_deno(self) -> None:
 
     def test_package_manager_pnpm(self) -> None:
         pkg = {"packageManager": "pnpm@8.15.4"}
-        assert detect_from_package_json(pkg) == DetectedToolchain(pm="pnpm")
+        assert detect_from_package_json(pkg) == DetectedToolchain(pm="pnpm", pm_version="8.15.4")
 
     def test_package_manager_bun(self) -> None:
         pkg = {"packageManager": "bun@1.1.0"}
-        assert detect_from_package_json(pkg) == DetectedToolchain(pm="bun")
+        assert detect_from_package_json(pkg) == DetectedToolchain(pm="bun", pm_version="1.1.0")
 
     def test_package_manager_npm(self) -> None:
         pkg = {"packageManager": "npm@10.2.4"}
-        assert detect_from_package_json(pkg) == DetectedToolchain(pm="npm")
+        assert detect_from_package_json(pkg) == DetectedToolchain(pm="npm", pm_version="10.2.4")
 
     def test_package_manager_yarn_classic(self) -> None:
         pkg = {"packageManager": "yarn@1.22.22"}
-        assert detect_from_package_json(pkg) == DetectedToolchain(pm="yarn-classic")
+        assert detect_from_package_json(pkg) == DetectedToolchain(
+            pm="yarn-classic", pm_version="1.22.22"
+        )
 
     def test_package_manager_yarn_berry(self) -> None:
         pkg = {"packageManager": "yarn@4.0.0"}
-        assert detect_from_package_json(pkg) == DetectedToolchain(pm="yarn-berry")
+        assert detect_from_package_json(pkg) == DetectedToolchain(
+            pm="yarn-berry", pm_version="4.0.0"
+        )
+
+    def test_package_manager_without_version_omits_pm_version(self) -> None:
+        pkg = {"packageManager": "pnpm"}
+        assert detect_from_package_json(pkg) == DetectedToolchain(pm="pnpm", pm_version=None)
 
     def test_ignores_unknown_package_manager(self) -> None:
         pkg = {"packageManager": "unknown@1.0"}
@@ -60,7 +68,7 @@ def test_engines_bun_overrides_package_manager(self) -> None:
     def test_engines_node_plus_package_manager_pnpm(self) -> None:
         pkg = {"engines": {"node": ">=18"}, "packageManager": "pnpm@8"}
         result = detect_from_package_json(pkg)
-        assert result == DetectedToolchain(runtime="node", pm="pnpm")
+        assert result == DetectedToolchain(runtime="node", pm="pnpm", pm_version="8")
 
 
 class TestDetectFromLockfiles:
@@ -130,4 +138,12 @@ def test_yarn_berry_from_package_manager(self, tmp_path) -> None:
         pkg = json.dumps({"packageManager": "yarn@4.5.0"})
         (tmp_path / "package.json").write_text(pkg)
         (tmp_path / "yarn.lock").touch()
-        assert detect(str(tmp_path)) == DetectedToolchain(pm="yarn-berry")
+        assert detect(str(tmp_path)) == DetectedToolchain(pm="yarn-berry", pm_version="4.5.0")
+
+    def test_pm_version_propagates_through_detect(self, tmp_path) -> None:
+        pkg = json.dumps({"packageManager": "pnpm@10.33.0"})
+        (tmp_path / "package.json").write_text(pkg)
+        (tmp_path / "pnpm-lock.yaml").touch()
+        result = detect(str(tmp_path))
+        assert result.pm == "pnpm"
+        assert result.pm_version == "10.33.0"

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

@@ -344,6 +344,54 @@ def test_skips_detection_when_pm_set(self, tmp_path) -> None:
         assert "pnpm install --frozen-lockfile" in p.install().cmd
 
 
+# ---------------------------------------------------------------------------
+# Corepack version pin (parity with the TypeScript DSL)
+# ---------------------------------------------------------------------------
+
+
+class TestCorepackPin:
+    def test_pnpm_command_includes_version(self, tmp_path) -> None:
+        (tmp_path / "package.json").write_text(json.dumps({"packageManager": "pnpm@10.33.0"}))
+        (tmp_path / "pnpm-lock.yaml").touch()
+        deps = js.project(path=str(tmp_path)).install()
+        corepack = deps.parent
+        assert corepack.cmd == "corepack enable pnpm && corepack install -g pnpm@10.33.0"
+
+    def test_yarn_berry_command_includes_version(self, tmp_path) -> None:
+        (tmp_path / "package.json").write_text(json.dumps({"packageManager": "yarn@4.5.0"}))
+        (tmp_path / "yarn.lock").touch()
+        deps = js.project(path=str(tmp_path)).install()
+        corepack = deps.parent
+        assert corepack.cmd == "corepack enable yarn && corepack install -g yarn@4.5.0"
+
+    def test_command_has_no_version_when_field_absent(self, tmp_path) -> None:
+        (tmp_path / "package.json").write_text("{}")
+        (tmp_path / "pnpm-lock.yaml").touch()
+        deps = js.project(path=str(tmp_path)).install()
+        corepack = deps.parent
+        assert corepack.cmd == "corepack enable pnpm"
+
+    def test_explicit_pm_omits_version(self) -> None:
+        # An explicit pm option skips detection entirely, so no pin is applied.
+        deps = js.project(pm="pnpm").install()
+        corepack = deps.parent
+        assert corepack.cmd == "corepack enable pnpm"
+
+    def test_cache_watches_package_json_when_pinned(self, tmp_path) -> None:
+        (tmp_path / "package.json").write_text(json.dumps({"packageManager": "pnpm@10.33.0"}))
+        (tmp_path / "pnpm-lock.yaml").touch()
+        deps = js.project(path=str(tmp_path)).install()
+        corepack = deps.parent
+        assert corepack.cache.paths == (f"{tmp_path}/package.json",)
+
+    def test_cache_is_forever_when_no_field(self, tmp_path) -> None:
+        (tmp_path / "package.json").write_text("{}")
+        (tmp_path / "pnpm-lock.yaml").touch()
+        deps = js.project(path=str(tmp_path)).install()
+        corepack = deps.parent
+        assert corepack.cache == hm.CacheForever(env_keys=())
+
+
 # ---------------------------------------------------------------------------
 # ts alias
 # ---------------------------------------------------------------------------

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

@@ -25,6 +25,23 @@ def test_pipeline_factory_rejects_no_leaves():
         _factory([])
 
 
+def test_pipeline_rejects_legacy_single_step_form():
+    # Pre-CLI-9 `pipeline(step)` must fail fast with the migration hint, not
+    # silently route to the @hm.pipeline decorator and blow up downstream.
+    step = scratch().sh("echo", label="echo")
+    with pytest.raises(TypeError, match="single list of leaves") as exc:
+        pipeline(step)
+    assert "hm.pipeline([step])" in str(exc.value)
+
+
+def test_pipeline_rejects_legacy_variadic_step_form():
+    a = scratch().sh("a", label="a")
+    b = scratch().sh("b", label="b")
+    with pytest.raises(TypeError, match="single list of leaves") as exc:
+        pipeline(a, b)
+    assert "hm.pipeline([a, b])" in str(exc.value)
+
+
 def test_pipeline_default_image_lowers_to_dict():
     p = pipeline(
         [scratch().sh("echo", label="a", image="ubuntu:24.04")],