feat(experimentalist): run in OpenShell with an external Harbor bridge

docker-bake.hcl

@@ -593,6 +593,23 @@ target "nmp-cpu-tasks-docker" {
   platforms  = get_platforms()
 }
 
+# Experimentalist control plane. This target is built on demand by the default
+# OpenShell runtime and remains outside the platform's default image groups.
+target "nmp-experimentalist-docker" {
+  target     = "runtime"
+  context    = "."
+  dockerfile = "plugins/nemo-experimentalist/Dockerfile"
+  contexts = {
+    nmp-python-base = "target:nmp-python-base"
+    nmp-workspace   = "target:nmp-workspace"
+  }
+  cache-to   = maybe_registry_cache_to("nmp-experimentalist")
+  cache-from = maybe_registry_cache_from("nmp-experimentalist")
+  tags       = sha_and_maybe_latest_tags("nmp-experimentalist")
+  output     = image_output()
+  platforms  = get_platforms()
+}
+
 # Python wheel builders (causal-conv1d, mamba-ssm, av, opencv-python-headless).
 # CUDA extensions only ship source on PyPI; av/opencv bundle FFmpeg. Pre-built for
 # amd64 and arm64. Wheels live at /wheels/*.whl inside each image.

packages/nmp_platform_runner/src/nmp/platform_runner/registry.py

@@ -38,6 +38,7 @@
 AVAILABLE_SIDECARS: dict[str, str] = {
     "adapters": "nmp.core.models.sidecars.adapters.main:run",
     "auth-proxy": "nmp.common.auth.workload_proxy.main:run",
+    "clickhouse": "nmp.intake.sidecars.clickhouse:run",
 }
 
 SERVICE_SIDECAR_DEPENDENCIES: dict[str, set[str]] = {

plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/agent.py

@@ -9,6 +9,7 @@
 
 import asyncio
 import logging
+from collections.abc import Callable
 from pathlib import Path
 from typing import Any
 
@@ -25,7 +26,10 @@
     Task,
     TrialResult,
 )
