fix(experimentalist): read a dataset as the directory of tasks Harbor evaluates

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

@@ -255,7 +255,7 @@ def stage(self, trace_refs: list[str]) -> list[StagedInsightTask]:
             slug = self.task_slug(index, trace_ref)
             task_dir = self._candidate_suite / slug
             shutil.copytree(self.template_dir, task_dir)
-            task = list(HarborDataset.from_path(task_dir).list_tasks())[0]
+            task = list(HarborDataset.from_path(task_dir, single_task=True).list_tasks())[0]
             staged.append(StagedInsightTask(index=index, trace_ref=trace_ref, slug=slug, path=task_dir, task=task))
         return staged
 

plugins/nemo-experimentalist/examples/smoke-agent/README.md

@@ -169,8 +169,8 @@ schema has no dataset field at all.
 `full.yaml` is the only one that exercises the evolutionary machinery — survivors
 carried between rounds, ranking over more than two candidates, and the
 convergence check. It runs against `dataset/groups/_all`, which is **generated
-and gitignored**; build it first, or the run loads zero tasks and reports
-`No tasks matched the filter(s)` rather than erroring:
+and gitignored**; build it first, or the run stops on the unbuilt dataset before
+it evaluates anything:
 
 ```bash
 sbx exec --workdir "$repo" nemo-experimentalist bash -lc \

plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/cli.py

@@ -397,10 +397,11 @@ def doctor(
                             hint="fix optimizer.yaml or its referenced agent_spec/experiment_config",
                         )
                     )
+            insight_ref = effective_insight.ref if effective_insight is not None else None
             results = check_profile(profile_obj, profile_error) + env_results + plan_results + insight_results
             results += check_environment(
                 profile=profile_obj,
-                insight=effective_insight.ref if effective_insight is not None else None,
+                insight=insight_ref,
                 insight_id=effective_insight.selector if effective_insight is not None else None,
                 base_url=base_url_resolved,
                 probes=_PREFLIGHT_PROBES,
@@ -417,7 +418,7 @@ def doctor(
                         require_template=plan.insight is not None,
                         probes=_PREFLIGHT_PROBES,
                     )
-                results += check_datasets(profile_obj)
+                results += check_datasets(profile_obj, require_tasks=insight_ref is None)
             typer.echo(format_report(results))
             if required_failures(results):
                 raise typer.Exit(code=1)

plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/dataset_layout.py

@@ -0,0 +1,35 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""The dataset directory shape Harbor evaluates: a directory of task directories.
+
+Harbor's ``DatasetConfig`` enumerates the *children* of a dataset path and keeps
+the ones that are task directories. It never reads the dataset path itself as a
+task, so a dataset whose tasks Harbor cannot enumerate is rejected by its job
+config rather than by anything we validate first.
+
+Preflight reads this contract as well and cannot import ``harbor`` — whether
+harbor is importable is one of its own checks — so the rule lives here, in a
+module that imports only the standard library.
+"""
+
+from pathlib import Path
+
+TASK_CONFIG_FILENAME = "task.toml"
+_TASK_TEMPLATE_DIRNAME = "task_template"
+
+
+def is_task_dir(path: Path) -> bool:
+    """Return whether *path* is a Harbor task directory."""
+    return path.is_dir() and (path / TASK_CONFIG_FILENAME).exists()
+
+
+def find_task_dirs(dataset_path: Path) -> list[Path]:
+    """Return the task directories *dataset_path* holds, in a stable order.
+
+    A ``task_template`` child is the shape generated tasks are cut from, not a
+    task of the dataset, so it is left out.
+    """
+    return sorted(
+        child for child in dataset_path.iterdir() if child.name != _TASK_TEMPLATE_DIRNAME and is_task_dir(child)
+    )

plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/factory.py

@@ -64,7 +64,7 @@ def build_task_template(self, evaluator_type: EvaluatorType, template_ref: Datas
         Returns:
             Task: The built task.
         """
-        tasks = list(self.build_dataset(evaluator_type, template_ref).list_tasks())
+        tasks = list(self.build_dataset(evaluator_type, template_ref, single_task=True).list_tasks())
         if len(tasks) != 1:
             raise ValueError(f"Task template must contain exactly one {evaluator_type} task; found {len(tasks)}")
         return tasks[0]

plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py

@@ -52,6 +52,10 @@
     EvaluatorConfig,
     EvaluatorType,
 )
+from nemo_experimentalist_plugin.experimentalist.components.evaluator.dataset_layout import (
+    find_task_dirs,
+    is_task_dir,
+)
 from nemo_experimentalist_plugin.experimentalist.components.evaluator.entrypoint import (
     DEFAULT_AGENT_IMPORT_PATH,
     split_import_path,
@@ -984,18 +988,30 @@ def from_path(
         *,
         dataset_id: str | None = None,
         allow_empty: bool = False,
+        single_task: bool = False,
         **_ignored_options: Any,
     ) -> HarborDataset:
-        """Build a Harbor dataset from a local Harbor task collection."""
+        """Build a Harbor dataset from a local Harbor task collection.
+
+        A dataset holds task directories, which is the only shape Harbor's job
+        config enumerates. ``single_task`` additionally reads *dataset_path*
+        itself as one task — the shape of a task template, which is a task
+        rather than a collection of them.
+        """
         dataset_path = dataset_path.expanduser().resolve()
         if not dataset_path.exists():
             raise FileNotFoundError(f"Harbor dataset path not found: {dataset_path}")
         if not dataset_path.is_dir():
             raise ValueError(f"Harbor dataset path is not a directory: {dataset_path}")
 
-        task_dirs = cls._find_task_dirs(dataset_path)
+        task_dirs = [dataset_path] if single_task and is_task_dir(dataset_path) else find_task_dirs(dataset_path)
         if not task_dirs and not allow_empty:
-            raise ValueError(f"Harbor dataset path contains no Harbor task directories: {dataset_path}")
+            detail = (
+                " (it is itself a task directory; point the dataset at the directory holding it)"
+                if is_task_dir(dataset_path)
+                else ""
+            )
+            raise ValueError(f"Harbor dataset path contains no Harbor task directories: {dataset_path}{detail}")
 
         tasks = [cls._from_task_dir(task_dir) for task_dir in task_dirs]
         return cls(
@@ -1007,16 +1023,6 @@ def from_path(
             tasks=tasks,
         )
 
-    @staticmethod
-    def _find_task_dirs(dataset_path: Path) -> list[Path]:
-        if dataset_path.is_dir() and (dataset_path / "task.toml").exists():
-            return [dataset_path]
-        return sorted(
-            path
-            for path in dataset_path.iterdir()
-            if path.is_dir() and path.name != "task_template" and (path / "task.toml").exists()
-        )
-
     @classmethod
     def _from_task_dir(cls, task_dir: Path) -> Task:
         config = tomllib.loads((task_dir / "task.toml").read_text(encoding="utf-8"))

plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/preflight.py

@@ -15,6 +15,11 @@
 from pathlib import Path
 
 import httpx
+from nemo_experimentalist_plugin.experimentalist.components.evaluator.dataset_layout import (
+    TASK_CONFIG_FILENAME,
+    find_task_dirs,
+    is_task_dir,
+)
 from nemo_experimentalist_plugin.experimentalist.components.evaluator.entrypoint import (
     DEFAULT_AGENT_IMPORT_PATH,
     find_entrypoint_module,
@@ -231,10 +236,16 @@ def check_artifacts(
     return results
 
 
-def check_datasets(profile: AgentProfile) -> list[CheckResult]:
+def check_datasets(profile: AgentProfile, *, require_tasks: bool = True) -> list[CheckResult]:
     """Classify and validate the profile's train/validation dataset refs
     (path-exists / registry-ref pass). Doctor-only: the experiment flow
-    resolves its datasets, which proves they exist."""
+    resolves its datasets, which proves they exist.
+
+    Existing is not enough to evaluate: Harbor reads the task directories a
+    dataset holds, so a directory holding none fails the run. ``require_tasks``
+    is off for an insight run, where Eval Author generates the tasks and both
+    splits legitimately start empty.
+    """
     results: list[CheckResult] = []
     for label, value in (("train", profile.datasets.train), ("validation", profile.datasets.validation)):
         try:
@@ -247,21 +258,7 @@ def check_datasets(profile: AgentProfile) -> list[CheckResult]:
             )
             continue
         if kind == "path":
-            path = resolve_profile_path(value, profile.profile_dir)
-            results.append(
-                make_check_result(
-                    f"dataset-{label}",
-                    "artifacts",
-                    path.is_dir(),
-                    "required",
-                    f"{label} dataset at {path}",
-                    (
-                        f"{label} dataset path is not a directory: {path}"
-                        if path.exists()
-                        else f"{label} dataset path missing: {path}"
-                    ),
-                )
-            )
+            results.append(_check_dataset_path(label, resolve_profile_path(value, profile.profile_dir), require_tasks))
         else:
             results.append(
                 CheckResult(
@@ -275,6 +272,36 @@ def check_datasets(profile: AgentProfile) -> list[CheckResult]:
     return results
 
 
+def _check_dataset_path(label: str, path: Path, require_tasks: bool) -> CheckResult:
+    """Check one resolved local dataset directory for the tasks a run evaluates."""
+    if not path.is_dir():
+        return CheckResult(
+            name=f"dataset-{label}",
+            group="artifacts",
+            status="fail",
+            severity="required",
+            message=(
+                f"{label} dataset path is not a directory: {path}"
+                if path.exists()
+                else f"{label} dataset path missing: {path}"
+            ),
+        )
+    has_tasks = not require_tasks or bool(find_task_dirs(path))
+    return make_check_result(
+        f"dataset-{label}",
+        "artifacts",
+        has_tasks,
+        "required",
+        f"{label} dataset at {path}",
+        f"{label} dataset holds no task directories: {path}",
+        hint=(
+            f"{path} is itself a task directory; point {label} at the directory holding it"
+            if is_task_dir(path)
+            else f"each task is a subdirectory with a {TASK_CONFIG_FILENAME}"
+        ),
+    )
+
+
 def _check_task_template(tt: Path) -> list[CheckResult]:
     results = [
         make_check_result(

plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py

@@ -327,7 +327,7 @@ def test_harbor_dataset_from_ref_validates_task_ids(tmp_path: Path, task_ids: ob
 
 
 def test_harbor_dataset_maps_step_edge_cases(tmp_path: Path) -> None:
-    task_dir = tmp_path / "step-edge-task"
+    task_dir = tmp_path / "dataset" / "step-edge-task"
     _write(
         task_dir / "task.toml",
         """
@@ -344,7 +344,7 @@ def test_harbor_dataset_maps_step_edge_cases(tmp_path: Path) -> None:
 """.lstrip(),
     )
 
-    dataset = HarborDataset.from_path(task_dir)
+    dataset = HarborDataset.from_path(task_dir.parent)
 
     expected_task = Task(
         uri=task_dir.resolve().as_uri(),
@@ -389,7 +389,7 @@ def test_harbor_dataset_maps_step_edge_cases(tmp_path: Path) -> None:
 
 
 def test_harbor_dataset_from_ref_and_multistep_output(tmp_path: Path) -> None:
-    task_dir = tmp_path / "multi-step-task"
+    task_dir = tmp_path / "dataset" / "multi-step-task"
     _write(
         task_dir / "task.toml",
         """
@@ -414,7 +414,7 @@ def test_harbor_dataset_from_ref_and_multistep_output(tmp_path: Path) -> None:
 
     dataset = HarborDataset.from_ref(
         DatasetRef(
-            uri=task_dir.resolve().as_uri(),
+            uri=task_dir.parent.resolve().as_uri(),
             metadata={"id": "custom-id"},
         )
     )
@@ -492,7 +492,7 @@ def test_harbor_dataset_from_ref_and_multistep_output(tmp_path: Path) -> None:
 
     assert dataset.id == "custom-id"
     assert dataset.source == ResourceRef(
-        uri=task_dir.resolve().as_uri(),
+        uri=task_dir.parent.resolve().as_uri(),
         description="Harbor dataset root directory.",
     )
     assert dataset.tasks == [expected_task]
@@ -527,10 +527,25 @@ def test_harbor_dataset_rejects_invalid_refs_and_paths(tmp_path: Path) -> None:
         HarborDataset.from_ref(DatasetRef(uri="s3://bucket/dataset"))
 
 
+def test_task_directory_is_a_template_not_a_dataset(tmp_path: Path) -> None:
+    """Harbor's job config enumerates the task directories a dataset holds, and
+    nothing else, so only a template reads its own directory as one task."""
+    task_dir = tmp_path / "train"
+    _write(task_dir / "task.toml", "")
+
+    with pytest.raises(ValueError, match="itself a task directory") as exc_info:
+        HarborDataset.from_path(task_dir)
+    assert str(task_dir) in str(exc_info.value)
+
+    template = HarborDataset.from_path(task_dir, single_task=True)
+
+    assert [task.id for task in template.list_tasks()] == ["train"]
+
+
 def test_harbor_dataset_rejects_missing_task_ids(tmp_path: Path) -> None:
-    task_dir = tmp_path / "task-a"
+    task_dir = tmp_path / "dataset" / "task-a"
     _write(task_dir / "task.toml", "")
-    dataset = HarborDataset.from_path(task_dir)
+    dataset = HarborDataset.from_path(task_dir.parent)
 
     with pytest.raises(ValueError, match="Task id not found"):
         dataset.get_task("missing")
@@ -962,10 +977,10 @@ async def test_harbor_evaluator_accepts_valid_python_verifier(
 ) -> None:
     agent_dir = tmp_path / "agent"
     agent_dir.mkdir()
-    task_dir = tmp_path / "task-a"
+    task_dir = tmp_path / "dataset" / "task-a"
     _write(task_dir / "task.toml", "")
     _write(task_dir / "tests" / "check.py", "def check():\n    return True\n")
-    dataset = HarborDataset.from_path(task_dir)
+    dataset = HarborDataset.from_path(task_dir.parent)
     fake_job = _recording_job(tmp_path / "jobs" / "valid-python")
     monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", fake_job)
 
@@ -983,11 +998,11 @@ async def test_harbor_evaluator_rejects_invalid_configured_test_sh_before_job_cr
 ) -> None:
     agent_dir = tmp_path / "agent"
     agent_dir.mkdir()
-    task_dir = tmp_path / "task-a"
+    task_dir = tmp_path / "dataset" / "task-a"
     _write_task_config(task_dir, verifier_dir="test")
     _write(task_dir / "tests" / "test.sh", "echo ignored\n")
     _write(task_dir / "test" / "test.sh", "if true; then\n  echo broken\n")
-    dataset = HarborDataset.from_path(task_dir)
+    dataset = HarborDataset.from_path(task_dir.parent)
     fake_job = _recording_job(tmp_path / "jobs" / "invalid-shell")
     monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", fake_job)
 
@@ -1009,10 +1024,10 @@ async def test_harbor_evaluator_accepts_valid_legacy_test_sh(
 ) -> None:
     agent_dir = tmp_path / "agent"
     agent_dir.mkdir()
-    task_dir = tmp_path / "task-a"
+    task_dir = tmp_path / "dataset" / "task-a"
     _write(task_dir / "task.toml", "")
     _write(task_dir / "test" / "test.sh", "if true; then\n  echo valid\nfi\n")
-    dataset = HarborDataset.from_path(task_dir)
+    dataset = HarborDataset.from_path(task_dir.parent)
     fake_job = _recording_job(tmp_path / "jobs" / "valid-shell")
     monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", fake_job)
 
@@ -1026,9 +1041,9 @@ async def test_harbor_evaluator_accepts_valid_legacy_test_sh(
 @pytest.mark.asyncio
 async def test_harbor_evaluator_rejects_invalid_inputs(tmp_path: Path) -> None:
     evaluator = HarborEvaluator()
-    task_dir = tmp_path / "task-a"
+    task_dir = tmp_path / "dataset" / "task-a"
     _write(task_dir / "task.toml", "")
-    harbor_dataset = HarborDataset.from_path(task_dir)
+    harbor_dataset = HarborDataset.from_path(task_dir.parent)
 
     with pytest.raises(ValueError, match="Dataset must be a Harbor dataset"):
         await evaluator.run(agent=tmp_path, dataset=Dataset(id="base"), options={"import_path": "agent:Agent"})
@@ -1378,9 +1393,9 @@ def test_trial_resources_fallback_artifact(tmp_path):
 
 
 def test_harbor_dataset_get_task_found(tmp_path):
-    task_dir = tmp_path / "task-a"
+    task_dir = tmp_path / "dataset" / "task-a"
     _write(task_dir / "task.toml", "")
-    dataset = HarborDataset.from_path(task_dir)
+    dataset = HarborDataset.from_path(task_dir.parent)
     task = dataset.get_task("task-a")
     assert task.id == "task-a"
 
@@ -1390,9 +1405,9 @@ async def test_harbor_evaluator_force_rerun(tmp_path, monkeypatch):
     evaluator = HarborEvaluator()
     agent_dir = tmp_path / "agent"
     agent_dir.mkdir()
-    task_dir = tmp_path / "task-a"
+    task_dir = tmp_path / "dataset" / "task-a"
     _write(task_dir / "task.toml", "")
-    harbor_dataset = HarborDataset.from_path(task_dir)
+    harbor_dataset = HarborDataset.from_path(task_dir.parent)
 
     jobs_dir = tmp_path / "jobs"
     job_name = f"agent-{harbor_dataset.id}"