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).is_file()
+
+
+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(