NVIDIA-NeMo · nabinchha · Sep 10, 2026 · Sep 10, 2026 · Sep 10, 2026 · Sep 10, 2026
@@ -272,6 +272,8 @@ def _install_overlay(
 
 def inspect_distributions(path: Path | None) -> tuple[InstalledDistribution, ...]:
     """Return an exact immutable distribution inventory."""
+    if path is not None:
+        importlib.invalidate_caches()
     distributions = (
         importlib.metadata.distributions() if path is None else importlib.metadata.distributions(path=[path.as_posix()])
     )

@@ -0,0 +1,36 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Bootstrap-safe process boundary for the plugin-aware client worker.
+
+This module must remain free of Data Designer configuration, interface, plugin,
+and worker imports. The child process activates its verified dependency overlay
+before importing any of those modules.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass, field
+
+ProcessExecutor = Callable[[tuple[str, ...]], int]
+
+
+def execute_process(command: tuple[str, ...]) -> int:
+    """Execute a child process and return its exit status."""
+    return subprocess.run(command, check=False).returncode
+
+
+@dataclass(frozen=True)
+class ClientWorkerProcess:
+    """Launch the plugin-aware client worker in a fresh Python interpreter."""
+
+    executable: str = field(default_factory=lambda: sys.executable)
+    executor: ProcessExecutor = execute_process
+
+    def run(self, arguments: Sequence[str]) -> int:
+        """Run one client-worker operation without inheriting imported modules."""
+        command = (self.executable, "-m", "data_designer.slurm.client.worker", *arguments)
+        return self.executor(command)
@@ -13,7 +13,7 @@
 from pathlib import Path
 
 from data_designer.slurm.client.filesystem import ensure_private_directory, replace_private_text
-from data_designer.slurm.client.worker import main as client_worker_main
+from data_designer.slurm.client.process import ClientWorkerProcess
 from data_designer.slurm.runtime.bootstrap import build_runtime_manifest
 from data_designer.slurm.runtime.context import load_allocation_context
 from data_designer.slurm.runtime.errors import SlurmRuntimeError, SlurmRuntimeErrorCode
@@ -145,8 +145,14 @@ def _ready(arguments: argparse.Namespace, environment: Mapping[str, str]) -> Non
     )
 
 
-def _client(arguments: argparse.Namespace, environment: Mapping[str, str]) -> None:
+def _client(
+    arguments: argparse.Namespace,
+    environment: Mapping[str, str],
+    *,
+    client_worker: ClientWorkerProcess | None = None,
+) -> None:
     context, writer = _load_context(arguments, environment)
+    worker_process = client_worker or ClientWorkerProcess()
     generation_started_at = _now(context.attempt, _load_optional_readiness(context, writer))
     resume_mode = (
         context.plan.invocation.authored.resume
@@ -158,7 +164,7 @@ def _client(arguments: argparse.Namespace, environment: Mapping[str, str]) -> No
         context.attempt.attempt_id,
         resume_mode,
     ):
-        return_code = client_worker_main(
+        return_code = worker_process.run(
             (
                 "run",
                 "--plan",

@@ -6,6 +6,7 @@
 import hashlib
 import importlib
 import json
+import os
 import subprocess
 import sys
 from pathlib import Path
@@ -144,6 +145,20 @@ def distributions(**kwargs: object) -> tuple[()]:
     assert calls == [{}]
 
 
+def test_inspect_distributions_refreshes_overlay_with_unchanged_directory_mtime(tmp_path: Path) -> None:
+    original_stat = tmp_path.stat()
+    assert inspect_distributions(tmp_path) == ()
+    distribution = tmp_path / "cache_probe-1.0.0.dist-info"
+    distribution.mkdir()
+    (distribution / "METADATA").write_text(
+        "Metadata-Version: 2.1\nName: cache-probe\nVersion: 1.0.0\n",
+        encoding="utf-8",
+    )
+    os.utime(tmp_path, ns=(original_stat.st_atime_ns, original_stat.st_mtime_ns))
+
+    assert inspect_distributions(tmp_path) == (InstalledDistribution(name="cache-probe", version="1.0.0"),)
+
+
 def test_inspect_distributions_rejects_unhashed_direct_url(monkeypatch: pytest.MonkeyPatch) -> None:
     distribution = Mock(
         metadata={"Name": "example"},

@@ -0,0 +1,26 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from data_designer.slurm.client.process import ClientWorkerProcess
+
+
+def test_client_worker_process_uses_a_fresh_python_interpreter() -> None:
+    commands: list[tuple[str, ...]] = []
+    process = ClientWorkerProcess(
+        executable="/client/python",
+        executor=lambda command: commands.append(command) or 17,
+    )
+
+    assert process.run(("run", "--plan", "/workspace/resolved-plan.json")) == 17
+    assert commands == [
+        (
+            "/client/python",
+            "-m",
+            "data_designer.slurm.client.worker",
+            "run",
+            "--plan",
+            "/workspace/resolved-plan.json",
+        )
+    ]
@@ -500,7 +500,14 @@ def test_preflight_rejects_plugin_secondary_model_alias(
 ) -> None:
     payload = client_worker_case.plan.model_dump(mode="json")
     builder = payload["builder"]["inline"]
-    builder["data_designer"]["columns"] = [{"name": "custom", "column_type": "fake-slurm-column"}]
+    builder["data_designer"]["columns"] = [
+        {
+            "name": "custom",
+            "column_type": "fake-slurm-column",
+            "model_alias": "generator",
+            "judge_model_alias": "missing",
+        }
+    ]
     payload["builder"]["content_sha256"] = compute_serialized_json_sha256(builder)
     lock_payload = client_worker_case.lock.model_dump(mode="json")
     wheel_path = (

@@ -0,0 +1,17 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from typing import Any
+
+from data_designer.engine.column_generators.generators.base import ColumnGeneratorCellByCell
+from fake_data_designer_plugin.plugin import FakePluginConfig
+
+
+class FakePluginImplementation(ColumnGeneratorCellByCell[FakePluginConfig]):
+    """Add a deterministic marker through the real generator contract."""
+
+    def generate(self, data: dict[str, Any]) -> dict[str, Any]:
+        """Return the complete record with the plugin-owned column."""
+        return {**data, self.config.name: "plugin-marker"}
@@ -13,8 +13,8 @@ class FakePluginConfig(SingleColumnConfig):
     """Minimal custom column configuration."""
 
     column_type: Literal["fake-slurm-column"] = "fake-slurm-column"
-    model_alias: str = "generator"
-    judge_model_alias: str = "judge"
+    model_alias: str | None = None
+    judge_model_alias: str | None = None
 
     @property
     def required_columns(self) -> list[str]:
@@ -25,19 +25,11 @@ def side_effect_columns(self) -> list[str]:
         return []
 
     def get_model_aliases(self) -> list[str]:
-        return [self.model_alias, self.judge_model_alias]
-
-
-class FakePluginImplementation:
-    """Minimal loadable implementation for entry-point verification."""
-
-    def generate(self, data: dict[str, object]) -> dict[str, object]:
-        """Return the provided record unchanged."""
-        return data
+        return [alias for alias in (self.model_alias, self.judge_model_alias) if alias is not None]
 
 
 plugin = Plugin(
     config_qualified_name="fake_data_designer_plugin.plugin.FakePluginConfig",
-    impl_qualified_name="fake_data_designer_plugin.plugin.FakePluginImplementation",
+    impl_qualified_name="fake_data_designer_plugin.implementation.FakePluginImplementation",
     plugin_type=PluginType.COLUMN_GENERATOR,
 )