-from nemo_experimentalist_plugin.experimentalist.components.evaluator.models import ResourceRef
+from nemo_experimentalist_plugin.experimentalist.components.evaluator.models import (
+    DependencyRuntimeError,
+    ResourceRef,
+)
 from nemo_experimentalist_plugin.experimentalist.components.tools import GuardedShellTools
 from nemo_experimentalist_plugin.experimentalist.components.trace_analyzer import (
     Diagnostic,
@@ -199,7 +203,11 @@ async def fill_task_template(
            The caller has already copied the template to its durable candidate path;
            edit this directory in place and do not copy or rename it.
         2. Fetch the trace via ``client`` and populate the template's placeholders
-           with values from the trace (instruction, environment config, etc.).
+           with values from the trace. A trusted task may contain
+           ``nemo-task-envelope.json``; in that case edit only its declared
+           ``task_data`` paths. Never edit Dockerfiles, Compose files, image
+           references, mounts, environment-variable bindings, network policy,
+           resource policy, or verifier files while filling a task.
            Leave unfillable placeholders as-is.
         3. Keep ``task.toml`` parseable and keep ``[task] name`` in ``org/name``
            format. The caller will deterministically finalize the name and provenance.
@@ -243,6 +251,7 @@ async def run(
         validation_dataset: Dataset,
         *,
         client: AsyncNeMoPlatform,
+        prepare_dataset: Callable[[Dataset], Dataset] | None = None,
     ) -> EvalAuthorResult:
         """Curate an evaluation suite and always close the owned shell session."""
         try:
@@ -253,6 +262,7 @@ async def run(
                 train_dataset=train_dataset,
                 validation_dataset=validation_dataset,
                 client=client,
+                prepare_dataset=prepare_dataset,
             )
         finally:
             await self.shell.close()
@@ -266,6 +276,7 @@ async def _run(
         validation_dataset: Dataset,
         *,
         client: AsyncNeMoPlatform,
+        prepare_dataset: Callable[[Dataset], Dataset] | None = None,
     ) -> EvalAuthorResult:
         """Curate an evaluation suite from an Insight and its production traces.
 
@@ -276,6 +287,7 @@ async def _run(
             train_dataset: The train dataset, returned unchanged.
             validation_dataset: The validation dataset, returned unchanged.
             client: Existing NeMo Platform client used for Intake requests.
+            prepare_dataset: Optional evaluator hook that binds task runtimes.
         """
         resolved_agent = self.experiment_dir / agent_path
         insight_id = insight.id
@@ -299,8 +311,11 @@ async def _run(
             staged_tasks = insight_suite.stage(refs)
             for staged in staged_tasks:
                 await self.fill_task_template(staged.trace_ref, staged.task, client, insight.workspace)
+                insight_suite.validate_fill_mutations(staged)
                 insight_suite.validate(staged)
             materialized_dataset = insight_suite.promote_local(refs, staged_tasks)
+            if prepare_dataset is not None:
+                materialized_dataset = prepare_dataset(materialized_dataset)
         except BaseException:
             insight_suite.discard()
             raise
@@ -332,6 +347,8 @@ async def _run(
         for task, ref, result in zip(tasks, refs, raw_diagnostics, strict=True):
             if isinstance(result, asyncio.CancelledError):
                 raise result
+            if isinstance(result, DependencyRuntimeError):
+                raise result
             if isinstance(result, BaseException):
                 logger.warning("Trace analysis failed for %s: %s", ref, result)
                 analysis_statuses[task.id] = ("failed", str(result))
@@ -343,12 +360,14 @@ async def _run(
 
         self.context["dataset_documentation"] = doc(type(materialized_dataset), inline_depth=1)
         runner_conventions = await self.discover_runner(materialized_dataset)
+        metric_mutation_snapshot = insight_suite.metric_mutation_snapshot()
         summary = await self.author_insight_metrics(
             insight,
             diagnostics,
             materialized_dataset,
             runner_conventions,
         )
+        insight_suite.validate_metric_mutations(metric_mutation_snapshot)
         for repair_attempt in range(self._config.max_validation_repair_attempts + 1):
             try:
                 await materialized_dataset.validate()
@@ -368,6 +387,7 @@ async def _run(
                     runner_conventions,
                     validation_feedback=str(exc),
                 )
+                insight_suite.validate_metric_mutations(metric_mutation_snapshot)
             else:
                 finalized_suite = insight_suite.finalize()
                 return EvalAuthorResult(

plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/materialization.py

@@ -25,6 +25,8 @@
 _CONTENT_HASH_SCHEMA_VERSION = 1
 _METRIC_CONTRACT_VERSION = 1
 _SLUG_RE = re.compile(r"[^a-z0-9]+")
+_ENVELOPE_DESCRIPTOR_FILENAME = ".nemo-trusted-harbor-envelope.json"
+_ENVELOPE_POLICY_FILENAME = "nemo-task-envelope.json"
 
 
 def _slug(value: str, *, fallback: str, max_length: int = 48) -> str:
@@ -52,6 +54,89 @@ def _file_hashes(root: Path) -> dict[str, str]:
     }
 
 
+def _contains(parent: str, child: str) -> bool:
+    parent_path = Path(parent)
+    child_path = Path(child)
+    return child_path == parent_path or parent_path in child_path.parents
+
+
+@dataclass(frozen=True, slots=True)
+class _MutationPolicy:
+    task_data_paths: tuple[str, ...]
+    verifier_paths: tuple[str, ...]
+
+
+def _mutation_policy(task_dir: Path) -> _MutationPolicy | None:
+    """Load the trusted-envelope policy without importing Experimentalist helpers."""
+    descriptor = task_dir / _ENVELOPE_DESCRIPTOR_FILENAME
+    if not descriptor.is_file():
+        return None
+    policy_path = task_dir / _ENVELOPE_POLICY_FILENAME
+    if not policy_path.is_file():
+        raise ValueError(f"Trusted Eval Author task is missing {_ENVELOPE_POLICY_FILENAME}: {task_dir}")
+    try:
+        payload = json.loads(policy_path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+        raise ValueError(f"Invalid trusted task mutation policy {policy_path}: {exc}") from exc
+    if payload.get("schema_version") != 1:
+        raise ValueError(f"Unsupported trusted task mutation policy version in {policy_path}")
+
+    def paths_from_values(raw: object, key: str) -> tuple[str, ...]:
+        if not isinstance(raw, list):
+            raise ValueError(f"Trusted task mutation policy {key} must be a list of paths: {policy_path}")
+        normalized: list[str] = []
+        for item in raw:
+            if not isinstance(item, str) or not item:
+                raise ValueError(f"Trusted task mutation policy {key} must be a list of paths: {policy_path}")
+            path = Path(item)
+            if path.is_absolute() or any(part in ("", ".", "..") for part in path.parts):
+                raise ValueError(f"Unsafe trusted task mutation path {item!r}: {policy_path}")
+            normalized.append(path.as_posix())
+        if len(set(normalized)) != len(normalized):
+            raise ValueError(f"Duplicate trusted task mutation paths in {policy_path}")
+        return tuple(normalized)
+
+    def paths(key: str) -> tuple[str, ...]:
+        return paths_from_values(payload.get(key, []), key)
+
+    raw_task_data = payload.get("task_data", [])
+    if not isinstance(raw_task_data, list):
+        raise ValueError(f"Trusted task mutation policy task_data must be a list: {policy_path}")
+    task_data_paths: list[str] = []
+    for slot in raw_task_data:
+        if not isinstance(slot, dict) or not isinstance(slot.get("path"), str):
+            raise ValueError(f"Trusted task mutation policy task_data contains an invalid slot: {policy_path}")
+        task_data_paths.extend(paths_from_values([slot["path"]], "task_data"))
+    return _MutationPolicy(
+        task_data_paths=tuple(task_data_paths),
+        verifier_paths=paths("verifier_paths"),
+    )
+
+
+def _immutable_snapshot(task_dir: Path, *, mutable_paths: tuple[str, ...]) -> dict[str, str]:
+    return {
+        relative: digest
+        for relative, digest in _file_hashes(task_dir).items()
+        if not any(_contains(mutable, relative) for mutable in mutable_paths)
+    }
+
+
+def _assert_snapshot(
+    task_dir: Path,
+    *,
+    mutable_paths: tuple[str, ...],
+    expected: dict[str, str],
+    phase: str,
+) -> None:
+    actual = _immutable_snapshot(task_dir, mutable_paths=mutable_paths)
+    if actual == expected:
+        return
+    changed = sorted(path for path in set(expected) | set(actual) if expected.get(path) != actual.get(path))
+    raise ValueError(
+        f"Eval Author {phase} modified trusted task paths outside its envelope slots: " + ", ".join(changed[:20])
+    )
+
+
 def _verifier_dir(task_dir: Path) -> Path:
     config = tomllib.loads((task_dir / "task.toml").read_text(encoding="utf-8"))
     verifier = config.get("verifier")
@@ -144,6 +229,8 @@ class StagedInsightTask:
     slug: str
     path: Path
     task: Task
+    mutation_policy: _MutationPolicy | None
+    immutable_before_fill: dict[str, str] | None
 
 
 @dataclass(frozen=True, slots=True)
@@ -200,9 +287,61 @@ def stage(self, trace_refs: list[str]) -> list[StagedInsightTask]:
             task_dir = self._candidate_suite / slug
             shutil.copytree(self.template_dir, task_dir)
             task = list(HarborDataset.from_path(task_dir).list_tasks())[0]
-            staged.append(StagedInsightTask(index=index, trace_ref=trace_ref, slug=slug, path=task_dir, task=task))
+            policy = _mutation_policy(task_dir)
+            staged.append(
+                StagedInsightTask(
+                    index=index,
+                    trace_ref=trace_ref,
+                    slug=slug,
+                    path=task_dir,
+                    task=task,
+                    mutation_policy=policy,
+                    immutable_before_fill=(
+                        _immutable_snapshot(task_dir, mutable_paths=policy.task_data_paths)
+                        if policy is not None
+                        else None
+                    ),
+                )
+            )
         return staged
 
+    def validate_fill_mutations(self, staged: StagedInsightTask) -> None:
+        """Reject coding-agent edits outside declared trace-data slots."""
+        if staged.mutation_policy is None or staged.immutable_before_fill is None:
+            return
+        _assert_snapshot(
+            staged.path,
+            mutable_paths=staged.mutation_policy.task_data_paths,
+            expected=staged.immutable_before_fill,
+            phase="template filling",
+        )
+
+    def metric_mutation_snapshot(self) -> dict[str, tuple[tuple[str, ...], dict[str, str]]]:
+        """Capture non-verifier content before metric authoring begins."""
+        snapshot: dict[str, tuple[tuple[str, ...], dict[str, str]]] = {}
+        for task_dir in sorted(path for path in self.suite_dir.iterdir() if path.is_dir()):
+            policy = _mutation_policy(task_dir)
+            if policy is None:
+                continue
+            snapshot[task_dir.name] = (
+                policy.verifier_paths,
+                _immutable_snapshot(task_dir, mutable_paths=policy.verifier_paths),
+            )
+        return snapshot
+
+    def validate_metric_mutations(
+        self,
+        snapshot: dict[str, tuple[tuple[str, ...], dict[str, str]]],
+    ) -> None:
+        """Reject metric-author edits outside declared verifier slots."""
+        for task_name, (mutable_paths, expected) in snapshot.items():
+            _assert_snapshot(
+                self.suite_dir / task_name,
+                mutable_paths=mutable_paths,
+                expected=expected,
+                phase="metric authoring",
+            )
+
     def discard(self) -> None:
         """Remove an unpromoted candidate suite and reset its staging state."""
         if self._candidate_root is not None and self._candidate_root.exists():