diff --git a/plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/config.yaml b/plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/config.yaml index 99fa4b0d93..08cc95dcee 100644 --- a/plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/config.yaml +++ b/plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/config.yaml @@ -10,7 +10,7 @@ workspace: default base_url: http://localhost:8080 mode: local -evaluator_type: harbor +evaluator_type: harbor_native # Required per run. insight: "" diff --git a/plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/run.py b/plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/run.py index e43e840c8d..bd4340fbb7 100644 --- a/plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/run.py +++ b/plugins/nemo-eval-author/src/nemo_eval_author_plugin/eval_author/run.py @@ -51,7 +51,7 @@ async def run_eval_author( base_url: str | None, config: EvalAuthorConfig, agent: Path | str | None = None, - evaluator_type: EvaluatorType = "harbor", + evaluator_type: EvaluatorType = "harbor_native", model_refs: ConfiguredModelRefs | None = None, ) -> EvalAuthorResult: """Stage evaluation inputs, resolve one Insight, then run Eval Author. diff --git a/plugins/nemo-eval-author/tests/test_eval_author_repair_e2e.py b/plugins/nemo-eval-author/tests/test_eval_author_repair_e2e.py index b506fa0a99..0e79e446c7 100644 --- a/plugins/nemo-eval-author/tests/test_eval_author_repair_e2e.py +++ b/plugins/nemo-eval-author/tests/test_eval_author_repair_e2e.py @@ -17,6 +17,8 @@ from nemo_experimentalist_plugin.entities import DatasetValidationError, local_path_from_uri from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import ( HarborDataset, +) +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native import ( HarborEvaluator, HarborEvaluatorConfig, ) diff --git a/plugins/nemo-eval-author/tests/test_eval_author_run.py b/plugins/nemo-eval-author/tests/test_eval_author_run.py index 81ad26135b..3ec5cfa33c 100644 --- a/plugins/nemo-eval-author/tests/test_eval_author_run.py +++ b/plugins/nemo-eval-author/tests/test_eval_author_run.py @@ -14,6 +14,8 @@ from nemo_eval_author_plugin.eval_author.agent import EvalAuthor from nemo_eval_author_plugin.eval_author.models import EvalAuthorConfig, EvalAuthorResult from nemo_experimentalist_plugin.entities import Dataset, DatasetRef, ResourceRef, Task +from nemo_experimentalist_plugin.experimentalist.components.evaluator.base import EvaluatorType +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import HarborDataset from nemo_insights_plugin.entities import Insight @@ -110,6 +112,22 @@ async def run( ) +def _write_minimal_harbor_task(task_dir: Path, *, name: str, instruction: str) -> None: + task_dir.mkdir(parents=True) + (task_dir / "task.toml").write_text(f'[task]\nname = "{name}"\n', encoding="utf-8") + (task_dir / "instruction.md").write_text(instruction, encoding="utf-8") + + +def _dataset_snapshot(dataset: Dataset) -> dict[str, Any]: + return { + "type": f"{type(dataset).__module__}.{type(dataset).__qualname__}", + "id": dataset.id, + "source": dataset.source.model_dump(mode="json") if dataset.source is not None else None, + "tasks": [task.model_dump(mode="json") for task in dataset.list_tasks()], + "metadata": dict(dataset.metadata), + } + + @pytest.mark.asyncio async def test_run_eval_author_fails_before_side_effects_when_model_configuration_is_missing( monkeypatch: pytest.MonkeyPatch, @@ -140,11 +158,13 @@ def missing_model_refs() -> eval_author_run.ConfiguredModelRefs: make_client.assert_not_called() +@pytest.mark.parametrize("evaluator_type", ["harbor_native", "harbor_evaluator"]) @pytest.mark.asyncio async def test_run_eval_author_resolves_inputs_and_returns_datasets( monkeypatch: pytest.MonkeyPatch, model_clients: ClosingModelClients, tmp_path: Path, + evaluator_type: EvaluatorType, ) -> None: client = ClosingClient() insight = Insight( @@ -178,6 +198,7 @@ async def test_run_eval_author_resolves_inputs_and_returns_datasets( workspace="workspace-a", base_url="http://platform.test", config=EvalAuthorConfig(), + evaluator_type=evaluator_type, ) experiment_dir = (tmp_path / "experiment").resolve() @@ -190,8 +211,8 @@ async def test_run_eval_author_resolves_inputs_and_returns_datasets( assert backend.agent_calls == [ ("workspace-a", "insight-agent", experiment_dir / "eval_author" / "source-agent"), ] - assert [call[0] for call in dataset_factory.dataset_calls] == ["harbor", "harbor"] - assert dataset_factory.template_calls[0][0] == "harbor" + assert [call[0] for call in dataset_factory.dataset_calls] == [evaluator_type, evaluator_type] + assert dataset_factory.template_calls[0][0] == evaluator_type assert eval_author.call == ( insight, experiment_dir / "eval_author" / "source-agent", @@ -224,11 +245,122 @@ def test_public_apis_accept_train_validation_and_generated_task_inputs() -> None assert set(agent_private_run) == expected +@pytest.mark.asyncio +async def test_run_eval_author_builds_equivalent_real_harbor_inputs_for_both_evaluator_types( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + train_dir = tmp_path / "train" + validation_dir = tmp_path / "validation" + template_dir = tmp_path / "task-template" + _write_minimal_harbor_task( + train_dir / "train-task", + name="parity/train-task", + instruction="Complete the training task.\n", + ) + _write_minimal_harbor_task( + validation_dir / "validation-task", + name="parity/validation-task", + instruction="Complete the validation task.\n", + ) + _write_minimal_harbor_task( + template_dir, + name="parity/task-template", + instruction="Complete {{ instruction }}.\n", + ) + + insight = Insight( + workspace="workspace-a", + title="failure", + description="description", + agent=str(tmp_path / "agent-src"), + trace_refs=["trace-1"], + ) + clients: list[ClosingClient] = [] + model_client_sets: list[ClosingModelClients] = [] + eval_authors: list[FakeEvalAuthor] = [] + refs = eval_author_run.ConfiguredModelRefs( + default="workspace-a/default-model", + fast="workspace-a/fast-model", + ) + + def make_client(_base_url: str | None) -> ClosingClient: + client = ClosingClient() + clients.append(client) + return client + + def make_backend(**_: object) -> FakeBackend: + return FakeBackend(insight) + + def build_eval_author_agent(**_: object) -> FakeEvalAuthor: + eval_author = FakeEvalAuthor() + eval_authors.append(eval_author) + return eval_author + + async def resolve_model_clients(*_: object) -> ClosingModelClients: + model_clients = ClosingModelClients() + model_client_sets.append(model_clients) + return model_clients + + monkeypatch.setattr(eval_author_run, "make_client", make_client) + monkeypatch.setattr(eval_author_run, "make_experimentalist_backend", make_backend) + monkeypatch.setattr(eval_author_run, "build_eval_author_agent", build_eval_author_agent) + monkeypatch.setattr(eval_author_run, "configured_model_refs", lambda: refs) + monkeypatch.setattr(eval_author_run, "resolve_model_clients", resolve_model_clients) + + train_ref = DatasetRef(uri=str(train_dir), metadata={"id": "train"}) + validation_ref = DatasetRef(uri=str(validation_dir), metadata={"id": "validation"}) + template_ref = DatasetRef(uri=str(template_dir), metadata={"id": "task-template"}) + experiment_dir = tmp_path / "eval-author" + results: list[EvalAuthorResult] = [] + calls: list[tuple[Insight, Path, Task, Dataset, Dataset, ClosingClient]] = [] + + for evaluator_type in ("harbor_native", "harbor_evaluator"): + results.append( + await eval_author_run.run_eval_author( + insight="insight-remote-123", + train_dataset=train_ref, + validation_dataset=validation_ref, + task_template=template_ref, + experiment_dir=experiment_dir, + workspace="workspace-a", + base_url="http://platform.test", + config=EvalAuthorConfig(), + evaluator_type=evaluator_type, + ) + ) + call = eval_authors[-1].call + assert call is not None + calls.append(call) + + native_call, sdk_call = calls + _, _, native_template, native_train, native_validation, _ = native_call + _, _, sdk_template, sdk_train, sdk_validation, _ = sdk_call + for call in calls: + assert isinstance(call[3], HarborDataset) + assert isinstance(call[4], HarborDataset) + + assert _dataset_snapshot(native_train) == _dataset_snapshot(sdk_train) + assert _dataset_snapshot(native_validation) == _dataset_snapshot(sdk_validation) + assert native_template.model_dump(mode="json") == sdk_template.model_dump(mode="json") + + native_result, sdk_result = results + assert _dataset_snapshot(native_result.train_dataset) == _dataset_snapshot(sdk_result.train_dataset) + assert _dataset_snapshot(native_result.validation_dataset) == _dataset_snapshot(sdk_result.validation_dataset) + assert native_result.summary == sdk_result.summary == "Eval Author complete." + assert len(clients) == 2 + assert all(client.closed for client in clients) + assert len(model_client_sets) == 2 + assert all(model_clients.closed for model_clients in model_client_sets) + + +@pytest.mark.parametrize("evaluator_type", ["harbor_native", "harbor_evaluator"]) @pytest.mark.asyncio async def test_run_eval_author_hydrates_fileset_task_template( monkeypatch: pytest.MonkeyPatch, model_clients: ClosingModelClients, tmp_path: Path, + evaluator_type: EvaluatorType, ) -> None: downloads: list[tuple[str, str, str]] = [] @@ -263,12 +395,14 @@ async def download(self, *, remote_path: str, local_path: str, workspace: str) - workspace="workspace-a", base_url=None, config=EvalAuthorConfig(), + evaluator_type=evaluator_type, ) staged = (tmp_path / "experiment").resolve() / "dataset" / "task-template" assert downloads == [(template_ref.uri, str(staged), "workspace-a")] - assert dataset_factory.template_calls == [("harbor", template_ref.model_copy(update={"uri": str(staged)}))] + assert dataset_factory.template_calls == [(evaluator_type, template_ref.model_copy(update={"uri": str(staged)}))] assert [Path(ref.uri).name for _, ref in dataset_factory.dataset_calls] == ["train", "validation"] + assert [call[0] for call in dataset_factory.dataset_calls] == [evaluator_type, evaluator_type] assert client.closed assert model_clients.closed diff --git a/plugins/nemo-experimentalist/benchmarks/run.py b/plugins/nemo-experimentalist/benchmarks/run.py index a85dcefeab..27362ea9a7 100644 --- a/plugins/nemo-experimentalist/benchmarks/run.py +++ b/plugins/nemo-experimentalist/benchmarks/run.py @@ -22,8 +22,8 @@ TrialResult, local_path_from_uri, ) -from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import ( - HarborDataset, +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import HarborDataset +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native import ( HarborEvaluator, HarborEvaluatorConfig, ) diff --git a/plugins/nemo-experimentalist/examples/tau3-nooa-agent/record_tau_airline_traces.py b/plugins/nemo-experimentalist/examples/tau3-nooa-agent/record_tau_airline_traces.py index 958dba5a67..9558a74b9e 100644 --- a/plugins/nemo-experimentalist/examples/tau3-nooa-agent/record_tau_airline_traces.py +++ b/plugins/nemo-experimentalist/examples/tau3-nooa-agent/record_tau_airline_traces.py @@ -18,8 +18,8 @@ from nemo_experimentalist_plugin.client import make_client from nemo_experimentalist_plugin.entities import TrialResult, local_path_from_uri -from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import ( - HarborDataset, +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import HarborDataset +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native import ( HarborEvaluator, HarborEvaluatorConfig, ) diff --git a/plugins/nemo-experimentalist/pyproject.toml b/plugins/nemo-experimentalist/pyproject.toml index a8ff595863..6e8b622a6d 100644 --- a/plugins/nemo-experimentalist/pyproject.toml +++ b/plugins/nemo-experimentalist/pyproject.toml @@ -10,6 +10,7 @@ dependencies = [ "pydantic>=2", "httpx", "harbor>=0.16", + "nemo-evaluator-sdk", "opentelemetry-proto>=1.42.1", "protobuf>=6.0.0", "nooa", diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/config.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/config.py index e5dc55da36..b6c4f93c40 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/config.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/config.py @@ -24,6 +24,7 @@ from nemo_eval_author_plugin.eval_author.models import EvalAuthorConfig from nemo_experimentalist_plugin.experimentalist.components.analyzer import AnalyzerConfig from nemo_experimentalist_plugin.experimentalist.components.coder import CoderConfig +from nemo_experimentalist_plugin.experimentalist.components.evaluator.base import EvaluatorTypeField from nemo_experimentalist_plugin.experimentalist.components.goal_tree import GoalTreeConfig from nemo_experimentalist_plugin.experimentalist.components.proposer import ProposerConfig from pydantic import BaseModel, Field, model_validator @@ -157,6 +158,7 @@ def reject_legacy_curator_config(cls, data: Any) -> Any: coder: CoderConfig = Field(default_factory=CoderConfig) analyzer: AnalyzerConfig = Field(default_factory=AnalyzerConfig) proposer: ProposerConfig = Field(default_factory=ProposerConfig) + evaluator_type: EvaluatorTypeField = "harbor_native" evaluator: dict[str, Any] = Field(default_factory=dict) eval_author: EvalAuthorConfig = Field(default_factory=EvalAuthorConfig) diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/base.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/base.py index c3d227ac07..73a4c9a31a 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/base.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/base.py @@ -5,19 +5,68 @@ from __future__ import annotations +import logging from abc import ABC, abstractmethod from collections.abc import Sequence from pathlib import Path -from typing import Literal, TypeAlias +from typing import Annotated, Any, Literal, TypeAlias from nemo_experimentalist_plugin.entities import ( Dataset, EvaluationResult, TrialResult, ) -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, BeforeValidator, ConfigDict, Field + +logger = logging.getLogger(__name__) + +EvaluatorType: TypeAlias = Literal["harbor_native", "harbor_evaluator"] +"""Selects which evaluator drives the run. + +``harbor_native`` is the default: the plugin builds and runs Harbor's ``Job`` +directly. ``harbor_evaluator`` routes orchestration through the NeMo Evaluator +SDK's ``HarborAgentTaskRunner``, which owns the ``JobConfig``, the success-aware +job-dir cache, and agent import scoping. +""" +_DEPRECATED_EVALUATOR_TYPES: dict[str, EvaluatorType] = {"harbor": "harbor_native"} +"""Retired spellings still accepted on input, mapped to their current name. + +``harbor`` shipped before the rename, so experiment configs in the wild carry it. +``harbor_agent_task_runner`` is deliberately absent: it never shipped, so nothing +can be pinned to it. +""" + +_warned_evaluator_types: set[str] = set() + + +def normalize_evaluator_type(value: Any) -> Any: + """Map a retired evaluator-type spelling onto its current name, warning once. + + Non-strings and unknown strings pass through untouched so pydantic still + produces its own error for a genuinely invalid value, rather than this + function masking it with a confusing one. + + Args: + value(Any): Raw ``evaluator_type`` as supplied by config or a caller. + + Returns: + Any: The canonical evaluator type, or ``value`` unchanged. + """ + replacement = _DEPRECATED_EVALUATOR_TYPES.get(value) if isinstance(value, str) else None + if replacement is None: + return value + if value not in _warned_evaluator_types: + _warned_evaluator_types.add(value) + logger.warning( + "evaluator_type %r is deprecated and will be removed; use %r instead.", + value, + replacement, + ) + return replacement + -EvaluatorType: TypeAlias = Literal["harbor"] +EvaluatorTypeField: TypeAlias = Annotated[EvaluatorType, BeforeValidator(normalize_evaluator_type)] +"""``EvaluatorType`` for config models, accepting the retired spellings on input.""" class EvaluatorConfig(BaseModel): @@ -43,10 +92,13 @@ async def aggregate_results(self, results: Sequence[TrialResult]) -> dict[str, f """ Aggregate evaluation results from multiple runs. - Defaults to averaging each metric across all trials, treating trials that - did not emit a metric (e.g. failed trials) as contributing 0. The denominator - is always ``len(results)``, not the number of trials that reported each metric, - so failure counts against the aggregate score. + Averages each metric over trials with ``status == "completed"``. Anything + else is excluded from both the sum and the denominator, so a crash does not + pull the mean down — it shrinks the sample the mean is taken over, and a + round with no completed trial aggregates to ``{}`` rather than to zeros. + + Completed trials must all report the same metric keys; a mismatch raises + rather than silently averaging over different denominators per metric. Args: results(Sequence[TrialResult]): List of trial results to aggregate. @@ -57,7 +109,10 @@ async def aggregate_results(self, results: Sequence[TrialResult]) -> dict[str, f if not results: return {} - completed = [r for r in results if r.status != "failed"] + # Positive predicate on purpose: `!= "failed"` is equivalent while TrialStatus + # is Literal["completed", "failed"], but it would silently start averaging any + # third status someone adds. Opt statuses in, do not opt "failed" out. + completed = [r for r in results if r.status == "completed"] if not completed: return {} diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/factory.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/factory.py index e4572b5c21..25751c8ed1 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/factory.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/factory.py @@ -12,14 +12,25 @@ EvaluatorConfig, EvaluatorType, ) -from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import ( - HarborDataset, +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import HarborDataset +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_evaluator import ( + HarborRunnerConfig, + HarborRunnerEvaluator, +) +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native import ( HarborEvaluator, HarborEvaluatorConfig, ) -_SUPPORTED_EVALUATOR_TYPES = { - "harbor": (HarborDataset, HarborEvaluator, HarborEvaluatorConfig), +# Both Harbor-backed types read the same Harbor dataset layout; only who drives +# the run differs, so they share ``HarborDataset``. +_SUPPORTED_EVALUATOR_TYPES: dict[EvaluatorType, tuple[type[Dataset], type[Evaluator], type[EvaluatorConfig]]] = { + "harbor_native": (HarborDataset, HarborEvaluator, HarborEvaluatorConfig), + "harbor_evaluator": ( + HarborDataset, + HarborRunnerEvaluator, + HarborRunnerConfig, + ), } @@ -98,7 +109,10 @@ def build_evaluator( if isinstance(config, EvaluatorConfig): config = config.model_dump() elif not isinstance(config, dict): - raise TypeError(f"{evaluator_type.capitalize()} evaluator config must be an EvaluatorConfig or dict") + # Quoted rather than .capitalize()d: these names are snake_case, so + # capitalizing produced "Harbor_native" — and it silently changes + # shape every time a type is renamed. + raise TypeError(f"{evaluator_type!r} evaluator config must be an EvaluatorConfig or dict") evaluator_config = _SUPPORTED_EVALUATOR_TYPES[evaluator_type][2].model_validate(config) return _SUPPORTED_EVALUATOR_TYPES[evaluator_type][1]( options=evaluator_config, experiment_dir=experiment_dir diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py index a866d56f11..a7840899f5 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor.py @@ -1,34 +1,30 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Harbor dataset adapter for evaluator-domain task objects.""" +"""Shared Harbor dataset, dependency, input, and result adapters.""" from __future__ import annotations import asyncio import hashlib -import importlib.machinery import json import logging import os import re import shutil -import sys import tempfile import tomllib from collections.abc import Sequence from dataclasses import dataclass from pathlib import Path -from types import ModuleType, TracebackType -from typing import Any, Literal, TypeAlias, TypedDict +from types import TracebackType +from typing import Any, Protocol, TypeAlias, TypedDict from uuid import uuid4 from harbor.constants import MAIN_SERVICE_NAME from harbor.environments.base import BaseEnvironment from harbor.environments.factory import EnvironmentFactory -from harbor.job import DatasetConfig, Job, JobConfig from harbor.models.environment_type import EnvironmentType -from harbor.models.job.config import AgentConfig, ArtifactConfig, RetryConfig from harbor.models.task.task import Task as HarborTaskModel from harbor.models.trial.config import ServiceVolumeConfig from harbor.models.trial.paths import EnvironmentPaths, TrialPaths @@ -47,20 +43,10 @@ run_dependency_command, subset_dataset_id, ) -from nemo_experimentalist_plugin.experimentalist.components.evaluator.base import ( - Evaluator, - 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, -) -from pydantic import Field class HarborResourceSpec(TypedDict): @@ -118,12 +104,9 @@ class HarborResourceSpec(TypedDict): (_ORACLE_DIRNAME, "oracle_dir", ()), (_STEPS_DIRNAME, "steps_dir", ()), ) -_TRACE_ARTIFACT_SOURCE = "/app/traces" -_TRACE_ARTIFACT_DESTINATION = "traces" +DEFAULT_TRACE_ARTIFACT_SOURCE = "/app/traces" _ATIF_TRACE_SUFFIX = ".atif.json" _SHELL_SYNTAX_TIMEOUT_SEC = 10.0 -_AGENT_IMPORT_ROOT = "_nemo_experimentalist_eval_agents" -_IDENTIFIER_RE = re.compile(r"\W+") _TRIAL_LOG_DESCRIPTIONS = { "agent/oracle.txt": "Oracle-agent log captured when Harbor runs the reference solution.", "agent/setup/stdout.txt": "Agent setup stdout captured while Harbor uploads the agent and installs dependencies.", @@ -176,36 +159,6 @@ class _VerifierSyntaxFailure: column: int | None = None -class HarborEvaluatorConfig(EvaluatorConfig): - """Configuration for Harbor evaluator.""" - - job_name: str | None = Field( - default=None, description="Name of the job to run. If not provided, a default name will be generated." - ) - jobs_dir: Path = Field( - default=Path("eval-and-optimize") / "results", - description="Directory to store job results, resolved relative to the experiment directory.", - ) - n_attempts: int = Field(default=1) - n_concurrent_trials: int = Field(default=os.cpu_count() or 4) - quiet: bool = Field(default=False) - verifier_timeout_multiplier: float | None = Field(default=1.0) - agent_timeout_multiplier: float | None = Field(default=1.0) - agent_setup_timeout_multiplier: float | None = Field(default=1.0) - environment_build_timeout_multiplier: float | None = Field(default=1.0) - artifacts: list[str] = Field(default=[]) - retry: RetryConfig = Field(default=RetryConfig(exclude_exceptions=set())) - import_path: str = Field(default=DEFAULT_AGENT_IMPORT_PATH) - trace_dir: str = Field(default=_TRACE_ARTIFACT_SOURCE) - trace_format: Literal["otlp", "atif"] = Field( - default="otlp", - description=( - "Which trace artifact becomes the trial's trace. Both are still collected and " - "exposed in resources; this only selects the one that gets uploaded and analysed." - ), - ) - - class HarborDependencyRuntime(DependencyRuntime): """Harbor API-backed task environment runtime.""" @@ -364,69 +317,6 @@ def _chmod_path_chain(path: Path, stop_at: Path) -> None: HarborDataValue: TypeAlias = DataValue | ResourceRef -def _safe_identifier(value: str) -> str: - identifier = _IDENTIFIER_RE.sub("_", value).strip("_") - if not identifier: - identifier = "path" - if not identifier[0].isalpha() and identifier[0] != "_": - identifier = f"_{identifier}" - return identifier - - -def _agent_import_package(agent_path: Path) -> str: - path_parts = [_safe_identifier(part) for part in agent_path.parts if part not in {"", agent_path.anchor}] - tail = path_parts[-6:] or ["agent"] - digest = hashlib.sha256(str(agent_path).encode("utf-8")).hexdigest()[:12] - tail[-1] = f"{tail[-1]}_{digest}" - return ".".join([_AGENT_IMPORT_ROOT, *tail]) - - -def _ensure_package(name: str, search_path: Path | None = None) -> None: - parts = name.split(".") - for idx in range(1, len(parts) + 1): - package_name = ".".join(parts[:idx]) - package = sys.modules.get(package_name) - if package is None: - package = ModuleType(package_name) - package.__package__ = package_name - package.__spec__ = importlib.machinery.ModuleSpec(package_name, loader=None, is_package=True) - package.__path__ = [] # type: ignore[attr-defined] - sys.modules[package_name] = package - if idx > 1: - parent_name = ".".join(parts[: idx - 1]) - setattr(sys.modules[parent_name], parts[idx - 1], package) - if search_path is not None and idx == len(parts): - package.__path__ = [str(search_path)] # type: ignore[attr-defined] - - -def _scoped_import_path(agent_path: Path, import_path: str) -> tuple[str, str]: - module_name, attribute = split_import_path(import_path) - package_name = _agent_import_package(agent_path) - _ensure_package(package_name, search_path=agent_path) - return f"{package_name}.{module_name}:{attribute}", package_name - - -def _cleanup_scoped_imports(package_name: str) -> None: - package = sys.modules.get(package_name) - for module_name in list(sys.modules): - if module_name == package_name or module_name.startswith(f"{package_name}."): - sys.modules.pop(module_name, None) - parent_name, _, child_name = package_name.rpartition(".") - parent = sys.modules.get(parent_name) - if parent is not None and getattr(parent, child_name, None) is package: - delattr(parent, child_name) - parts = package_name.split(".") - for idx in range(len(parts) - 1, 0, -1): - module_name = ".".join(parts[:idx]) - if any(name.startswith(f"{module_name}.") for name in sys.modules): - break - package = sys.modules.pop(module_name, None) - parent_name, _, child_name = module_name.rpartition(".") - parent = sys.modules.get(parent_name) - if parent is not None and getattr(parent, child_name, None) is package: - delattr(parent, child_name) - - def _resolve_verifier_dir(task_dir: Path, config: dict[str, Any]) -> Path: verifier_config = config.get("verifier") if isinstance(verifier_config, dict): @@ -589,18 +479,6 @@ def _is_trial_log_path(relative_path: str) -> bool: return len(parts) == 3 and parts[0] == "agent" and parts[1].startswith("command-") and parts[2] == "stdout.txt" -def _with_trace_artifact(artifacts: Sequence[str | ArtifactConfig], trace_source: str) -> list[str | ArtifactConfig]: - for artifact in artifacts: - if isinstance(artifact, ArtifactConfig): - if artifact.source == trace_source or artifact.destination == _TRACE_ARTIFACT_DESTINATION: - return list(artifacts) - elif isinstance(artifact, str) and artifact in {trace_source, _TRACE_ARTIFACT_DESTINATION}: - return list(artifacts) - - trace_artifact = ArtifactConfig(source=trace_source, destination=_TRACE_ARTIFACT_DESTINATION) - return [trace_artifact, *artifacts] - - def _trial_error(exception_info: Any) -> dict[str, DataValue] | None: if exception_info is None: return None @@ -740,6 +618,172 @@ def _trial_resources( return resources, selected +class HarborJobOptions(Protocol): + """The two fields any Harbor-backed evaluator config must supply to locate a run. + + Declared structurally so :func:`resolve_harbor_run_inputs` can serve both + evaluator configs without importing either — the SDK-backed one lives in a + module that already imports this one. + """ + + jobs_dir: Path + job_name: str | None + + +@dataclass(frozen=True) +class HarborRunInputs: + """Validated inputs both Harbor-backed evaluators resolve the same way. + + ``dataset`` is carried through already narrowed to :class:`HarborDataset` so + callers need no second ``isinstance`` check to satisfy a type checker — the + validation happened once, in :func:`resolve_harbor_run_inputs`. + """ + + dataset: HarborDataset + dataset_path: Path + agent_path: Path + jobs_dir: Path + job_name: str + + @property + def job_dir(self) -> Path: + """Directory Harbor writes this run's per-trial results into.""" + return self.jobs_dir / self.job_name + + +async def resolve_harbor_run_inputs( + agent: Path, + dataset: Dataset, + options: HarborJobOptions, + experiment_dir: Path | None, +) -> HarborRunInputs: + """Validate an evaluation request and resolve the paths Harbor needs. + + The counterpart to :func:`trials_from_job_dir`: that one owns reading results + back, this one owns getting in. Both evaluator types must agree on what "the + same inputs" means — if one tightened its agent-path check or moved the + verifier preflight, the A/B parity tests would still pass while the two + silently diverged. Keeping the entry symmetric with the exit is what stops that. + + Verifier syntax is validated here, before any caller starts Docker: a typo in + ``tests/test.sh`` is far cheaper to catch now than after an image build. + + Args: + agent: Candidate directory to evaluate. + dataset: Must be a :class:`HarborDataset` with a resolvable source. + options: Evaluator options supplying ``jobs_dir`` and optional ``job_name``. + experiment_dir: Experiment root that ``jobs_dir`` resolves against; the + current working directory when ``None``. + + Returns: + HarborRunInputs: Resolved dataset/agent paths and the run's job location. + + Raises: + ValueError: If the dataset is not a Harbor dataset or has no source. + FileNotFoundError: If the agent directory does not exist. + DatasetValidationError: If a selected task's verifier fails preflight. + """ + if not isinstance(dataset, HarborDataset): + raise ValueError("Dataset must be a Harbor dataset") + if dataset.source is None: + raise ValueError("Harbor dataset source is required") + + agent_path = agent.expanduser().resolve() + if not agent_path.is_dir(): + raise FileNotFoundError(f"Harbor agent path not found: {agent_path}") + + await dataset.validate() + + return HarborRunInputs( + dataset=dataset, + dataset_path=local_path_from_uri(dataset.source.uri, context="Harbor dataset reference").resolve(), + agent_path=agent_path, + jobs_dir=(experiment_dir or Path.cwd()) / options.jobs_dir, + # Derived from the *resolved* directory, not the caller's spelling of it. + # `job_name` is the cache identity, and the SDK's scoped agent import derives + # its package name from the resolved dir too — the two must agree or a job dir + # can be reused for a different agent (`--agent .` has an empty `.name`; a + # symlink keeps its own name while resolving elsewhere). + job_name=options.job_name or f"{agent_path.name}-{dataset.id}", + ) + + +def trials_from_job_dir( + job_dir: Path, + tasks: Sequence[Task], + *, + trace_format: str = "otlp", +) -> list[TrialResult]: + """Adapt a finished Harbor job directory into evaluator-domain trial results. + + The job directory is the authoritative source for both Harbor-backed + evaluators: it carries every verifier metric (not just the primary reward), + the attempt index, the trial's error shape, and the on-disk trace and + artifact references that the Analyzer and the Coder read. Whoever + orchestrated the run — Harbor's ``Job`` directly or the SDK's + ``HarborAgentTaskRunner`` — writes the same tree, so both evaluators share + this adapter and produce equivalent :class:`TrialResult` objects. + + Args: + job_dir: Harbor job directory holding one ``/result.json`` per attempt. + tasks: Dataset tasks the run was asked to cover, used to resolve each + trial back to its short Experimentalist task id and metric spec. + trace_format: Trace artifact format selected as the trial's primary trace. + + Returns: + list[TrialResult]: One result per trial directory that wrote a ``result.json``. + + Raises: + FileNotFoundError: If the job directory does not exist. Returning no trials + would be aggregated as an empty-but-valid result and read as a run that + legitimately scored nothing, so an orchestrator that produced no job + directory at all is surfaced instead of swallowed. + """ + if not job_dir.is_dir(): + raise FileNotFoundError( + f"Harbor job directory not found: {job_dir}. The run produced no results — " + "check the orchestrator's logs for a job that failed before writing any trial." + ) + + task_map = {task.id: task for task in tasks} + trials: list[TrialResult] = [] + for trial_dir in sorted(path for path in job_dir.iterdir() if path.is_dir()): + result_path = trial_dir / "result.json" + if not result_path.is_file(): + continue + + trial_data = json.loads(result_path.read_text(encoding="utf-8")) + trial_id = trial_data.get("trial_name") + if not isinstance(trial_id, str) or not trial_id: + trial_id = trial_dir.name + + task_id = _resolve_trial_task_id(trial_id, trial_data, task_map) + task = task_map.get(task_id) + # Prefer the dataset's own spec, but only when it points at a verifier; + # a ref-less spec carries no more than the one derived from the trial dir. + metric_spec = task.metric_specs.get("reward") if task is not None else None + if metric_spec is None or metric_spec.ref is None: + metric_spec = _trial_metric_spec(trial_dir, trial_data) + + exception_info = trial_data.get("exception_info") + resources, trace = _trial_resources(trial_dir, trace_format=trace_format) + + trials.append( + TrialResult( + id=trial_id, + task_id=task_id, + attempt=_trial_attempt(trial_id), + status="completed" if exception_info is None else "failed", + error=_trial_error(exception_info), + trace=trace, + outputs={}, + resources=resources, + metrics=_trial_metrics(trial_dir, trial_data, metric_spec), + ) + ) + return trials + + class HarborDataset(Dataset): """Harbor task collection mapped onto generic evaluator-domain objects. @@ -1317,96 +1361,3 @@ def subset(self, task_ids: Sequence[str]) -> HarborDataset: tasks=tasks, metadata=dict(self.metadata), ) - - -class HarborEvaluator(Evaluator): - """Run Harbor evaluations and return parsed reward payloads.""" - - evaluator_type: EvaluatorType = "harbor" - - def __init__(self, options: HarborEvaluatorConfig | None = None, experiment_dir: Path | None = None) -> None: - super().__init__(options or HarborEvaluatorConfig(), experiment_dir=experiment_dir) - - async def _run(self, agent: Path, dataset: Dataset, options: HarborEvaluatorConfig) -> Sequence[TrialResult]: - if not isinstance(dataset, HarborDataset): - raise ValueError("Dataset must be a Harbor dataset") - - if dataset.source is None: - raise ValueError("Harbor dataset source is required") - dataset_path = local_path_from_uri(dataset.source.uri, context="Harbor dataset reference").resolve() - - options_dict = options.model_dump() - experiment_dir = self.experiment_dir or Path.cwd() - options_dict["jobs_dir"] = experiment_dir / options.jobs_dir - options_dict["job_name"] = options.job_name or f"{agent.name}-{dataset.id}" - import_path: str = options_dict.pop("import_path") - trace_dir: str = options_dict.pop("trace_dir", _TRACE_ARTIFACT_SOURCE) - # Harbor's JobConfig forbids unknown keys, so this must not survive into it. - trace_format: str = options_dict.pop("trace_format", "otlp") - options_dict["artifacts"] = _with_trace_artifact(options_dict.get("artifacts") or [], trace_dir) - force_rerun: bool = options_dict.pop("force_rerun", False) - - agent_path = agent.expanduser().resolve() - - if not agent_path.is_dir(): - raise FileNotFoundError(f"Harbor agent path not found: {agent_path}") - - await dataset.validate() - - scoped_import_path, scoped_package = _scoped_import_path(agent_path, import_path) - agents_config = [AgentConfig(import_path=scoped_import_path)] - datasets_config = [DatasetConfig(path=dataset_path, task_names=[task.id for task in dataset.tasks])] - job_config = JobConfig(**options_dict, agents=agents_config, datasets=datasets_config) - if force_rerun: - job_dir = job_config.jobs_dir / job_config.job_name - if job_dir.exists(): - shutil.rmtree(job_dir) - - try: - job = await Job.create(job_config) - await job.run() - finally: - _cleanup_scoped_imports(scoped_package) - - trials = await self._trials_from_dir(job.job_dir, dataset.tasks, trace_format=trace_format) - return trials - - async def _trials_from_dir( - self, job_dir: Path, tasks: Sequence[Task], *, trace_format: str = "otlp" - ) -> Sequence[TrialResult]: - task_map = {task.id: task for task in tasks} - trials: list[TrialResult] = [] - for trial_dir in sorted(path for path in job_dir.iterdir() if path.is_dir()): - result_path = trial_dir / "result.json" - if not result_path.is_file(): - continue - - trial_data = json.loads(result_path.read_text(encoding="utf-8")) - trial_id = trial_data.get("trial_name") - if not isinstance(trial_id, str) or not trial_id: - trial_id = trial_dir.name - - task_id = _resolve_trial_task_id(trial_id, trial_data, task_map) - task = task_map.get(task_id) - metric_spec = task.metric_specs["reward"] if task is not None and "reward" in task.metric_specs else None - if metric_spec is not None and metric_spec.ref is None: - metric_spec = None - metric_spec = metric_spec or _trial_metric_spec(trial_dir, trial_data) - - exception_info = trial_data.get("exception_info") - resources, trace = _trial_resources(trial_dir, trace_format=trace_format) - - trials.append( - TrialResult( - id=trial_id, - task_id=task_id, - attempt=_trial_attempt(trial_id), - status="completed" if exception_info is None else "failed", - error=_trial_error(exception_info), - trace=trace, - outputs={}, - resources=resources, - metrics=_trial_metrics(trial_dir, trial_data, metric_spec), - ) - ) - return trials diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor_evaluator.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor_evaluator.py new file mode 100644 index 0000000000..dde0ec6059 --- /dev/null +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor_evaluator.py @@ -0,0 +1,256 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Harbor evaluator that delegates orchestration to the NeMo Evaluator SDK. + +``HarborEvaluator`` builds Harbor's ``JobConfig`` and drives ``Job`` itself. This +evaluator hands that job to the SDK's ``HarborAgentTaskRunner`` instead: the SDK +owns the ``JobConfig``, the success-aware job-directory cache, and the scoped +agent import. Harbor still does the work underneath — the difference is who owns +the orchestration. + +Results are read back off the Harbor job directory through the same +:func:`~...evaluator.harbor.trials_from_job_dir` adapter ``harbor_native`` uses. +**That sharing is the point**: one adapter over one source of truth is what makes +the two evaluator types produce equivalent :class:`TrialResult` objects, rather +than two parsers that have to be kept in agreement. + +Scoring is deliberately left to Harbor. Its verifier already computes the rewards +and writes them to ``/result.json``; the SDK's metric layer only reads them +back, so running the trials through ``AgentEvaluator`` would add a scoring pass +whose output this evaluator discards. When ``harbor_native`` is eventually removed, +the natural move is to consume ``AgentEvalTrial`` directly — see +``temp/evaluator/plans/AgentEvalResult_as_shared_harbor_interface.md``. +""" + +from __future__ import annotations + +import logging +import os +from collections.abc import Sequence +from pathlib import Path +from typing import Literal + +from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import ( + HarborAgentTaskRunner, + HarborRuntimeConfig, + discover_harbor_tasks, +) +from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalTask +from nemo_experimentalist_plugin.entities import Dataset, TrialResult, local_path_from_uri +from nemo_experimentalist_plugin.experimentalist.components.evaluator.base import ( + Evaluator, + EvaluatorConfig, + EvaluatorType, +) +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import ( + DEFAULT_TRACE_ARTIFACT_SOURCE, + HarborDataset, + resolve_harbor_run_inputs, + trials_from_job_dir, +) +from pydantic import ConfigDict, Field + +logger = logging.getLogger(__name__) + + +class HarborTaskNameError(ValueError): + """A dataset task could not be mapped onto exactly one Harbor task name.""" + + +class HarborRunnerConfig(EvaluatorConfig): + """Configuration for the SDK-backed Harbor evaluator. + + Every field maps onto exactly one ``HarborRuntimeConfig`` field. Unknown keys + are rejected rather than silently ignored: several plain-``HarborEvaluator`` + options (notably the full ``retry`` model) have no unambiguous SDK equivalent, + so passing them here is a configuration error, not a no-op. + + ``agent_dir`` is deliberately absent — it is always derived from the candidate + being evaluated, so a config cannot point the run at a different agent. + + One asymmetry to know when A/B-ing against ``HarborEvaluatorConfig``: at their + defaults the two are equivalent (Harbor resolves an unset phase multiplier to + the global ``timeout_multiplier``, which defaults to ``1.0``), but they diverge + once tuned. This config exposes the global ``timeout_multiplier`` and leaves the + phase multipliers unset so they inherit it; the plain config has no global knob + and pins each phase to ``1.0``, which masks it. Set the phase multipliers + explicitly on both sides when comparing non-default timeouts. + """ + + model_config = ConfigDict(extra="forbid") + + job_name: str | None = Field( + default=None, + description=( + "Harbor job name. Defaults to the loop's deterministic '-', " + "which is what makes the SDK's success-aware job-dir cache usable." + ), + ) + jobs_dir: Path = Field( + default=Path("eval-and-optimize") / "results", + description="Directory to store job results, resolved relative to the experiment directory.", + ) + n_attempts: int = Field(default=1, ge=1, description="Number of attempts Harbor runs per task.") + n_concurrent_trials: int = Field( + default=os.cpu_count() or 4, ge=1, description="Maximum number of concurrent Harbor trials." + ) + quiet: bool = Field(default=False, description="Suppress Harbor's trial progress display.") + artifacts: list[str] = Field(default=[], description="Additional Harbor artifact sources to collect per trial.") + trace_dir: str = Field( + default=DEFAULT_TRACE_ARTIFACT_SOURCE, + description="Container path of agent traces, collected into the trial's 'traces' artifact directory.", + ) + trace_format: Literal["otlp", "atif"] = Field( + default="otlp", + description=( + "Which collected trace artifact becomes the trial's primary trace. Both formats remain available in " + "the trial resources." + ), + ) + max_retries: int = Field(default=0, ge=0, description="Harbor per-trial retries on transient failures.") + timeout_multiplier: float | None = Field(default=None, description="Global Harbor timeout multiplier.") + agent_timeout_multiplier: float | None = Field(default=None, description="Agent-phase timeout multiplier.") + verifier_timeout_multiplier: float | None = Field(default=None, description="Verifier-phase timeout multiplier.") + agent_setup_timeout_multiplier: float | None = Field(default=None, description="Agent-setup timeout multiplier.") + environment_build_timeout_multiplier: float | None = Field( + default=None, description="Environment-build timeout multiplier." + ) + import_path: str = Field( + default="harbor_wrapper:WrappedAgent", + description="Harbor agent import path resolved inside the candidate directory.", + ) + + +class HarborRunnerEvaluator(Evaluator): + """Run Harbor through the SDK's ``HarborAgentTaskRunner`` and parse the job dir.""" + + evaluator_type: EvaluatorType = "harbor_evaluator" + + def __init__( + self, + options: HarborRunnerConfig | None = None, + experiment_dir: Path | None = None, + ) -> None: + super().__init__(options or HarborRunnerConfig(), experiment_dir=experiment_dir) + + async def _run( + self, + agent: Path, + dataset: Dataset, + options: EvaluatorConfig, + ) -> Sequence[TrialResult]: + if not isinstance(options, HarborRunnerConfig): + raise TypeError("Options must be a HarborRunnerConfig") + + inputs = await resolve_harbor_run_inputs(agent, dataset, options, self.experiment_dir) + harbor_dataset = inputs.dataset + sdk_tasks = _sdk_tasks_for(harbor_dataset) + + runtime_config = HarborRuntimeConfig( + jobs_dir=inputs.jobs_dir, + job_name=inputs.job_name, + agent_import_path=options.import_path, + agent_dir=inputs.agent_path, + n_attempts=options.n_attempts, + n_concurrent_trials=options.n_concurrent_trials, + quiet=options.quiet, + force_rerun=options.force_rerun, + artifacts=list(options.artifacts), + trace_dir=options.trace_dir, + max_retries=options.max_retries, + timeout_multiplier=options.timeout_multiplier, + agent_timeout_multiplier=options.agent_timeout_multiplier, + verifier_timeout_multiplier=options.verifier_timeout_multiplier, + agent_setup_timeout_multiplier=options.agent_setup_timeout_multiplier, + environment_build_timeout_multiplier=options.environment_build_timeout_multiplier, + ) + + # Two different name spaces, and mixing them up produces either an empty + # run or an empty cache: + # * Harbor's local-dataset `task_names` filter matches the task + # *directory* name, which is the Experimentalist task id. + # * `result.json` records `[task].name` from task.toml, which is what + # the SDK's tasks are keyed by and what its cache counts. + runner = HarborAgentTaskRunner( + config=runtime_config, + dataset_path=inputs.dataset_path, + task_names=[task.id for task in harbor_dataset.tasks], + ) + # Called for its effect — running (or resuming) the Harbor job. The returned + # trials are not the contract: the job dir is, and it is shared with + # `harbor_native`, which is what keeps the two types equivalent. Note the + # trials are not *lossy* — `metadata["reward_details"]` carries every verifier + # reward — so the reason to ignore them is the shared source of truth, not + # missing data. + sdk_trials = await runner.run_tasks(list(sdk_tasks.values())) + logger.debug("SDK Harbor runner returned %d trial(s) for job %s", len(sdk_trials), inputs.job_name) + + return trials_from_job_dir(inputs.job_dir, harbor_dataset.tasks, trace_format=options.trace_format) + + +def _sdk_tasks_for(dataset: HarborDataset) -> dict[str, AgentEvalTask]: + """Map each selected dataset task onto the SDK task carrying its full Harbor name. + + The mapping is by task *directory*, never by name similarity: the SDK reads + ``[task].name`` from the same ``task.toml`` Harbor will read, so matching on + the directory guarantees the ids we hand the runner are exactly the + ``task_name`` values Harbor writes into ``result.json``. + + Args: + dataset: The (possibly subset) Harbor dataset being evaluated. + + Returns: + dict[str, AgentEvalTask]: Selected tasks keyed by Experimentalist task id, + in dataset order. + + Raises: + ValueError: If the dataset has no resolvable source directory. + HarborTaskNameError: If a selected task has no discovered counterpart, or + if two selected tasks resolve to the same full Harbor name. + """ + if dataset.source is None: + raise ValueError("Harbor dataset source is required") + dataset_path = local_path_from_uri(dataset.source.uri, context="Harbor dataset reference").resolve() + discovered = discover_harbor_tasks(dataset_path) + by_dir: dict[Path, AgentEvalTask] = {} + for sdk_task in discovered: + task_dir = sdk_task.metadata.get("harbor_task_dir") + if isinstance(task_dir, str) and task_dir: + by_dir[Path(task_dir).resolve()] = sdk_task + + selected: dict[str, AgentEvalTask] = {} + full_names: dict[str, str] = {} + for task in dataset.tasks: + if not task.uri: + raise HarborTaskNameError(f"Harbor task {task.id!r} has no URI, so its Harbor name cannot be resolved") + task_dir = local_path_from_uri(task.uri, context="Harbor task reference").resolve() + sdk_task = by_dir.get(task_dir) + if sdk_task is None: + raise HarborTaskNameError( + f"Harbor task {task.id!r} at {task_dir} was not discovered under dataset {dataset_path}; " + "the dataset directory and the task directories must agree" + ) + if sdk_task.id in full_names: + raise HarborTaskNameError( + f"Harbor tasks {full_names[sdk_task.id]!r} and {task.id!r} both declare [task].name = " + f"{sdk_task.id!r}; task names must be unique within a dataset" + ) + full_names[sdk_task.id] = task.id + selected[task.id] = sdk_task + return selected + + +def harbor_task_names(dataset: HarborDataset) -> dict[str, str]: + """Return ``{experimentalist_task_id: full_harbor_name}`` for a Harbor dataset. + + The readable view of the two-namespace translation ``_run`` depends on. Pass a + ``dataset.subset(...)`` to scope it to selected tasks. + + Args: + dataset: Harbor dataset whose source directory holds the tasks. + + Returns: + dict[str, str]: Short Experimentalist task id to full Harbor ``[task].name``. + """ + return {task_id: sdk_task.id for task_id, sdk_task in _sdk_tasks_for(dataset).items()} diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor_native.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor_native.py new file mode 100644 index 0000000000..0c1bac90c2 --- /dev/null +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/evaluator/harbor_native.py @@ -0,0 +1,215 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Direct Harbor evaluator orchestration.""" + +import hashlib +import importlib.machinery +import os +import re +import shutil +import sys +from collections.abc import Sequence +from pathlib import Path +from types import ModuleType +from typing import Literal + +from harbor.job import DatasetConfig, Job, JobConfig +from harbor.models.job.config import AgentConfig, ArtifactConfig, RetryConfig +from nemo_experimentalist_plugin.entities import Dataset, Task, TrialResult +from nemo_experimentalist_plugin.experimentalist.components.evaluator.base import ( + Evaluator, + EvaluatorConfig, + EvaluatorType, +) +from nemo_experimentalist_plugin.experimentalist.components.evaluator.entrypoint import ( + DEFAULT_AGENT_IMPORT_PATH, + split_import_path, +) +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import ( + DEFAULT_TRACE_ARTIFACT_SOURCE, + resolve_harbor_run_inputs, + trials_from_job_dir, +) +from pydantic import Field + +_TRACE_ARTIFACT_DESTINATION = "traces" +_AGENT_IMPORT_ROOT = "_nemo_experimentalist_eval_agents" +_IDENTIFIER_RE = re.compile(r"\W+") + + +def _safe_identifier(value: str) -> str: + identifier = _IDENTIFIER_RE.sub("_", value).strip("_") + if not identifier: + identifier = "path" + if not identifier[0].isalpha() and identifier[0] != "_": + identifier = f"_{identifier}" + return identifier + + +def _agent_import_package(agent_path: Path) -> str: + path_parts = [_safe_identifier(part) for part in agent_path.parts if part not in {"", agent_path.anchor}] + tail = path_parts[-6:] or ["agent"] + digest = hashlib.sha256(str(agent_path).encode("utf-8")).hexdigest()[:12] + tail[-1] = f"{tail[-1]}_{digest}" + return ".".join([_AGENT_IMPORT_ROOT, *tail]) + + +def _ensure_package(name: str, search_path: Path | None = None) -> None: + parts = name.split(".") + for idx in range(1, len(parts) + 1): + package_name = ".".join(parts[:idx]) + package = sys.modules.get(package_name) + if package is None: + package = ModuleType(package_name) + package.__package__ = package_name + package.__spec__ = importlib.machinery.ModuleSpec(package_name, loader=None, is_package=True) + package.__path__ = [] # type: ignore[attr-defined] + sys.modules[package_name] = package + if idx > 1: + parent_name = ".".join(parts[: idx - 1]) + setattr(sys.modules[parent_name], parts[idx - 1], package) + if search_path is not None and idx == len(parts): + package.__path__ = [str(search_path)] # type: ignore[attr-defined] + + +def _scoped_import_path(agent_path: Path, import_path: str) -> tuple[str, str]: + module_name, attribute = split_import_path(import_path) + + package_name = _agent_import_package(agent_path) + _ensure_package(package_name, search_path=agent_path) + scoped = f"{package_name}.{module_name}" + return f"{scoped}:{attribute}", package_name + + +def _cleanup_scoped_imports(package_name: str) -> None: + package = sys.modules.get(package_name) + for module_name in list(sys.modules): + if module_name == package_name or module_name.startswith(f"{package_name}."): + sys.modules.pop(module_name, None) + parent_name, _, child_name = package_name.rpartition(".") + parent = sys.modules.get(parent_name) + if parent is not None and getattr(parent, child_name, None) is package: + delattr(parent, child_name) + parts = package_name.split(".") + for idx in range(len(parts) - 1, 0, -1): + module_name = ".".join(parts[:idx]) + if any(name.startswith(f"{module_name}.") for name in sys.modules): + break + package = sys.modules.pop(module_name, None) + parent_name, _, child_name = module_name.rpartition(".") + parent = sys.modules.get(parent_name) + if parent is not None and getattr(parent, child_name, None) is package: + delattr(parent, child_name) + + +def _validated_job_dir(jobs_dir: Path, job_name: str) -> Path: + """Resolve ``jobs_dir / job_name`` and require it stay under ``jobs_dir``.""" + resolved_jobs_dir = jobs_dir.resolve() + candidate = (jobs_dir / job_name).resolve() + if candidate == resolved_jobs_dir or not candidate.is_relative_to(resolved_jobs_dir): + raise ValueError( + f"Resolved job directory {candidate} is not a strict descendant of " + f"{resolved_jobs_dir} (job_name={job_name!r})" + ) + return candidate + + +def _with_trace_artifact(artifacts: Sequence[str | ArtifactConfig], trace_source: str) -> list[str | ArtifactConfig]: + for artifact in artifacts: + if isinstance(artifact, ArtifactConfig): + if artifact.source == trace_source or artifact.destination == _TRACE_ARTIFACT_DESTINATION: + return list(artifacts) + elif isinstance(artifact, str) and artifact in {trace_source, _TRACE_ARTIFACT_DESTINATION}: + return list(artifacts) + + trace_artifact = ArtifactConfig(source=trace_source, destination=_TRACE_ARTIFACT_DESTINATION) + return [trace_artifact, *artifacts] + + +class HarborEvaluatorConfig(EvaluatorConfig): + """Configuration for direct Harbor evaluation.""" + + job_name: str | None = Field( + default=None, description="Name of the job to run. If not provided, a default name will be generated." + ) + jobs_dir: Path = Field( + default=Path("eval-and-optimize") / "results", + description="Directory to store job results, resolved relative to the experiment directory.", + ) + n_attempts: int = Field(default=1) + n_concurrent_trials: int = Field(default=os.cpu_count() or 4) + quiet: bool = Field(default=False) + verifier_timeout_multiplier: float | None = Field(default=1.0) + agent_timeout_multiplier: float | None = Field(default=1.0) + agent_setup_timeout_multiplier: float | None = Field(default=1.0) + environment_build_timeout_multiplier: float | None = Field(default=1.0) + artifacts: list[str] = Field(default=[]) + retry: RetryConfig = Field(default=RetryConfig(exclude_exceptions=set())) + import_path: str = Field(default=DEFAULT_AGENT_IMPORT_PATH) + trace_dir: str = Field(default=DEFAULT_TRACE_ARTIFACT_SOURCE) + trace_format: Literal["otlp", "atif"] = Field( + default="otlp", + description=( + "Which trace artifact becomes the trial's trace. Both are still collected and " + "exposed in resources; this only selects the one that gets uploaded and analysed." + ), + ) + + +class HarborEvaluator(Evaluator): + """Run Harbor evaluations directly and return parsed reward payloads.""" + + evaluator_type: EvaluatorType = "harbor_native" + + def __init__(self, options: HarborEvaluatorConfig | None = None, experiment_dir: Path | None = None) -> None: + super().__init__(options or HarborEvaluatorConfig(), experiment_dir=experiment_dir) + + async def _run(self, agent: Path, dataset: Dataset, options: EvaluatorConfig) -> Sequence[TrialResult]: + # Widened from HarborEvaluatorConfig to match the base class contract: + # Evaluator.run() passes an EvaluatorConfig instance through unchanged, so + # narrowing here would be an unsound override. The guard is defensive only — + # both the factory and the loop build this config via type(self.options). + if not isinstance(options, HarborEvaluatorConfig): + raise TypeError("Options must be a HarborEvaluatorConfig") + + inputs = await resolve_harbor_run_inputs(agent, dataset, options, self.experiment_dir) + harbor_dataset = inputs.dataset + dataset_path = inputs.dataset_path + agent_path = inputs.agent_path + + options_dict = options.model_dump() + options_dict["jobs_dir"] = inputs.jobs_dir + options_dict["job_name"] = inputs.job_name + import_path: str = options_dict.pop("import_path") + trace_dir: str = options_dict.pop("trace_dir", DEFAULT_TRACE_ARTIFACT_SOURCE) + trace_format: str = options_dict.pop("trace_format", "otlp") + options_dict["artifacts"] = _with_trace_artifact(options_dict.get("artifacts") or [], trace_dir) + force_rerun: bool = options_dict.pop("force_rerun", False) + + scoped_import_path, scoped_package = _scoped_import_path(agent_path, import_path) + agents_config = [AgentConfig(import_path=scoped_import_path)] + datasets_config = [DatasetConfig(path=dataset_path, task_names=[task.id for task in harbor_dataset.tasks])] + job_config = JobConfig(**options_dict, agents=agents_config, datasets=datasets_config) + if force_rerun: + job_dir = _validated_job_dir(job_config.jobs_dir, job_config.job_name) + if job_dir.exists(): + shutil.rmtree(job_dir) + + try: + job = await Job.create(job_config) + await job.run() + finally: + _cleanup_scoped_imports(scoped_package) + + trials = await self._trials_from_dir(job.job_dir, harbor_dataset.tasks, trace_format=trace_format) + return trials + + async def _trials_from_dir( + self, + job_dir: Path, + tasks: Sequence[Task], + *, + trace_format: str = "otlp", + ) -> Sequence[TrialResult]: + return trials_from_job_dir(job_dir, tasks, trace_format=trace_format) diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/deps.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/deps.py index f2547335a5..be1275332a 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/deps.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/deps.py @@ -40,6 +40,9 @@ class ExperimentalistDeps(BaseModel): A :class:`~pathlib.Path` means a local directory; a plain string means a fileset ID that the backend will resolve at evaluation time. Defaults to None and must be set before ``run()``. + evaluator_type: Which evaluator drives the run. ``run_experimentalist`` + passes the value resolved from ``EvolutionaryOptimizerConfig``; the + default here only applies to callers that construct deps directly. backend: Shared data-access backend used by every tool. The CLI owns the backend's client lifecycle — tools must not close it. config: Optional per-run override of the EvolutionaryOptimizerConfig. @@ -54,7 +57,7 @@ class ExperimentalistDeps(BaseModel): train_dataset: DatasetRef validation_dataset: DatasetRef task_template: DatasetRef | None = None - evaluator_type: EvaluatorType = "harbor" + evaluator_type: EvaluatorType = "harbor_native" agent_spec: str | None = None backend: ExperimentalistBackend | None = None reporter: RunReporter | None = None diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/run.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/run.py index c54055067e..a1b960911a 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/run.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/run.py @@ -116,6 +116,7 @@ async def run_experimentalist( train_dataset=train_dataset, validation_dataset=validation_dataset, task_template=task_template, + evaluator_type=config.evaluator_type, backend=backend, reporter=reporter, config=config, diff --git a/plugins/nemo-experimentalist/tests/experimentalist/conftest.py b/plugins/nemo-experimentalist/tests/experimentalist/conftest.py new file mode 100644 index 0000000000..f09ad1d71f --- /dev/null +++ b/plugins/nemo-experimentalist/tests/experimentalist/conftest.py @@ -0,0 +1,49 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shared fixtures for the evaluator test modules. + +Test directories carry no ``__init__.py`` in this repo, so helpers are shared as +fixtures rather than imports. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from typing import Any + +import pytest + + +def _comparable_trials(trials: Sequence[Any], *, include_id: bool = False) -> list[dict[str, Any]]: + projected = [] + for trial in trials: + entry: dict[str, Any] = { + "task_id": trial.task_id, + "attempt": trial.attempt, + "status": trial.status, + "error": trial.error, + "metrics": {name: metric.value for name, metric in trial.metrics.items()}, + "has_trace": trial.trace is not None, + "resource_kinds": sorted({key.split(":")[0] for key in trial.resources}), + } + if include_id: + entry["id"] = trial.id + projected.append(entry) + return sorted(projected, key=lambda entry: str(entry["task_id"])) + + +@pytest.fixture +def comparable_trials() -> Callable[..., list[dict[str, Any]]]: + """Project trials down to the fields the optimizer loop actually consumes. + + Both A/B parity tests compare evaluator output through this one projection, so + they cannot drift on what "equivalent trials" means — which is the single thing + those tests exist to pin down. ``resources`` is compared by key *kind* (the part + before ``:``) rather than by full key, because artifact keys embed per-trial file + names that legitimately differ between runs. + + Pass ``include_id=True`` to also compare trial ids. That is only meaningful when + both sides read the same job directory — Harbor mints a random suffix per run. + """ + return _comparable_trials diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_dataset_staging_loop.py b/plugins/nemo-experimentalist/tests/experimentalist/test_dataset_staging_loop.py index 6d2404c76c..a53863ec39 100644 --- a/plugins/nemo-experimentalist/tests/experimentalist/test_dataset_staging_loop.py +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_dataset_staging_loop.py @@ -101,7 +101,7 @@ async def run( backend=backend, workspace="default", config=config, - evaluator_type="harbor", + evaluator_type="harbor_native", train_dataset=DatasetRef(uri=str(train)), validation_dataset=DatasetRef(uri=str(validation)), task_template=DatasetRef(uri=str(template)), diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_factory.py b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_factory.py index eccf422cd0..d67de2e875 100644 --- a/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_factory.py +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_factory.py @@ -14,7 +14,7 @@ TrialResult, ) from nemo_experimentalist_plugin.experimentalist.components.evaluator.factory import DatasetFactory, EvaluatorFactory -from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import ( +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native import ( HarborEvaluator, HarborEvaluatorConfig, ) @@ -117,7 +117,7 @@ def test_build_dataset_falsy_evaluator_type(): def test_build_dataset_falsy_dataset_ref(): with pytest.raises(ValueError, match="Evaluator type and dataset reference are required"): - DatasetFactory().build_dataset("harbor", None) + DatasetFactory().build_dataset("harbor_native", None) def test_build_task_template_zero_tasks(tmp_path): @@ -125,7 +125,7 @@ def test_build_task_template_zero_tasks(tmp_path): empty_dir.mkdir() (empty_dir / "not-a-task").mkdir() with pytest.raises(ValueError, match="contains no Harbor task directories"): - DatasetFactory().build_task_template("harbor", DatasetRef(uri=str(empty_dir))) + DatasetFactory().build_task_template("harbor_native", DatasetRef(uri=str(empty_dir))) def test_build_task_template_multiple_tasks(tmp_path): @@ -134,30 +134,42 @@ def test_build_task_template_multiple_tasks(tmp_path): (dataset_dir / "task-a" / "task.toml").write_text("") (dataset_dir / "task-b").mkdir() (dataset_dir / "task-b" / "task.toml").write_text("") - with pytest.raises(ValueError, match="exactly one harbor task"): - DatasetFactory().build_task_template("harbor", DatasetRef(uri=str(dataset_dir))) + with pytest.raises(ValueError, match="exactly one harbor_native task"): + DatasetFactory().build_task_template("harbor_native", DatasetRef(uri=str(dataset_dir))) def test_build_task_template_single_task(tmp_path): task_dir = tmp_path / "task-only" task_dir.mkdir() (task_dir / "task.toml").write_text("") - task = DatasetFactory().build_task_template("harbor", DatasetRef(uri=str(task_dir))) + task = DatasetFactory().build_task_template("harbor_native", DatasetRef(uri=str(task_dir))) assert task.id == "task-only" def test_evaluator_factory_build_evaluator_with_config(): factory = EvaluatorFactory() - evaluator = factory.build_evaluator("harbor", HarborEvaluatorConfig()) + evaluator = factory.build_evaluator("harbor_native", HarborEvaluatorConfig()) assert isinstance(evaluator, HarborEvaluator) def test_evaluator_factory_build_evaluator_with_dict(): factory = EvaluatorFactory() - evaluator = factory.build_evaluator("harbor", {"import_path": "x:Y"}) + evaluator = factory.build_evaluator("harbor_native", {"import_path": "x:Y"}) assert isinstance(evaluator, HarborEvaluator) +def test_harbor_orchestrators_live_outside_the_shared_harbor_module(): + factory = EvaluatorFactory() + + native = factory.build_evaluator("harbor_native", {}) + sdk_backed = factory.build_evaluator("harbor_evaluator", {}) + + assert type(native).__module__ == ("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native") + assert type(sdk_backed).__module__ == ( + "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_evaluator" + ) + + def test_evaluator_factory_build_evaluator_unsupported(): factory = EvaluatorFactory() with pytest.raises(ValueError, match="Unsupported evaluator type"): @@ -166,5 +178,5 @@ def test_evaluator_factory_build_evaluator_unsupported(): def test_evaluator_factory_build_evaluator_wrong_config_type(): factory = EvaluatorFactory() - with pytest.raises(TypeError, match="Harbor evaluator config must be an EvaluatorConfig or dict"): - factory.build_evaluator("harbor", 42) + with pytest.raises(TypeError, match="'harbor_native' evaluator config must be an EvaluatorConfig or dict"): + factory.build_evaluator("harbor_native", 42) diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py index 21712d477b..625d95ab7f 100644 --- a/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor.py @@ -27,25 +27,28 @@ subset_dataset_id, ) from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import ( - _TRACE_ARTIFACT_DESTINATION, - _TRACE_ARTIFACT_SOURCE, + DEFAULT_TRACE_ARTIFACT_SOURCE, HarborDataset, HarborDependencyContext, HarborDependencyRuntime, - HarborEvaluator, - HarborEvaluatorConfig, HarborVerifierValidationError, _chmod_path_chain, - _cleanup_scoped_imports, - _ensure_package, _python_syntax_failure, - _safe_identifier, - _scoped_import_path, _shell_syntax_failure, _trial_error, _trial_metric_spec, _trial_metrics, _trial_resources, +) +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native import ( + _TRACE_ARTIFACT_DESTINATION, + HarborEvaluator, + HarborEvaluatorConfig, + _cleanup_scoped_imports, + _ensure_package, + _safe_identifier, + _scoped_import_path, + _validated_job_dir, _with_trace_artifact, ) @@ -741,7 +744,7 @@ async def create(cls, config): async def run(self): return SimpleNamespace(id="job-id", stats=FakeStats()) - monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", FakeJob) + monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.Job", FakeJob) result = await evaluator.run( agent=tmp_path / "agent", @@ -866,7 +869,7 @@ async def test_harbor_evaluator_rejects_invalid_python_verifiers_before_job_crea dataset = HarborDataset.from_path(dataset_dir) fake_job = _recording_job(tmp_path / "jobs" / "preflight") - monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", fake_job) + monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.Job", fake_job) compile_calls = 0 original_compile = compile @@ -987,7 +990,7 @@ async def test_harbor_evaluator_accepts_valid_python_verifier( _write(task_dir / "tests" / "check.py", "def check():\n return True\n") 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) + monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.Job", fake_job) trials = await HarborEvaluator()._run(agent_dir, dataset, HarborEvaluatorConfig()) @@ -1009,7 +1012,7 @@ async def test_harbor_evaluator_rejects_invalid_configured_test_sh_before_job_cr _write(task_dir / "test" / "test.sh", "if true; then\n echo broken\n") 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) + monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.Job", fake_job) with pytest.raises(HarborVerifierValidationError) as exc_info: await HarborEvaluator()._run(agent_dir, dataset, HarborEvaluatorConfig()) @@ -1034,7 +1037,7 @@ async def test_harbor_evaluator_accepts_valid_legacy_test_sh( _write(task_dir / "test" / "test.sh", "if true; then\n echo valid\nfi\n") 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) + monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.Job", fake_job) trials = await HarborEvaluator()._run(agent_dir, dataset, HarborEvaluatorConfig()) @@ -1352,8 +1355,8 @@ def test_trial_metric_spec_task_dir_verifier(tmp_path): def test_with_trace_artifact_already_has_source(): from harbor.models.job.config import ArtifactConfig - existing = ArtifactConfig(source=_TRACE_ARTIFACT_SOURCE, destination="traces") - result = _with_trace_artifact([existing], _TRACE_ARTIFACT_SOURCE) + existing = ArtifactConfig(source=DEFAULT_TRACE_ARTIFACT_SOURCE, destination="traces") + result = _with_trace_artifact([existing], DEFAULT_TRACE_ARTIFACT_SOURCE) assert result == [existing] @@ -1361,7 +1364,7 @@ def test_with_trace_artifact_already_has_destination(): from harbor.models.job.config import ArtifactConfig existing = ArtifactConfig(source="/other", destination=_TRACE_ARTIFACT_DESTINATION) - result = _with_trace_artifact([existing], _TRACE_ARTIFACT_SOURCE) + result = _with_trace_artifact([existing], DEFAULT_TRACE_ARTIFACT_SOURCE) assert result == [existing] @@ -1369,9 +1372,9 @@ def test_with_trace_artifact_adds_when_missing(): from harbor.models.job.config import ArtifactConfig other = ArtifactConfig(source="/other", destination="other") - result = _with_trace_artifact([other], _TRACE_ARTIFACT_SOURCE) + result = _with_trace_artifact([other], DEFAULT_TRACE_ARTIFACT_SOURCE) assert len(result) == 2 - assert result[0].source == _TRACE_ARTIFACT_SOURCE + assert result[0].source == DEFAULT_TRACE_ARTIFACT_SOURCE def test_trial_error_non_dict(): @@ -1426,7 +1429,7 @@ def fake_rmtree(path, **kwargs): rmtree_calls.append(path) monkeypatch.setattr( - "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.shutil.rmtree", fake_rmtree + "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.shutil.rmtree", fake_rmtree ) class FakeJob: @@ -1444,7 +1447,7 @@ async def create(cls, config): async def run(self): return SimpleNamespace(id="job-id", stats=None) - monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor.Job", FakeJob) + monkeypatch.setattr("nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.Job", FakeJob) trials = await evaluator._run( agent=agent_dir, @@ -1457,7 +1460,60 @@ async def run(self): ) assert trials == [] assert len(rmtree_calls) == 1 - assert rmtree_calls[0] == job_dir + assert rmtree_calls[0] == job_dir.resolve() + + +def test_validated_job_dir_rejects_parent_escape(tmp_path: Path) -> None: + jobs_dir = tmp_path / "jobs" + jobs_dir.mkdir() + with pytest.raises(ValueError, match="strict descendant"): + _validated_job_dir(jobs_dir, "../outside") + + +@pytest.mark.asyncio +async def test_harbor_evaluator_force_rerun_rejects_job_name_outside_jobs_dir( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + evaluator = HarborEvaluator() + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + task_dir = tmp_path / "dataset" / "task-a" + _write(task_dir / "task.toml", "") + harbor_dataset = HarborDataset.from_path(task_dir.parent) + + outside = tmp_path / "outside" + outside.mkdir() + marker = outside / "keep.txt" + marker.write_text("do not delete", encoding="utf-8") + + rmtree_calls: list[Path] = [] + + def fake_rmtree(path: Path, **kwargs: object) -> None: + rmtree_calls.append(path) + + monkeypatch.setattr( + "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.shutil.rmtree", + fake_rmtree, + ) + monkeypatch.setattr( + "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.Job", + object, + ) + + with pytest.raises(ValueError, match="strict descendant"): + await evaluator._run( + agent=agent_dir, + dataset=harbor_dataset, + options=HarborEvaluatorConfig( + import_path="harbor_wrapper:WrappedAgent", + jobs_dir=tmp_path / "jobs", + job_name="../outside", + force_rerun=True, + ), + ) + + assert rmtree_calls == [] + assert marker.read_text(encoding="utf-8") == "do not delete" @pytest.mark.asyncio @@ -1905,12 +1961,12 @@ def test_resolve_trial_task_id_fallback_to_trial_base(): def test_with_trace_artifact_string_match(): - result = _with_trace_artifact([_TRACE_ARTIFACT_SOURCE], _TRACE_ARTIFACT_SOURCE) - assert result == [_TRACE_ARTIFACT_SOURCE] + result = _with_trace_artifact([DEFAULT_TRACE_ARTIFACT_SOURCE], DEFAULT_TRACE_ARTIFACT_SOURCE) + assert result == [DEFAULT_TRACE_ARTIFACT_SOURCE] def test_with_trace_artifact_string_destination_match(): - result = _with_trace_artifact([_TRACE_ARTIFACT_DESTINATION], _TRACE_ARTIFACT_SOURCE) + result = _with_trace_artifact([_TRACE_ARTIFACT_DESTINATION], DEFAULT_TRACE_ARTIFACT_SOURCE) assert result == [_TRACE_ARTIFACT_DESTINATION] diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor_evaluator.py b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor_evaluator.py new file mode 100644 index 0000000000..dbfbf5dc73 --- /dev/null +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_evaluator_harbor_evaluator.py @@ -0,0 +1,719 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Regression coverage for the SDK-backed Harbor evaluator. + +Nothing here starts Docker: Harbor's ``Job`` is replaced with a fake that writes +the same on-disk tree a real run would, which is exactly the seam both evaluator +types read their results from. +""" + +from __future__ import annotations + +import inspect +import json +import logging +import sys +from pathlib import Path +from typing import Any + +import pytest +from nemo_experimentalist_plugin.config import EvolutionaryOptimizerConfig +from nemo_experimentalist_plugin.entities import local_path_from_uri +from nemo_experimentalist_plugin.experimentalist.components.evaluator.base import ( + EvaluatorConfig, + _warned_evaluator_types, +) +from nemo_experimentalist_plugin.experimentalist.components.evaluator.factory import EvaluatorFactory +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import HarborDataset +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_evaluator import ( + HarborRunnerConfig, + HarborRunnerEvaluator, + HarborTaskNameError, + harbor_task_names, +) +from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native import ( + HarborEvaluator, + HarborEvaluatorConfig, +) +from nemo_experimentalist_plugin.experimentalist.deps import ExperimentalistDeps +from pydantic import ValidationError + + +def _write(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + +pytestmark = pytest.mark.asyncio + + +def _dataset_root(dataset: HarborDataset) -> Path: + assert dataset.source is not None + return local_path_from_uri(dataset.source.uri, context="dataset").resolve() + + +def _write_task(task_dir: Path, full_name: str | None = None) -> None: + """Write a minimal Harbor task whose ``[task].name`` differs from its directory.""" + name_block = f'\n[task]\nname = "{full_name}"\n' if full_name is not None else "" + _write(task_dir / "task.toml", f'schema_version = "1.3"\n{name_block}') + _write(task_dir / "instruction.md", f"do {task_dir.name}") + _write(task_dir / "tests" / "test.sh", "echo reward") + + +def _write_trial( + job_dir: Path, + *, + trial_name: str, + task_name: str, + task_dir: Path, + rewards: dict[str, float] | None = None, + exception_info: dict[str, str] | None = None, +) -> None: + trial_dir = job_dir / trial_name + _write( + trial_dir / "result.json", + json.dumps( + { + "trial_name": trial_name, + "task_name": task_name, + "task_id": {"path": str(task_dir.resolve())}, + "verifier_result": {"rewards": rewards if rewards is not None else {}}, + "exception_info": exception_info, + } + ), + ) + + +class _FakeJob: + """Stand-in for Harbor's ``Job`` that records its config and writes trials.""" + + calls: list[Any] = [] + on_run: Any = None + + def __init__(self, config: Any) -> None: + self.config = config + + @classmethod + async def create(cls, config: Any) -> _FakeJob: + cls.calls.append(config) + # Harbor's Job.create lays down the job dir before any trial runs. + (Path(config.jobs_dir) / config.job_name).mkdir(parents=True, exist_ok=True) + return cls(config) + + async def run(self) -> None: + if type(self).on_run is not None: + type(self).on_run(self.config) + + +@pytest.fixture +def fake_job(monkeypatch: pytest.MonkeyPatch) -> type[_FakeJob]: + """Replace Harbor's ``Job`` for the duration of a test. + + The SDK imports ``Job`` inside its ``run_job`` closure, so patching the + module attribute is enough — and it is the only Harbor piece faked, so the + real ``JobConfig`` still validates everything the runtime builds. + """ + pytest.importorskip("harbor") + import harbor.job + + _FakeJob.calls = [] + _FakeJob.on_run = None + monkeypatch.setattr(harbor.job, "Job", _FakeJob) + return _FakeJob + + +@pytest.fixture +def dataset(tmp_path: Path) -> HarborDataset: + """Two tasks whose full Harbor names are namespaced and share a basename prefix.""" + dataset_dir = tmp_path / "dataset" / "validation" + _write_task(dataset_dir / "sum-two", "hello/sum-two") + _write_task(dataset_dir / "sum-three", "hello/sum-three") + return HarborDataset.from_path(dataset_dir) + + +@pytest.fixture +async def cached_job_dir( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + fake_job: type[_FakeJob], +) -> Path: + """A complete, all-successful cached job dir left by a genuine prior run. + + Driven through ``_run`` rather than hand-built so the SDK stamps its own cache + key exactly as it would in production. Hand-stamping here would couple the test + to the plugin-config → ``HarborRuntimeConfig`` mapping, and an *unstamped* dir is + correctly untrusted — which would make every test using this fixture pass for + the wrong reason. + """ + job_dir = tmp_path / "jobs" / f"{agent_dir.name}-{dataset.id}" + + def write_complete_results(config: Any) -> None: + for task in dataset.tasks: + _write_trial( + Path(config.jobs_dir) / config.job_name, + trial_name=f"{task.id}__0", + task_name=f"hello/{task.id}", + task_dir=_dataset_root(dataset) / task.id, + rewards={"reward": 1.0}, + ) + + fake_job.on_run = write_complete_results + await HarborRunnerEvaluator(experiment_dir=tmp_path)._run( + agent_dir, dataset, HarborRunnerConfig(jobs_dir=Path("jobs")) + ) + fake_job.calls = [] + fake_job.on_run = None + return job_dir + + +@pytest.fixture +def agent_dir(tmp_path: Path) -> Path: + path = tmp_path / "agents" / "agent-0" + _write(path / "harbor_wrapper.py", "class WrappedAgent: ...\n") + return path + + +# -------------------------------------------------------------------------- +# Factory and configuration +# -------------------------------------------------------------------------- + + +async def test_optimizer_config_defaults_to_native_harbor() -> None: + assert EvolutionaryOptimizerConfig().evaluator_type == "harbor_native" + + +async def test_deps_default_matches_the_optimizer_config_default() -> None: + """The two defaults must not drift: run.py threads one into the other.""" + assert ( + ExperimentalistDeps.model_fields["evaluator_type"].default + == EvolutionaryOptimizerConfig.model_fields["evaluator_type"].default + == "harbor_native" + ) + + +async def test_optimizer_config_still_accepts_plain_harbor() -> None: + """Plain Harbor stays selectable — it is the A/B baseline.""" + config = EvolutionaryOptimizerConfig.model_validate({"evaluator_type": "harbor_native"}) + assert config.evaluator_type == "harbor_native" + + +async def test_retired_harbor_spelling_still_resolves_and_warns(caplog: pytest.LogCaptureFixture) -> None: + # `harbor` shipped before the rename, so experiment YAMLs in the wild are pinned + # to it. Those configs must keep running, and the operator must be told once. + _warned_evaluator_types.clear() + with caplog.at_level(logging.WARNING): + config = EvolutionaryOptimizerConfig.model_validate({"evaluator_type": "harbor"}) + again = EvolutionaryOptimizerConfig.model_validate({"evaluator_type": "harbor"}) + + assert config.evaluator_type == "harbor_native" + assert again.evaluator_type == "harbor_native", "the alias must keep resolving, not just the first time" + assert caplog.text.count("is deprecated") == 1, "a pinned config must not warn once per round" + + +async def test_retired_spelling_is_not_extended_to_the_never_shipped_name() -> None: + # `harbor_agent_task_runner` only ever existed on an unmerged branch, so nothing + # can be pinned to it. Accepting it would advertise a name we never released. + with pytest.raises(ValidationError) as excinfo: + EvolutionaryOptimizerConfig.model_validate({"evaluator_type": "harbor_agent_task_runner"}) + + # Pin the error *location*, not just the type: a BeforeValidator that raised for + # some unrelated field would otherwise satisfy this test. + assert [error["loc"] for error in excinfo.value.errors()] == [("evaluator_type",)] + + +async def test_job_name_comes_from_the_resolved_agent_dir( + tmp_path: Path, dataset: HarborDataset, monkeypatch: pytest.MonkeyPatch +) -> None: + """`job_name` is the cache identity, so it must not depend on how the path is spelled. + + `Path(".").name` is empty, so deriving the name from the caller's spelling makes + every `--agent .` run collide on one job dir no matter which directory it points + at — and the SDK's scoped import derives its package name from the *resolved* + directory, so the two identities would disagree. + """ + from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import resolve_harbor_run_inputs + + job_names: list[str] = [] + for agent_name in ("agent-a", "agent-b"): + agent = tmp_path / "agents" / agent_name + _write(agent / "harbor_wrapper.py", "class WrappedAgent: ...\n") + monkeypatch.chdir(agent) + inputs = await resolve_harbor_run_inputs(Path("."), dataset, HarborRunnerConfig(), tmp_path) + job_names.append(inputs.job_name) + + assert job_names == [f"agent-a-{dataset.id}", f"agent-b-{dataset.id}"] + assert len(set(job_names)) == 2, "two different agents must not share one job dir" + + +async def test_job_name_survives_a_symlinked_agent_dir( + tmp_path: Path, dataset: HarborDataset, monkeypatch: pytest.MonkeyPatch +) -> None: + # A symlink keeps its own name while resolving elsewhere. Following it keeps + # `job_name` in step with the scoped import package, which resolves too. + from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import resolve_harbor_run_inputs + + real = tmp_path / "agents" / "agent-3" + _write(real / "harbor_wrapper.py", "class WrappedAgent: ...\n") + link = tmp_path / "agents" / "current" + link.symlink_to(real, target_is_directory=True) + + inputs = await resolve_harbor_run_inputs(link, dataset, HarborRunnerConfig(), tmp_path) + + assert inputs.job_name == f"agent-3-{dataset.id}", "the job dir must follow the agent, not the alias" + assert inputs.agent_path == real.resolve() + + +async def test_eval_author_default_tracks_the_experimentalist_default() -> None: + """The two plugins must not disagree about which adapter is canonical. + + Inert today — `run_eval_author` only consults the *dataset* half of the registry + and both types map to `HarborDataset` — but it silently stops being inert the day + the two types get different Dataset classes. + """ + from nemo_eval_author_plugin.eval_author.run import run_eval_author + + assert ( + inspect.signature(run_eval_author).parameters["evaluator_type"].default + == EvolutionaryOptimizerConfig.model_fields["evaluator_type"].default + ) + + +async def test_optimizer_config_rejects_unknown_evaluator_type() -> None: + with pytest.raises(ValidationError): + EvolutionaryOptimizerConfig.model_validate({"evaluator_type": "not-an-evaluator"}) + + +async def test_factory_builds_sdk_evaluator(tmp_path: Path) -> None: + evaluator = EvaluatorFactory().build_evaluator( + "harbor_evaluator", + {"n_attempts": 3, "quiet": True}, + experiment_dir=tmp_path, + ) + + assert isinstance(evaluator, HarborRunnerEvaluator) + assert evaluator.evaluator_type == "harbor_evaluator" + assert isinstance(evaluator.options, HarborRunnerConfig) + assert evaluator.options.n_attempts == 3 + assert evaluator.experiment_dir == tmp_path + + +@pytest.mark.parametrize( + "unsupported", + [ + {"retry": {"max_retries": 2}}, # plain-Harbor RetryConfig has no SDK equivalent + {"agent_dir": "/somewhere/else"}, # always derived from the candidate + {"typo_option": 1}, + ], +) +async def test_sdk_config_rejects_unsupported_options(unsupported: dict[str, Any]) -> None: + with pytest.raises(ValidationError, match="Extra inputs are not permitted"): + HarborRunnerConfig.model_validate(unsupported) + + +async def test_plain_harbor_still_builds() -> None: + """``harbor_native`` stays constructible — it is the A/B baseline. + + This used to also assert it survived a *missing* SDK, back when the runtime was + imported lazily. That premise is gone: ``nemo-evaluator-sdk`` is a declared, + workspace-linked dependency, so it ships with the plugin and cannot be absent. + """ + assert isinstance(EvaluatorFactory().build_evaluator("harbor_native", {}), HarborEvaluator) + + +# -------------------------------------------------------------------------- +# Task-name mapping +# -------------------------------------------------------------------------- + + +async def test_short_ids_map_to_full_harbor_names(dataset: HarborDataset) -> None: + assert harbor_task_names(dataset) == { + "sum-three": "hello/sum-three", + "sum-two": "hello/sum-two", + } + + +async def test_mapping_follows_dataset_subsets(dataset: HarborDataset) -> None: + assert harbor_task_names(dataset.subset(["sum-two"])) == {"sum-two": "hello/sum-two"} + + +async def test_mapping_falls_back_to_directory_name_without_task_block(tmp_path: Path) -> None: + dataset_dir = tmp_path / "unnamed" + _write_task(dataset_dir / "plain-task", full_name=None) + + assert harbor_task_names(HarborDataset.from_path(dataset_dir)) == {"plain-task": "plain-task"} + + +async def test_duplicate_full_names_are_rejected(tmp_path: Path) -> None: + dataset_dir = tmp_path / "dupes" + _write_task(dataset_dir / "task-a", "hello/same") + _write_task(dataset_dir / "task-b", "hello/same") + + with pytest.raises(HarborTaskNameError, match="both declare"): + harbor_task_names(HarborDataset.from_path(dataset_dir)) + + +async def test_task_outside_the_dataset_directory_is_rejected(tmp_path: Path, dataset: HarborDataset) -> None: + stray_dir = tmp_path / "stray" / "sum-four" + _write_task(stray_dir, "hello/sum-four") + stray = HarborDataset.from_path(stray_dir.parent).tasks[0] + dataset.tasks.append(stray) + + with pytest.raises(HarborTaskNameError, match="was not discovered under dataset"): + harbor_task_names(dataset) + + +# -------------------------------------------------------------------------- +# Execution +# -------------------------------------------------------------------------- + + +async def test_runner_receives_expected_job_config( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + fake_job: type[_FakeJob], +) -> None: + evaluator = HarborRunnerEvaluator(experiment_dir=tmp_path) + options = HarborRunnerConfig( + jobs_dir=Path("jobs"), + n_attempts=2, + n_concurrent_trials=3, + quiet=True, + max_retries=4, + trace_dir="/app/traces", + agent_timeout_multiplier=1.5, + verifier_timeout_multiplier=2.0, + ) + + await evaluator._run(agent_dir, dataset, options) + + assert len(fake_job.calls) == 1 + config = fake_job.calls[0] + assert config.job_name == f"{agent_dir.name}-{dataset.id}" + assert config.jobs_dir == tmp_path / "jobs" + assert config.n_attempts == 2 + assert config.n_concurrent_trials == 3 + assert config.quiet is True + assert config.retry.max_retries == 4 + assert config.agent_timeout_multiplier == 1.5 + assert config.verifier_timeout_multiplier == 2.0 + + # Harbor's local-dataset filter matches directory names, not [task].name. + assert config.datasets[0].path == _dataset_root(dataset) + assert sorted(config.datasets[0].task_names) == ["sum-three", "sum-two"] + + # Traces are collected as the 'traces' artifact so the Analyzer can read them. + trace_artifacts = [a for a in config.artifacts if getattr(a, "destination", None) == "traces"] + assert [a.source for a in trace_artifacts] == ["/app/traces"] + + # The wrapper is imported out of the candidate directory under a scoped package. + import_path = config.agents[0].import_path + assert import_path.endswith(".harbor_wrapper:WrappedAgent") + assert import_path.startswith("_nemo_evaluator_harbor_agents.") + # ...and the scoped package is torn down once the run finishes. + assert not [name for name in sys.modules if name.startswith("_nemo_evaluator_harbor_agents.")] + + +async def test_complete_cached_job_is_not_rerun( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + cached_job_dir: Path, + fake_job: type[_FakeJob], +) -> None: + trials = await HarborRunnerEvaluator(experiment_dir=tmp_path)._run( + agent_dir, dataset, HarborRunnerConfig(jobs_dir=Path("jobs")) + ) + + assert fake_job.calls == [] + assert {trial.task_id for trial in trials} == {"sum-two", "sum-three"} + + +async def test_errored_cached_job_is_rerun( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + cached_job_dir: Path, + fake_job: type[_FakeJob], +) -> None: + """An errored trial must force a rerun even when the cache is otherwise valid. + + Built on the *stamped* `cached_job_dir` on purpose. A hand-rolled job dir has + no fingerprint, so it is rejected as untrusted and the run happens for that + reason instead — the assertion would then hold even if error-awareness were + completely broken. Mutating one trial in place keeps the stamp valid, so the + error is the only thing left that can trigger the rerun. + """ + errored = json.loads((cached_job_dir / "sum-three__0" / "result.json").read_text(encoding="utf-8")) + errored["exception_info"] = {"exception_type": "TimeoutError"} + _write(cached_job_dir / "sum-three__0" / "result.json", json.dumps(errored)) + + await HarborRunnerEvaluator(experiment_dir=tmp_path)._run( + agent_dir, dataset, HarborRunnerConfig(jobs_dir=Path("jobs")) + ) + + assert len(fake_job.calls) == 1, "an errored cached trial must not be served from cache" + + +async def test_under_sampled_cached_job_is_rerun( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + cached_job_dir: Path, + fake_job: type[_FakeJob], +) -> None: + await HarborRunnerEvaluator(experiment_dir=tmp_path)._run( + agent_dir, dataset, HarborRunnerConfig(jobs_dir=Path("jobs"), n_attempts=2) + ) + + assert len(fake_job.calls) == 1, "one cached attempt must not satisfy n_attempts=2" + + +async def test_force_rerun_discards_a_complete_cache( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + cached_job_dir: Path, + fake_job: type[_FakeJob], +) -> None: + trials = await HarborRunnerEvaluator(experiment_dir=tmp_path)._run( + agent_dir, dataset, HarborRunnerConfig(jobs_dir=Path("jobs"), force_rerun=True) + ) + + assert len(fake_job.calls) == 1 + assert list(cached_job_dir.glob("*/result.json")) == [], "force_rerun must clear the stale results" + assert trials == [] + + +async def test_concurrent_candidates_use_distinct_job_dirs( + tmp_path: Path, + dataset: HarborDataset, + fake_job: type[_FakeJob], +) -> None: + evaluator = HarborRunnerEvaluator(experiment_dir=tmp_path) + options = HarborRunnerConfig(jobs_dir=Path("jobs")) + for name in ("agent-0", "agent-1"): + candidate = tmp_path / "agents" / name + _write(candidate / "harbor_wrapper.py", "class WrappedAgent: ...\n") + await evaluator._run(candidate, dataset, options) + + job_names = [config.job_name for config in fake_job.calls] + assert job_names == [f"agent-0-{dataset.id}", f"agent-1-{dataset.id}"] + assert len(set(job_names)) == 2 + + +async def test_missing_agent_directory_fails_before_docker( + tmp_path: Path, + dataset: HarborDataset, + fake_job: type[_FakeJob], +) -> None: + with pytest.raises(FileNotFoundError, match="Harbor agent path not found"): + await HarborRunnerEvaluator(experiment_dir=tmp_path)._run(tmp_path / "nope", dataset, HarborRunnerConfig()) + assert fake_job.calls == [] + + +async def test_broken_verifier_fails_before_docker( + tmp_path: Path, + agent_dir: Path, + fake_job: type[_FakeJob], +) -> None: + dataset_dir = tmp_path / "broken" + _write_task(dataset_dir / "task-a", "hello/task-a") + _write(dataset_dir / "task-a" / "tests" / "test.sh", "if [ ; then\n") + + from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import ( + HarborVerifierValidationError, + ) + + with pytest.raises(HarborVerifierValidationError): + await HarborRunnerEvaluator(experiment_dir=tmp_path)._run( + agent_dir, HarborDataset.from_path(dataset_dir), HarborRunnerConfig() + ) + assert fake_job.calls == [] + + +async def test_wrong_options_type_is_rejected(tmp_path: Path, dataset: HarborDataset, agent_dir: Path) -> None: + with pytest.raises(TypeError, match="HarborRunnerConfig"): + await HarborRunnerEvaluator(experiment_dir=tmp_path)._run(agent_dir, dataset, EvaluatorConfig()) + + +# -------------------------------------------------------------------------- +# Result parity with the plain Harbor evaluator +# -------------------------------------------------------------------------- + + +async def test_both_evaluators_produce_equivalent_trials( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + fake_job: type[_FakeJob], + monkeypatch: pytest.MonkeyPatch, + comparable_trials: Any, +) -> None: + """The two orchestrators differ; the trials they hand the loop must not.""" + dataset_dir = _dataset_root(dataset) + + def write_results(config: Any) -> None: + job_dir = Path(config.jobs_dir) / config.job_name + _write_trial( + job_dir, + trial_name="sum-two__0", + task_name="hello/sum-two", + task_dir=dataset_dir / "sum-two", + rewards={"reward": 1.0, "format_ok": 1.0}, + ) + _write_trial( + job_dir, + trial_name="sum-three__0", + task_name="hello/sum-three", + task_dir=dataset_dir / "sum-three", + rewards={"reward": 0.0, "format_ok": 1.0}, + ) + + fake_job.on_run = write_results + sdk_result = await HarborRunnerEvaluator(experiment_dir=tmp_path).run( + agent_dir, dataset, HarborRunnerConfig(jobs_dir=Path("sdk-jobs")) + ) + + # The plain evaluator imports Job into its own module namespace. + class PlainJob(_FakeJob): + def __init__(self, config: Any) -> None: + super().__init__(config) + self.job_dir = Path(config.jobs_dir) / config.job_name + + monkeypatch.setattr( + "nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor_native.Job", + PlainJob, + ) + PlainJob.on_run = write_results + plain_result = await HarborEvaluator(experiment_dir=tmp_path).run( + agent_dir, dataset, HarborEvaluatorConfig(jobs_dir=Path("plain-jobs")) + ) + + assert comparable_trials(sdk_result.trials, include_id=True) == comparable_trials( + plain_result.trials, include_id=True + ) + assert sdk_result.aggregate_metrics == plain_result.aggregate_metrics + # Every verifier metric survives, not just the primary reward. + assert sdk_result.aggregate_metrics == {"reward": 0.5, "format_ok": 1.0} + + +async def test_sdk_evaluator_selects_the_configured_atif_trace( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + fake_job: type[_FakeJob], +) -> None: + selected_dataset = dataset.subset(["sum-two"]) + dataset_dir = _dataset_root(dataset) + + def write_results(config: Any) -> None: + job_dir = Path(config.jobs_dir) / config.job_name + _write_trial( + job_dir, + trial_name="sum-two__0", + task_name="hello/sum-two", + task_dir=dataset_dir / "sum-two", + rewards={"reward": 1.0}, + ) + traces = job_dir / "sum-two__0" / "artifacts" / "traces" + _write(traces / "trace.jsonl", '{"resourceSpans": []}\n') + _write( + traces / "trajectory.atif.json", + '{"schema_version": "ATIF-v1.7", "session_id": "session-1"}', + ) + + fake_job.on_run = write_results + + result = await HarborRunnerEvaluator(experiment_dir=tmp_path).run( + agent_dir, + selected_dataset, + HarborRunnerConfig(jobs_dir=Path("jobs"), trace_format="atif"), + ) + + assert len(result.trials) == 1 + trace = result.trials[0].trace + assert trace is not None + assert trace.uri.endswith("trajectory.atif.json") + assert trace.metadata["trace_format"] == "atif" + + +async def test_failed_trials_keep_their_error_shape( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + fake_job: type[_FakeJob], +) -> None: + dataset_dir = _dataset_root(dataset) + + def write_results(config: Any) -> None: + job_dir = Path(config.jobs_dir) / config.job_name + _write_trial( + job_dir, + trial_name="sum-two__0", + task_name="hello/sum-two", + task_dir=dataset_dir / "sum-two", + rewards={"reward": 1.0}, + ) + _write_trial( + job_dir, + trial_name="sum-three__0", + task_name="hello/sum-three", + task_dir=dataset_dir / "sum-three", + exception_info={"exception_type": "TimeoutError", "exception_message": "boom"}, + ) + + fake_job.on_run = write_results + trials = await HarborRunnerEvaluator(experiment_dir=tmp_path)._run( + agent_dir, dataset, HarborRunnerConfig(jobs_dir=Path("jobs")) + ) + + by_task = {trial.task_id: trial for trial in trials} + assert by_task["sum-three"].status == "failed" + assert by_task["sum-three"].error == {"type": "TimeoutError", "message": "boom"} + assert by_task["sum-two"].status == "completed" + assert by_task["sum-two"].attempt == 0 + + +# -------------------------------------------------------------------------- +# The SDK owns cache identity now — verify the plugin is actually covered by it +# -------------------------------------------------------------------------- + + +async def test_editing_the_candidate_invalidates_the_cache_through_the_sdk( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + cached_job_dir: Path, + fake_job: type[_FakeJob], +) -> None: + """The staleness guard lives in the SDK; this asserts the plugin inherits it. + + Without it, editing a candidate and re-running in the same experiment directory + silently returns the previous candidate's scores — which is the whole reason + AALGO-427 exists. + """ + _write(agent_dir / "harbor_wrapper.py", "class WrappedAgent:\n version = 2\n") + + await HarborRunnerEvaluator(experiment_dir=tmp_path)._run( + agent_dir, dataset, HarborRunnerConfig(jobs_dir=Path("jobs")) + ) + + assert len(fake_job.calls) == 1, "a changed candidate must not be served from cache" + + +async def test_unchanged_candidate_still_hits_the_cache( + tmp_path: Path, + dataset: HarborDataset, + agent_dir: Path, + cached_job_dir: Path, + fake_job: type[_FakeJob], +) -> None: + """The guard must not be so strict that it defeats caching entirely.""" + await HarborRunnerEvaluator(experiment_dir=tmp_path)._run( + agent_dir, dataset, HarborRunnerConfig(jobs_dir=Path("jobs")) + ) + + assert fake_job.calls == [] diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_loop_failure.py b/plugins/nemo-experimentalist/tests/experimentalist/test_loop_failure.py index 66d0cb2fe5..efd005c6b3 100644 --- a/plugins/nemo-experimentalist/tests/experimentalist/test_loop_failure.py +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_loop_failure.py @@ -74,7 +74,7 @@ async def test_baseline_failure_marks_run_failed(monkeypatch, tmp_path, failure_ backend=backend, workspace="default", config=config, - evaluator_type="harbor", + evaluator_type="harbor_native", train_dataset=object(), validation_dataset=object(), insight=None, diff --git a/plugins/nemo-experimentalist/tests/test_deps.py b/plugins/nemo-experimentalist/tests/test_deps.py index 21bf8bc69d..16017eed70 100644 --- a/plugins/nemo-experimentalist/tests/test_deps.py +++ b/plugins/nemo-experimentalist/tests/test_deps.py @@ -8,6 +8,7 @@ import pytest from nemo_experimentalist_plugin.entities import DatasetRef from nemo_experimentalist_plugin.experimentalist.deps import ExperimentalistDeps +from pydantic import ValidationError def _datasets(tmp_path: Path) -> dict: @@ -53,3 +54,14 @@ def test_insight_without_task_template_raises(tmp_path: Path) -> None: def test_neither_raises(tmp_path: Path) -> None: with pytest.raises(ValueError, match="must be set"): ExperimentalistDeps(**_datasets(tmp_path)) + + +def test_deprecated_evaluator_alias_is_rejected(tmp_path: Path) -> None: + with pytest.raises(ValidationError): + ExperimentalistDeps.model_validate( + { + "agent": "ssh://git@h/g/r.git@main", + "evaluator_type": "harbor", + **_datasets(tmp_path), + } + ) diff --git a/uv.lock b/uv.lock index f011e5f2c2..6ae1844d43 100644 --- a/uv.lock +++ b/uv.lock @@ -4481,6 +4481,7 @@ dependencies = [ { name = "harbor", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "httpx", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-eval-author-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, + { name = "nemo-evaluator-sdk", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-insights-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, { name = "nemo-platform-plugin", marker = "(platform_machine == 'arm64' and sys_platform == 'darwin') or (platform_machine == 'aarch64' and sys_platform == 'linux') or (platform_machine == 'x86_64' and sys_platform == 'linux')" }, @@ -4497,6 +4498,7 @@ requires-dist = [ { name = "harbor", specifier = ">=0.16" }, { name = "httpx" }, { name = "nemo-eval-author-plugin", editable = "plugins/nemo-eval-author" }, + { name = "nemo-evaluator-sdk", editable = "packages/nemo_evaluator_sdk" }, { name = "nemo-insights-plugin", editable = "plugins/nemo-insights" }, { name = "nemo-platform", editable = "packages/nemo_platform" }, { name = "nemo-platform-plugin", editable = "packages/nemo_platform_plugin" },