From fdc3712808f5eaa7e94b485292c0f06982dfa629 Mon Sep 17 00:00:00 2001 From: Nick Goncharenko Date: Tue, 28 Jul 2026 23:50:34 -0700 Subject: [PATCH 1/3] fix(evaluator-sdk): make Harbor's per-trial resume reachable with agent_dir set scoped_harbor_agent_import minted the synthetic package name with a fresh uuid4 on every call. That string lands in AgentConfig.import_path and therefore in Harbor's JobConfig, which Harbor compares field-by-field before resuming a job directory, so the comparison failed on every rerun and Harbor raised FileExistsError instead of resuming. Its per-trial resume was unreachable for any caller that sets agent_dir -- which is every Experimentalist run. Derive the suffix from a content digest of agent_dir instead. Identical contents now share a package name, so the sys.modules injection is refcounted and torn down on the last scope exit rather than the first. The digest takes the same jobs_dir exclusion the cache stamp uses, so a jobs_dir nested under agent_dir cannot shift the import path as results accumulate. Dropping the unconditional discard exposed a second problem. The SDK cache stamp is deliberately looser than Harbor's JobConfig comparison -- quiet, n_concurrent_trials and the task_names filter change the JobConfig without changing the results, and keying on them would discard completed candidates over a core-count difference -- so a directory that passes the stamp can still be refused, and Harbor refuses by raising. _build_native_job now identifies that refusal positively (bare message, no errno, naming this job dir), answers it by discarding and re-running, and logs which field forced it. Anything else propagates untouched rather than costing a job directory. Adds a drift guard that fails when a Harbor upgrade adds a compared JobConfig field the stamp does not classify, a pin on the refusal wording the predicate matches, and a real-Docker e2e for the resume this change exists to enable. Fixes AALGO-430. Refs AALGO-427. Co-Authored-By: Claude Opus 5 Signed-off-by: Nick Goncharenko --- .../agent_eval/runtimes/harbor_runtime.py | 223 +++++++-- .../tests/agent_eval/test_harbor_runtime.py | 455 +++++++++++++++++- .../agent_eval/test_harbor_runtime_e2e.py | 68 +++ 3 files changed, 710 insertions(+), 36 deletions(-) diff --git a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/harbor_runtime.py b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/harbor_runtime.py index 48b12c7a95..d219a3e9b8 100644 --- a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/harbor_runtime.py +++ b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/harbor_runtime.py @@ -47,7 +47,6 @@ from pathlib import Path from types import ModuleType from typing import Any -from uuid import uuid4 from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult from nemo_evaluator_sdk.agent_eval.scores import AgentEvalScoreStatus @@ -73,6 +72,13 @@ _AGENT_IMPORT_ROOT = "_nemo_evaluator_harbor_agents" # Guards the sys.modules mutation while injecting/removing scoped agent packages. _IMPORT_LOCK = threading.Lock() +# Open scopes per content-addressed agent package. Identical agent contents share a +# package name, so teardown must wait for the last scope rather than the first. +_AGENT_PACKAGE_REFCOUNTS: dict[str, int] = {} +# Characters of the agent-content digest used to disambiguate the package name. Long +# enough that distinct agents don't collide; short enough to keep import paths (and +# Harbor's persisted JobConfig) readable. +_IMPORT_DIGEST_CHARS = 12 # Records which inputs produced a job dir, so a rerun can tell a reusable cache from # a stale one. A file, not a directory: Harbor rmtree's stray directories in a job dir. CACHE_STAMP_FILENAME = ".nemo-eval-harbor-cache.json" @@ -82,6 +88,20 @@ _CACHE_IRRELEVANT_OPTIONS = frozenset( {"jobs_dir", "job_name", "force_rerun", "quiet", "n_concurrent_trials", "agent_dir", "reward_key"} ) +# Where Harbor persists the JobConfig it will compare a resume against. +_HARBOR_JOB_CONFIG_FILENAME = "config.json" +# Fields Harbor's own JobConfig equality ignores, so they can never be why it refused +# to resume. Pinned against Harbor upstream by the drift-guard test. +_HARBOR_EQ_IGNORED_FIELDS = frozenset({"job_name", "debug"}) +# How Harbor says "this job dir cannot be resumed": one from the JobConfig comparison +# in `Job.create`, one from the lock.json check early in `Job.run`. Matching on the +# message is deliberate coupling, and it fails in the safe direction — an +# unrecognized FileExistsError propagates untouched rather than costing a job dir, so +# a Harbor reword degrades to a loud crash, never to a silent deletion. +_HARBOR_RESUME_REFUSALS = ("resumed with a different config", "does not match the resolved job lock") +# Cap on each value rendered into the "what differed" log line: enough for a scalar +# like `n_concurrent_trials`, bounded for a whole nested `agents` list. +_DRIFT_VALUE_CHARS = 80 # Derived/VCS noise skipped when digesting a directory. Deliberately NOT skipped: # `node_modules` and other vendored dependency trees, which ship with the agent and # change what it does. `.venv`/`.uv` stay skipped because they are environment, not @@ -228,13 +248,19 @@ async def run_tasks( :func:`discover_harbor_tasks`) unless a ``dataset_path`` override was given, so callers don't repeat it. - ``job_dir`` doubles as a cache, and it is reused only when **both** hold: - every requested task already has ``n_attempts`` completed, non-errored - results there, *and* the directory carries a cache stamp matching this run's - inputs (agent contents, task contents, result-affecting options). Anything - else re-runs from scratch — the directory is discarded rather than handed to - Harbor, because a surviving directory plus a changed agent is exactly the - case Harbor itself refuses. + ``job_dir`` doubles as a cache. Results are served straight off it, without + importing Harbor at all, only when **both** hold: every requested task already + has ``n_attempts`` completed, non-errored results there, *and* the directory + carries a cache stamp matching this run's inputs (agent contents, task + contents, result-affecting options). + + Otherwise Harbor runs, and what happens to the directory depends on *which* + check failed. A **stamp mismatch** discards it first: those results came from + different inputs, so there is nothing safe to resume onto. A directory that + merely lacks **coverage** — stamp matches, but not enough completed results — + is handed to Harbor intact so its per-trial resume keeps the finished trials + and runs only what is missing. Harbor may still refuse a directory on its own + (stricter) terms; :func:`_build_native_job` then discards it and re-runs. The cache only engages when the config pins a stable ``job_name``; with the default timestamped name no fingerprint is computed at all. @@ -263,13 +289,12 @@ async def run_tasks( dataset_path, self._task_names, job_name=job_name, - # Discard when the inputs changed, and whenever `agent_dir` is - # set: Harbor bakes a fresh uuid into the scoped agent import - # path, so its own JobConfig never matches on a rerun and it - # raises FileExistsError instead of resuming. With `agent_dir` - # unset the AgentConfig is deterministic, so leaving force_rerun - # off lets Harbor resume per trial and keep completed work. - force_rerun=(self._config.force_rerun or stale or self._config.agent_dir is not None), + # Discard only when the inputs changed. Otherwise leave it off so + # Harbor resumes per trial and keeps completed work — including + # with `agent_dir` set, now that the scoped import path is + # content-addressed rather than a fresh uuid per run and Harbor's + # JobConfig comparison can therefore match (AALGO-430). + force_rerun=(self._config.force_rerun or stale), ) # Fingerprint the inputs *before* running and confirm they are # unchanged afterwards. Stamping only the post-run state would label @@ -736,15 +761,52 @@ async def _create_and_run(agent: Any) -> None: datasets=[DatasetConfig(path=dataset_path, task_names=list(task_names) if task_names else None)], **timeout_kwargs, ) - job = await Job.create(job_config) - await job.run() + + async def _attempt() -> None: + job = await Job.create(job_config) + await job.run() + + try: + await _attempt() + except FileExistsError as exc: + # Harbor refuses to resume a job dir whose persisted `config.json` or + # `lock.json` differs from this run's — and it refuses by raising, not + # by re-running. Its comparison is deliberately stricter than the SDK + # cache stamp: `quiet`, `n_concurrent_trials` and the `task_names` + # filter all change the JobConfig without changing the results, so the + # stamp excludes them (a full cache hit must not pay for a concurrency + # tweak) while Harbor still rejects the directory. Honour the intent of + # the rerun rather than propagating a crash. + # + # Identify the refusal positively before deleting anything. Both of + # Harbor's refusals fire before any trial executes, so discarding costs + # only completed work — but that reasoning holds *only* for those two. + # An ordinary "file exists" raised from inside a trial, a hook, or an + # environment build must not be mistaken for drift and answered by + # destroying the directory. + if not (job_dir.exists() and _is_harbor_resume_refusal(exc, job_dir)): + raise + drift = _describe_job_config_drift(job_dir, job_config) + logger.warning( + "Harbor refused to resume job dir %s, so it is being discarded and re-run from scratch: %s%s", + job_dir, + exc, + f" Differing config: {drift}." if drift else "", + ) + shutil.rmtree(job_dir) + await _attempt() if config.agent_import_path is None: await _create_and_run(AgentConfig(name=config.agent_name or "oracle", model_name=config.agent_model_name)) elif config.agent_dir is not None: - # Loose wrapper file: make its directory importable for the run. + # Loose wrapper file: make its directory importable for the run. The + # jobs_dir exclusion must match _cache_stamp's, or a jobs_dir nested under + # agent_dir would shift the package name as results accumulate. agent_dir = config.agent_dir.expanduser().resolve() - with scoped_harbor_agent_import(agent_dir, config.agent_import_path) as scoped_import: + excluded_roots = frozenset({config.jobs_dir.expanduser().resolve()}) + with scoped_harbor_agent_import( + agent_dir, config.agent_import_path, exclude=excluded_roots + ) as scoped_import: await _create_and_run(AgentConfig(import_path=scoped_import, model_name=config.agent_model_name)) else: # Already-importable module (installed package): let Harbor import it directly. @@ -753,13 +815,73 @@ async def _create_and_run(agent: Any) -> None: return job_dir, run_job +def _is_harbor_resume_refusal(exc: FileExistsError, job_dir: Path) -> bool: + """Return True when ``exc`` is Harbor declining to resume ``job_dir``. + + Separates Harbor's refusal — the one case where deleting the directory is the + right answer — from an ordinary "file exists" surfacing from a trial, a hook or an + environment build, where deleting it would destroy completed work to no purpose. + + Two signals, both required. Harbor constructs its refusals with a bare message, so + ``errno`` is unset, while an OS-level ``EEXIST`` always carries one; and both + refusals name the job directory and end in a known phrase. + """ + if exc.errno is not None: + return False + message = str(exc) + return str(job_dir) in message and any(phrase in message for phrase in _HARBOR_RESUME_REFUSALS) + + +def _describe_job_config_drift(job_dir: Path, job_config: Any) -> str: + """Name the fields that differ between ``job_dir``'s persisted JobConfig and this run's. + + Harbor reports *that* an existing config differs, never *which* field, which + leaves the resulting discard looking arbitrary. This reproduces enough of its + comparison to say — turning "Harbor refused" into "n_concurrent_trials: 10 -> 4". + + Best-effort by construction. Returns ``""`` when the difference cannot be + located: no ``config.json``, unparseable, or a refusal that came from + ``lock.json`` instead, which has no JobConfig difference to report. Diagnostics + must never mask the failure they explain, so every error here is swallowed. + """ + try: + stored_text = (job_dir / _HARBOR_JOB_CONFIG_FILENAME).read_text(encoding="utf-8") + # Harbor persists with exclude_defaults=True, so the JSON omits every field + # left at its default and comparing it raw would report phantom differences. + # Round-tripping through the model refills them, which is what Harbor itself + # compares after re-validating the stored config. + stored = type(job_config).model_validate_json(stored_text).model_dump() + current = job_config.model_dump() + return ", ".join( + f"{field}: {_truncated_repr(stored.get(field))} -> {_truncated_repr(value)}" + for field, value in current.items() + if field not in _HARBOR_EQ_IGNORED_FIELDS and stored.get(field) != value + ) + except Exception: + return "" + + +def _truncated_repr(value: Any) -> str: + """Render ``value`` for a log line, bounded so a nested config can't flood it.""" + text = repr(value) + return text if len(text) <= _DRIFT_VALUE_CHARS else f"{text[:_DRIFT_VALUE_CHARS]}..." + + @contextlib.contextmanager -def scoped_harbor_agent_import(agent_dir: Path, import_path: str) -> Iterator[str]: - """Make ``agent_dir`` importable under a unique synthetic package for the block. +def scoped_harbor_agent_import( + agent_dir: Path, import_path: str, *, exclude: frozenset[Path] = frozenset() +) -> Iterator[str]: + """Make ``agent_dir`` importable under a content-addressed package for the block. Args: agent_dir: directory containing the module referenced by ``import_path``. import_path: Harbor agent path, ``"module"`` or ``"module:attribute"``. + exclude: resolved directories to leave out of the content digest. Pass the + same set :func:`_cache_stamp` uses — in practice ``jobs_dir``, which is + caller-chosen and may sit *under* ``agent_dir``. Omitting it lets the + growing results tree feed the package name, so the import path would + change on every run and the resume this function exists to enable would + never happen. See :func:`_digest_directory`. Yields: str: the rewritten import path Harbor should load (the module rooted under @@ -768,9 +890,33 @@ def scoped_harbor_agent_import(agent_dir: Path, import_path: str) -> Iterator[st Raises: ValueError: if ``import_path`` has no module component. - On exit the injected ``sys.modules`` entries are removed. The mutation is - guarded by a process-wide lock so concurrent runs don't corrupt import state; - each run gets its own uniquely-named package so distinct agents never collide. + **The package name is derived from the directory's contents, not a random + UUID, and that is load-bearing.** This string becomes ``AgentConfig.import_path`` + and therefore part of Harbor's ``JobConfig``, which Harbor compares field-by-field + when deciding whether an existing job directory may be resumed. A random suffix + made that comparison fail on every rerun, so Harbor raised ``FileExistsError`` + instead of resuming and its per-trial resume was unreachable for any caller that + sets ``agent_dir`` (AALGO-430). Content-addressing keeps distinct agents isolated + while letting an unchanged agent resume — and makes an *edited* agent invalidate + the job dir on Harbor's own terms. + + Identical contents therefore share a package name, so overlapping scopes are + refcounted: the injected ``sys.modules`` entries are removed when the last + scope exits, not the first (see :func:`_uninstall_agent_package`). The mutation + is guarded by a process-wide lock. ``sys.modules`` is per-process, so concurrent + *processes* were never at risk here. + + The name is ``_``, so it tracks the directory's *location* as + well as its contents — deliberately, because an opaque hash makes every traceback + and import error unreadable. That is a narrow, knowing divergence from the cache + stamp, which excludes ``agent_dir`` so a relocated but identical agent still hits + (see :func:`_cache_stamp`). Relocating an agent while pinning the same + ``job_name`` therefore leaves the stamp valid but changes this string, and Harbor + declines to resume; :func:`_build_native_job` absorbs that into a clean re-run. + The results stay correct — it costs one repeated job. Callers that rebuild agents + under changing directory names (the Experimentalist does) are unaffected, because + the agent name feeds their ``job_name`` too, so a rename lands in a different job + dir with nothing to resume. Only ``agent_dir`` (not ``sys.path``) is made importable, so a loose wrapper must be self-contained: a single module, or one that reaches siblings via @@ -782,7 +928,10 @@ def scoped_harbor_agent_import(agent_dir: Path, import_path: str) -> Iterator[st module_name = module_name.strip().lstrip(".") if not module_name: raise ValueError("import_path must be 'module' or 'module:attribute'") - package = f"{_AGENT_IMPORT_ROOT}.{_safe_identifier(agent_dir.name)}_{uuid4().hex[:8]}" + # Hashed here rather than reused from the cache stamp: this must describe the tree + # as it is about to be imported, and the extra walk is noise next to Docker. + suffix = _digest_directory(agent_dir, exclude=exclude)[:_IMPORT_DIGEST_CHARS] + package = f"{_AGENT_IMPORT_ROOT}.{_safe_identifier(agent_dir.name)}_{suffix}" with _IMPORT_LOCK: _install_agent_package(package, agent_dir) try: @@ -802,7 +951,11 @@ def _safe_identifier(value: str) -> str: def _install_agent_package(package: str, agent_dir: Path) -> None: - """Register ``package`` (and its parents) in ``sys.modules`` rooted at ``agent_dir``.""" + """Register ``package`` (and its parents) in ``sys.modules`` rooted at ``agent_dir``. + + Refcounted: package names are content-addressed, so two overlapping scopes on the + same agent directory legitimately share one. Callers must hold ``_IMPORT_LOCK``. + """ parts = package.split(".") for idx in range(1, len(parts) + 1): name = ".".join(parts[:idx]) @@ -814,10 +967,26 @@ def _install_agent_package(package: str, agent_dir: Path) -> None: if idx > 1: setattr(sys.modules[".".join(parts[: idx - 1])], parts[idx - 1], module) sys.modules[package].__path__ = [str(agent_dir)] + # Counted only once the injection it guards has succeeded. Incrementing first + # would strand the count above zero if any step above raised — the scope never + # opens, so nothing ever decrements it, and the package could never be torn down + # again for the life of the process. + _AGENT_PACKAGE_REFCOUNTS[package] = _AGENT_PACKAGE_REFCOUNTS.get(package, 0) + 1 def _uninstall_agent_package(package: str) -> None: - """Remove ``package`` and any submodules imported through it from ``sys.modules``.""" + """Remove ``package`` and its submodules from ``sys.modules`` on the last exit. + + Tearing down on the *first* exit would break a still-open scope sharing the same + content-addressed name, so the removal waits for the refcount to reach zero. + Callers must hold ``_IMPORT_LOCK``. + """ + remaining = _AGENT_PACKAGE_REFCOUNTS.get(package, 0) - 1 + if remaining > 0: + _AGENT_PACKAGE_REFCOUNTS[package] = remaining + return + _AGENT_PACKAGE_REFCOUNTS.pop(package, None) + for name in [n for n in sys.modules if n == package or n.startswith(f"{package}.")]: sys.modules.pop(name, None) parent, _, child = package.rpartition(".") diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime.py index e0fe5567ef..882f37f438 100644 --- a/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime.py +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime.py @@ -9,7 +9,9 @@ import logging import os import sys +from collections.abc import Awaitable, Callable from pathlib import Path +from types import ModuleType import pytest from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator @@ -18,6 +20,7 @@ HarborRewardMetric, HarborRuntimeConfig, HarborTasksetLoader, + _build_native_job, build_trials_from_job_dir, discover_harbor_tasks, reward_payload_from_result, @@ -26,7 +29,7 @@ from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrialStatus from nemo_evaluator_sdk.metrics.utils import metric_type_name -from pydantic import ValidationError +from pydantic import BaseModel, ValidationError _HELLO_WORLD_DATASET = Path(__file__).resolve().parents[2] / "examples" / "harbor" / "hello_world_dataset" @@ -282,10 +285,12 @@ async def test_changed_inputs_discard_the_job_dir(tmp_path: Path, monkeypatch: p @pytest.mark.asyncio -async def test_scoped_agent_import_always_discards(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - # With agent_dir set, scoped_harbor_agent_import bakes a fresh uuid into the - # import path, so Harbor's own JobConfig never matches on a rerun and it raises - # FileExistsError instead of resuming. Discard even when the stamp matches. +async def test_under_covered_job_resumes_with_agent_dir_set(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # Regression for AALGO-430. This used to discard unconditionally: the scoped + # import path carried a fresh uuid per run, so Harbor's JobConfig never matched + # and it raised FileExistsError instead of resuming. Now the path is + # content-addressed, so an unchanged agent resumes and keeps completed Docker + # work — the same as the agent_dir-unset case below. agent_dir = tmp_path / "agent" agent_dir.mkdir() (agent_dir / "wrapper.py").write_text("x = 1\n") @@ -297,14 +302,15 @@ async def test_scoped_agent_import_always_discards(tmp_path: Path, monkeypatch: await HarborAgentTaskRunner(config=config).run_tasks([task]) - assert calls == [True] + assert calls == [False], "an unchanged agent must resume, not discard completed trials" @pytest.mark.asyncio async def test_under_covered_job_resumes_when_harbor_can(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - # Inputs unchanged, some attempts missing, and agent_dir unset — Harbor's - # AgentConfig is deterministic here, so it can resume per trial. Discarding - # would throw away completed Docker work for nothing. + # Inputs unchanged, some attempts missing, agent_dir unset — Harbor's + # AgentConfig is deterministic, so it resumes per trial. Discarding would throw + # away completed Docker work for nothing. The agent_dir-set case above now + # behaves identically (AALGO-430). config, job_dir, task = _seed_cached_job(tmp_path) config = config.model_copy(update={"n_attempts": 2}) _stamp_for(config, task, job_dir) # stamp matches the new config @@ -719,3 +725,434 @@ def exploding_readlink(*_args: object, **_kwargs: object) -> str: monkeypatch.setattr(os, "readlink", exploding_readlink) assert _digest_directory(agent) # must not raise + + +def _scoped_path(agent_dir: Path) -> str: + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import scoped_harbor_agent_import + + with scoped_harbor_agent_import(agent_dir, "wrapper:Agent") as scoped: + return scoped + + +def test_scoped_import_path_is_stable_for_unchanged_contents(tmp_path: Path) -> None: + # The import path lands in Harbor's JobConfig, which Harbor compares field-by-field + # when deciding whether a job dir may be resumed. A per-run random suffix made that + # comparison fail every time, so Harbor could never resume (AALGO-430). + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + (agent_dir / "wrapper.py").write_text("x = 1\n") + + assert _scoped_path(agent_dir) == _scoped_path(agent_dir) + + +def test_scoped_import_path_changes_when_the_agent_changes(tmp_path: Path) -> None: + # The flip side: an edited agent must NOT resume a job dir built from the old one. + # Harbor's own config check now catches that without help from the cache stamp. + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + (agent_dir / "wrapper.py").write_text("x = 1\n") + before = _scoped_path(agent_dir) + + (agent_dir / "wrapper.py").write_text("x = 2\n") + + assert _scoped_path(agent_dir) != before + + +def test_distinct_agents_do_not_share_a_scoped_package(tmp_path: Path) -> None: + # Content-addressing must not collapse different agents onto one sys.modules entry. + first = tmp_path / "a" + second = tmp_path / "b" + for path, body in ((first, "x = 1\n"), (second, "x = 2\n")): + path.mkdir() + (path / "wrapper.py").write_text(body) + + assert _scoped_path(first) != _scoped_path(second) + + +def test_overlapping_scopes_on_one_agent_survive_the_inner_exit(tmp_path: Path) -> None: + # Identical contents now share a package name, so teardown is refcounted: the inner + # scope exiting must not strip sys.modules out from under the outer one. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import scoped_harbor_agent_import + + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + (agent_dir / "wrapper.py").write_text("VALUE = 7\n") + + with scoped_harbor_agent_import(agent_dir, "wrapper:Agent") as outer: + package = outer.split(":")[0].rsplit(".", 1)[0] + with scoped_harbor_agent_import(agent_dir, "wrapper:Agent"): + pass + # Inner scope closed; the outer one is still open and must still resolve. + assert package in sys.modules + assert importlib.import_module(f"{package}.wrapper").VALUE == 7 + + assert package not in sys.modules, "the last scope to exit must clean up" + + +def test_scoped_import_teardown_is_complete_after_overlap(tmp_path: Path) -> None: + # Refcounting must not leak: no stray refcount entries or sys.modules residue. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import ( + _AGENT_IMPORT_ROOT, + _AGENT_PACKAGE_REFCOUNTS, + scoped_harbor_agent_import, + ) + + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + (agent_dir / "wrapper.py").write_text("x = 1\n") + + with scoped_harbor_agent_import(agent_dir, "wrapper:Agent"): + with scoped_harbor_agent_import(agent_dir, "wrapper:Agent"): + pass + + assert _AGENT_PACKAGE_REFCOUNTS == {} + assert not [name for name in sys.modules if name.startswith(f"{_AGENT_IMPORT_ROOT}.")] + + +def test_scoped_import_path_ignores_a_jobs_dir_nested_under_the_agent(tmp_path: Path) -> None: + # jobs_dir is caller-chosen and may sit *under* agent_dir. Without the same + # exclusion the cache stamp applies, Harbor's own results would feed the package + # name, so the import path would move every run and the resume this whole change + # exists to enable could never happen. + agent_dir = tmp_path / "agent" + jobs_dir = agent_dir / "results" + jobs_dir.mkdir(parents=True) + (agent_dir / "wrapper.py").write_text("x = 1\n") + excluded = frozenset({jobs_dir.resolve()}) + + with scoped_harbor_agent_import(agent_dir, "wrapper:Agent", exclude=excluded) as before: + pass + (jobs_dir / "trial-a").mkdir() + (jobs_dir / "trial-a" / "result.json").write_text("{}") + with scoped_harbor_agent_import(agent_dir, "wrapper:Agent", exclude=excluded) as after: + pass + + assert before == after, "accumulating results must not move the agent's import path" + + +def test_failed_scoped_import_install_does_not_wedge_the_refcount( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # The refcount is taken only once the sys.modules injection has succeeded. Taking + # it first would strand the count at 1 when the injection raises — no scope ever + # opened, so nothing decrements it, and the package could never be torn down again. + from nemo_evaluator_sdk.agent_eval.runtimes import harbor_runtime + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _AGENT_IMPORT_ROOT, _AGENT_PACKAGE_REFCOUNTS + + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + (agent_dir / "wrapper.py").write_text("x = 1\n") + + def explode(_name: str) -> ModuleType: + raise RuntimeError("synthetic package could not be built") + + with monkeypatch.context() as patched: + patched.setattr(harbor_runtime, "ModuleType", explode) + with pytest.raises(RuntimeError, match="synthetic package"): + with scoped_harbor_agent_import(agent_dir, "wrapper:Agent"): + pass + + assert _AGENT_PACKAGE_REFCOUNTS == {}, "a failed install must not leave a refcount behind" + # And a later scope must still install and then fully tear down. + with scoped_harbor_agent_import(agent_dir, "wrapper:Agent"): + pass + assert _AGENT_PACKAGE_REFCOUNTS == {} + assert not [name for name in sys.modules if name.startswith(f"{_AGENT_IMPORT_ROOT}.")] + + +class _DriftConfig(BaseModel): + """Stands in for Harbor's JobConfig: a field it ignores, one it compares, one defaulted.""" + + job_name: str = "job" + n_concurrent_trials: int = 4 + quiet: bool = True + + +def _stub_harbor(monkeypatch: pytest.MonkeyPatch, job_create: Callable[[object], Awaitable[object]]) -> None: + """Install a minimal fake ``harbor`` package so ``run_job`` can execute. + + Only the names ``_build_native_job``'s ``run_job`` imports are provided. + ``job_create`` becomes ``Job.create``; every config class is a permissive stub, + since what is under test is the control flow around Harbor, not the payload. + """ + + def _module(name: str, **attrs: object) -> None: + module = ModuleType(name) + for key, value in attrs.items(): + setattr(module, key, value) + monkeypatch.setitem(sys.modules, name, module) + + def _anything(*_args: object, **_kwargs: object) -> object: + return object() + + class _Job: + create = staticmethod(job_create) + + _module("harbor") + # JobConfig is a real model, not `_anything`: the resume-refusal path reads and + # re-validates the persisted one to report what differed. Pydantic ignores the + # kwargs _build_native_job passes that this stand-in doesn't declare. + _module("harbor.job", DatasetConfig=_anything, Job=_Job, JobConfig=_DriftConfig) + _module("harbor.models") + _module("harbor.models.job") + _module("harbor.models.job.config", RetryConfig=_anything) + _module("harbor.models.trial") + _module("harbor.models.trial.config", AgentConfig=_anything, ArtifactConfig=_anything) + + +class _FakeJob: + async def run(self) -> None: + return None + + +@pytest.mark.asyncio +async def test_harbor_refusing_to_resume_discards_and_reruns( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + # Harbor compares its whole persisted JobConfig (and lock.json) before resuming and + # raises FileExistsError on any mismatch. The SDK cache stamp is looser on purpose: + # `quiet`, `n_concurrent_trials` and the `task_names` filter change the JobConfig + # without changing the results, so a job dir that passes the stamp — and is + # therefore handed over with force_rerun=False — can still be rejected by Harbor. + # That must degrade to a clean re-run rather than crash the evaluation. + jobs_dir = tmp_path / "jobs" + job_dir = jobs_dir / "pinned" + (job_dir / "old-trial").mkdir(parents=True) + # The dir was produced on a 10-core box; this run defaults to 4. Nothing about the + # results changed, so the SDK stamp would still call it fresh — Harbor won't. + (job_dir / "config.json").write_text( + _DriftConfig(job_name="pinned", n_concurrent_trials=10).model_dump_json(exclude_defaults=True), + encoding="utf-8", + ) + dir_existed_at_attempt: list[bool] = [] + + async def create(_config: object) -> _FakeJob: + dir_existed_at_attempt.append(job_dir.exists()) + if len(dir_existed_at_attempt) == 1: + raise FileExistsError( + f"Job directory {job_dir} already exists and cannot be resumed with a different config." + ) + return _FakeJob() + + _stub_harbor(monkeypatch, create) + config = HarborRuntimeConfig(jobs_dir=jobs_dir, job_name="pinned") + _built, run_job = _build_native_job(config, tmp_path / "dataset", None, job_name="pinned", force_rerun=False) + + with caplog.at_level(logging.WARNING): + await run_job() + + assert dir_existed_at_attempt == [True, False], "the refused dir must be discarded before the retry" + assert not (job_dir / "old-trial").exists(), "the stale trial must be gone, not resumed onto" + assert "refused to resume" in caplog.text, "silently deleting completed trials must be visible" + assert "n_concurrent_trials: 10 -> 4" in caplog.text, "the warning must name what forced the discard" + + +@pytest.mark.asyncio +async def test_file_exists_error_without_a_job_dir_propagates(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # The retry is scoped to Harbor's resume refusal. A FileExistsError raised with no + # job dir to discard is something else entirely and must not be swallowed, nor + # turned into a second Docker run. + attempts: list[int] = [] + + async def create(_config: object) -> _FakeJob: + attempts.append(1) + raise FileExistsError("something unrelated") + + _stub_harbor(monkeypatch, create) + config = HarborRuntimeConfig(jobs_dir=tmp_path / "jobs", job_name="pinned") + _built, run_job = _build_native_job(config, tmp_path / "dataset", None, job_name="pinned", force_rerun=False) + + with pytest.raises(FileExistsError, match="something unrelated"): + await run_job() + + assert attempts == [1], "an unrelated FileExistsError must not be retried" + + +@pytest.mark.asyncio +async def test_unrelated_file_exists_error_mid_run_leaves_the_job_dir_alone( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # The dangerous shape: a job dir that *does* exist, and a FileExistsError raised + # from inside the run rather than by Harbor's resume check — a trial, a hook, an + # environment build. Treating that as drift would delete completed work and re-run + # for an error that has nothing to do with the config. + jobs_dir = tmp_path / "jobs" + job_dir = jobs_dir / "pinned" + (job_dir / "finished-trial").mkdir(parents=True) + attempts: list[int] = [] + + class _ExplodingJob: + async def run(self) -> None: + attempts.append(1) + raise FileExistsError(17, "File exists", str(tmp_path / "scratch" / "artifact.tar")) + + async def create(_config: object) -> _ExplodingJob: + return _ExplodingJob() + + _stub_harbor(monkeypatch, create) + config = HarborRuntimeConfig(jobs_dir=jobs_dir, job_name="pinned") + _built, run_job = _build_native_job(config, tmp_path / "dataset", None, job_name="pinned", force_rerun=False) + + with pytest.raises(FileExistsError): + await run_job() + + assert attempts == [1], "an unrelated failure must not be retried" + assert (job_dir / "finished-trial").exists(), "completed work must survive an error that is not resume drift" + + +@pytest.mark.parametrize( + ("message", "errno", "expected"), + [ + ("Job directory {job_dir} already exists and cannot be resumed with a different config.", None, True), + ("Job directory {job_dir} already has a lock.json that does not match the resolved job lock.", None, True), + # Same words, but an OS-level EEXIST: errno is set, so it is not Harbor's refusal. + ("Job directory {job_dir} already exists and cannot be resumed with a different config.", 17, False), + # A refusal naming a *different* job dir is not ours to act on. + ("Job directory /somewhere/else already exists and cannot be resumed with a different config.", None, False), + ("[Errno 17] File exists: '{job_dir}/trial/artifact.tar'", None, False), + ], +) +def test_only_harbors_resume_refusal_authorises_deleting_the_job_dir( + tmp_path: Path, message: str, errno: int | None, expected: bool +) -> None: + # Deleting a job dir is the one irreversible thing this runtime does, so the + # predicate that authorises it is pinned directly. Anything unrecognised must + # answer False and let the error propagate — the safe direction if Harbor rewords. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _is_harbor_resume_refusal + + job_dir = tmp_path / "jobs" / "pinned" + rendered = message.format(job_dir=job_dir) + exc = FileExistsError(rendered) if errno is None else FileExistsError(errno, "File exists", rendered) + + assert _is_harbor_resume_refusal(exc, job_dir) is expected + + +def test_job_config_drift_names_the_field_that_forced_the_discard(tmp_path: Path) -> None: + # Harbor says only *that* a config differs, so the discard looks arbitrary in the + # log. This pins three things at once: the differing field is named, a field Harbor + # ignores is not, and a field left at its default is not — the last only holds + # because the persisted JSON (written with exclude_defaults=True) is re-validated + # rather than compared raw, which would see a missing key as a difference. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _describe_job_config_drift + + job_dir = tmp_path / "job" + job_dir.mkdir() + stored = _DriftConfig(job_name="pinned", n_concurrent_trials=10) + (job_dir / "config.json").write_text(stored.model_dump_json(exclude_defaults=True), encoding="utf-8") + + drift = _describe_job_config_drift(job_dir, _DriftConfig(job_name="renamed", n_concurrent_trials=4)) + + assert drift == "n_concurrent_trials: 10 -> 4" + + +def test_job_config_drift_is_silent_when_it_cannot_tell(tmp_path: Path) -> None: + # No config.json is the lock.json-refusal case: there is no JobConfig difference to + # report. Diagnostics must degrade to silence, never to a raised exception that + # would mask the FileExistsError being explained. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _describe_job_config_drift + + job_dir = tmp_path / "job" + job_dir.mkdir() + + assert _describe_job_config_drift(job_dir, _DriftConfig()) == "" + (job_dir / "config.json").write_text("{not json", encoding="utf-8") + assert _describe_job_config_drift(job_dir, _DriftConfig()) == "" + + +# Harbor JobConfig field -> the HarborRuntimeConfig fields that feed it. Each of these +# must sit inside the cache fingerprint: if one drops out, the stamp could call a job +# dir reusable that Harbor will then reject. `agents` also carries the agent contents, +# which the stamp digests separately (that is why `agent_dir` itself is irrelevant). +_STAMP_COVERED_HARBOR_FIELDS = { + "n_attempts": {"n_attempts"}, + "artifacts": {"artifacts", "trace_dir"}, + "retry": {"max_retries"}, + "agents": {"agent_name", "agent_import_path", "agent_model_name"}, + "timeout_multiplier": {"timeout_multiplier"}, + "agent_timeout_multiplier": {"agent_timeout_multiplier"}, + "verifier_timeout_multiplier": {"verifier_timeout_multiplier"}, + "agent_setup_timeout_multiplier": {"agent_setup_timeout_multiplier"}, + "environment_build_timeout_multiplier": {"environment_build_timeout_multiplier"}, +} +# Left at Harbor's defaults by _build_native_job, so two SDK-built configs can never +# disagree on them. (A dir written by the Harbor CLI could, but it carries no SDK cache +# stamp, so it is stale and gets discarded before Harbor ever sees it.) +_SDK_NEVER_SETS = {"install_only", "environment", "verifier", "metrics", "tasks", "extra_instruction_paths"} +# Compared by Harbor, deliberately *not* keyed by the SDK stamp. Harbor asks "can I +# resume this directory?"; the stamp asks "did these inputs produce these results?". +# Where the answers diverge, _build_native_job absorbs Harbor's refusal. +_KNOWINGLY_LOOSER = { + "jobs_dir": "implied by having found the job dir at all", + "n_concurrent_trials": "scheduling only; keying it would discard a cached run on a box with a different core count", + "quiet": "display only; changes nothing about the results", + "datasets": "`path` is covered by the per-task digests; the `task_names` filter is left unkeyed so a subset of a " + "cached job still hits (see _stamp_coverage)", +} + + +def test_harbor_still_words_its_resume_refusals_the_way_we_match_them() -> None: + # The predicate that authorises deleting a job dir keys off Harbor's message text. + # If Harbor rewords, the predicate stops matching and the refusal propagates as a + # crash — the safe direction, but a silent loss of the graceful re-run. Catch that + # at upgrade time here instead of in someone's failed experiment. + pytest.importorskip("harbor.job", reason="harbor needs python >= 3.12") + import inspect + + from harbor.job import Job + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _HARBOR_RESUME_REFUSALS + + source = inspect.getsource(Job) + for phrase in _HARBOR_RESUME_REFUSALS: + assert phrase in source, ( + f"Harbor no longer raises its resume refusal with {phrase!r}. Re-read Job._maybe_init_existing_job " + "and Job._write_job_lock, then update _HARBOR_RESUME_REFUSALS. Keep each phrase inside a single " + "source string literal — one spanning an implicit concatenation will not be found here." + ) + + +def test_harbor_job_config_equality_still_behaves_as_the_retry_assumes() -> None: + # The FileExistsError retry exists because Harbor compares its whole JobConfig and + # ignores only identity/logging fields. Pin that behaviourally, so a Harbor upgrade + # that changes the rule surfaces here rather than as a mystery re-run in production. + job_config = pytest.importorskip("harbor.models.job.config", reason="harbor needs python >= 3.12") + + baseline = job_config.JobConfig(job_name="a") + assert baseline == job_config.JobConfig(job_name="b"), "job_name must stay outside Harbor's comparison" + assert baseline != job_config.JobConfig(job_name="a", n_concurrent_trials=99), ( + "n_concurrent_trials must stay inside it — that is the case the retry absorbs" + ) + + +def test_every_harbor_job_config_field_is_classified_against_the_sdk_stamp() -> None: + # Drift guard. The SDK's fingerprint is deliberately looser than Harbor's + # comparison, but only in ways we have reasoned about. A Harbor upgrade that adds a + # compared field would silently widen that gap into unexplained full re-runs, so + # every field must land in exactly one bucket before it can ship. + job_config = pytest.importorskip("harbor.models.job.config", reason="harbor needs python >= 3.12") + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import ( + _CACHE_IRRELEVANT_OPTIONS, + _HARBOR_EQ_IGNORED_FIELDS, + ) + + # Derived, not hardcoded: dropping a field into _CACHE_IRRELEVANT_OPTIONS re-checks + # here instead of quietly diverging from a copied list. + fingerprinted = set(HarborRuntimeConfig.model_fields) - set(_CACHE_IRRELEVANT_OPTIONS) + for harbor_field, sdk_fields in _STAMP_COVERED_HARBOR_FIELDS.items(): + missing = sdk_fields - fingerprinted + assert not missing, ( + f"Harbor compares {harbor_field!r}, but {sorted(missing)} left the cache fingerprint. " + "Either restore it, or move the field to _KNOWINGLY_LOOSER with a reason." + ) + + classified = ( + set(_HARBOR_EQ_IGNORED_FIELDS) | _SDK_NEVER_SETS | set(_STAMP_COVERED_HARBOR_FIELDS) | set(_KNOWINGLY_LOOSER) + ) + actual = set(job_config.JobConfig.model_fields) + assert not actual - classified, ( + f"Harbor's JobConfig grew {sorted(actual - classified)}. Classify each one: covered by the cache stamp " + "(_STAMP_COVERED_HARBOR_FIELDS), never set by the SDK (_SDK_NEVER_SETS), or knowingly unkeyed " + "(_KNOWINGLY_LOOSER, with a reason)." + ) + assert not classified - actual, ( + f"{sorted(classified - actual)} no longer exist on Harbor's JobConfig; drop them from the classification." + ) diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime_e2e.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime_e2e.py index 5d6266ab6b..c726ecf9fa 100644 --- a/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime_e2e.py +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime_e2e.py @@ -28,6 +28,27 @@ _DATASET_DIR = Path(__file__).resolve().parents[2] / "examples" / "harbor" / "hello_world_dataset" _TASK_NAME = "harbor/hello-world" +# Minimal BaseAgent for the resume probe. Satisfies the bundled hello-world verifier, +# which passes iff /app/hello.txt contains exactly "Hello, world!". +_RESUME_PROBE_AGENT = """\ +from harbor import BaseAgent + + +class WrappedAgent(BaseAgent): + @staticmethod + def name() -> str: + return "resume-probe" + + def version(self) -> str | None: + return "1.0.0" + + async def setup(self, environment) -> None: + return None + + async def run(self, instruction, environment, context) -> None: + await environment.exec("printf 'Hello, world!' > /app/hello.txt") +""" + def _docker_available() -> bool: if shutil.which("docker") is None: @@ -66,3 +87,50 @@ async def test_sdk_runs_harbor_hello_world_natively(tmp_path: Path) -> None: payload = reward_payload_from_result(result) assert payload["reward"]["harbor_reward.reward"] == 1.0 assert payload["exceptions"] == {} + + +@pytest.mark.asyncio +async def test_harbor_resumes_a_partial_job_with_a_custom_agent_dir(tmp_path: Path) -> None: + """Regression for AALGO-430 — a real Harbor resume with ``agent_dir`` set. + + This is the case every faked-``Job`` test misses, and the reason the bug went + unnoticed: the scoped agent import path used to carry a fresh uuid per run, so + Harbor's ``JobConfig`` comparison failed on the second call and it raised + ``FileExistsError`` rather than resuming. Now the path is content-addressed, so + an unchanged agent resumes and only the missing trial is re-run. + """ + pytest.importorskip("harbor") + if not _docker_available(): + pytest.skip("Docker daemon is required to run a Harbor job") + + # A loose wrapper file next to the dataset — the shape `agent_dir` exists for, + # and the shape the Experimentalist always uses. + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + (agent_dir / "harbor_wrapper.py").write_text(_RESUME_PROBE_AGENT, encoding="utf-8") + + jobs_dir = tmp_path / "jobs" + config = HarborRuntimeConfig( + jobs_dir=jobs_dir, + job_name="resume-job", # pinned: the cache and Harbor's resume both need it + agent_import_path="harbor_wrapper:WrappedAgent", + agent_dir=agent_dir, + ) + + first = await run_harbor_eval(config, _DATASET_DIR) + assert [trial.status for trial in first.trials] == [AgentEvalTrialStatus.COMPLETED] + + job_dir = jobs_dir / "resume-job" + trial_dirs = {path.parent.name for path in job_dir.glob("*/result.json")} + assert trial_dirs, "the first run must have written a trial" + + # Drop the only trial's result so the job is under-covered, leaving the job dir + # (and its config.json) in place — the exact state that used to raise. + for path in job_dir.glob("*/result.json"): + path.unlink() + + second = await run_harbor_eval(config, _DATASET_DIR) + + assert [trial.status for trial in second.trials] == [AgentEvalTrialStatus.COMPLETED] + assert second.trials[0].task_id == _TASK_NAME + assert second.trials[0].metadata["reward"] == 1.0 From e2a1b6e9201078ca5dadee87c5bbd882e8f9a200 Mon Sep 17 00:00:00 2001 From: Nick Goncharenko Date: Tue, 28 Jul 2026 23:50:42 -0700 Subject: [PATCH 2/3] chore(sdk): re-vendor nemo_evaluator_sdk after the Harbor resume fix Generated by `make vendor`; import-path rewrites only, no hand edits. Refs AALGO-430. Co-Authored-By: Claude Opus 5 Signed-off-by: Nick Goncharenko --- .../agent_eval/runtimes/harbor_runtime.py | 223 +++++++++++++++--- 1 file changed, 196 insertions(+), 27 deletions(-) diff --git a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/harbor_runtime.py b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/harbor_runtime.py index 07fae46200..09d31c913e 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/harbor_runtime.py +++ b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/harbor_runtime.py @@ -47,7 +47,6 @@ from pathlib import Path from types import ModuleType from typing import Any -from uuid import uuid4 from nemo_platform.beta.evaluator.agent_eval.results import AgentEvalResult from nemo_platform.beta.evaluator.agent_eval.scores import AgentEvalScoreStatus @@ -73,6 +72,13 @@ _AGENT_IMPORT_ROOT = "_nemo_evaluator_harbor_agents" # Guards the sys.modules mutation while injecting/removing scoped agent packages. _IMPORT_LOCK = threading.Lock() +# Open scopes per content-addressed agent package. Identical agent contents share a +# package name, so teardown must wait for the last scope rather than the first. +_AGENT_PACKAGE_REFCOUNTS: dict[str, int] = {} +# Characters of the agent-content digest used to disambiguate the package name. Long +# enough that distinct agents don't collide; short enough to keep import paths (and +# Harbor's persisted JobConfig) readable. +_IMPORT_DIGEST_CHARS = 12 # Records which inputs produced a job dir, so a rerun can tell a reusable cache from # a stale one. A file, not a directory: Harbor rmtree's stray directories in a job dir. CACHE_STAMP_FILENAME = ".nemo-eval-harbor-cache.json" @@ -82,6 +88,20 @@ _CACHE_IRRELEVANT_OPTIONS = frozenset( {"jobs_dir", "job_name", "force_rerun", "quiet", "n_concurrent_trials", "agent_dir", "reward_key"} ) +# Where Harbor persists the JobConfig it will compare a resume against. +_HARBOR_JOB_CONFIG_FILENAME = "config.json" +# Fields Harbor's own JobConfig equality ignores, so they can never be why it refused +# to resume. Pinned against Harbor upstream by the drift-guard test. +_HARBOR_EQ_IGNORED_FIELDS = frozenset({"job_name", "debug"}) +# How Harbor says "this job dir cannot be resumed": one from the JobConfig comparison +# in `Job.create`, one from the lock.json check early in `Job.run`. Matching on the +# message is deliberate coupling, and it fails in the safe direction — an +# unrecognized FileExistsError propagates untouched rather than costing a job dir, so +# a Harbor reword degrades to a loud crash, never to a silent deletion. +_HARBOR_RESUME_REFUSALS = ("resumed with a different config", "does not match the resolved job lock") +# Cap on each value rendered into the "what differed" log line: enough for a scalar +# like `n_concurrent_trials`, bounded for a whole nested `agents` list. +_DRIFT_VALUE_CHARS = 80 # Derived/VCS noise skipped when digesting a directory. Deliberately NOT skipped: # `node_modules` and other vendored dependency trees, which ship with the agent and # change what it does. `.venv`/`.uv` stay skipped because they are environment, not @@ -228,13 +248,19 @@ async def run_tasks( :func:`discover_harbor_tasks`) unless a ``dataset_path`` override was given, so callers don't repeat it. - ``job_dir`` doubles as a cache, and it is reused only when **both** hold: - every requested task already has ``n_attempts`` completed, non-errored - results there, *and* the directory carries a cache stamp matching this run's - inputs (agent contents, task contents, result-affecting options). Anything - else re-runs from scratch — the directory is discarded rather than handed to - Harbor, because a surviving directory plus a changed agent is exactly the - case Harbor itself refuses. + ``job_dir`` doubles as a cache. Results are served straight off it, without + importing Harbor at all, only when **both** hold: every requested task already + has ``n_attempts`` completed, non-errored results there, *and* the directory + carries a cache stamp matching this run's inputs (agent contents, task + contents, result-affecting options). + + Otherwise Harbor runs, and what happens to the directory depends on *which* + check failed. A **stamp mismatch** discards it first: those results came from + different inputs, so there is nothing safe to resume onto. A directory that + merely lacks **coverage** — stamp matches, but not enough completed results — + is handed to Harbor intact so its per-trial resume keeps the finished trials + and runs only what is missing. Harbor may still refuse a directory on its own + (stricter) terms; :func:`_build_native_job` then discards it and re-runs. The cache only engages when the config pins a stable ``job_name``; with the default timestamped name no fingerprint is computed at all. @@ -263,13 +289,12 @@ async def run_tasks( dataset_path, self._task_names, job_name=job_name, - # Discard when the inputs changed, and whenever `agent_dir` is - # set: Harbor bakes a fresh uuid into the scoped agent import - # path, so its own JobConfig never matches on a rerun and it - # raises FileExistsError instead of resuming. With `agent_dir` - # unset the AgentConfig is deterministic, so leaving force_rerun - # off lets Harbor resume per trial and keep completed work. - force_rerun=(self._config.force_rerun or stale or self._config.agent_dir is not None), + # Discard only when the inputs changed. Otherwise leave it off so + # Harbor resumes per trial and keeps completed work — including + # with `agent_dir` set, now that the scoped import path is + # content-addressed rather than a fresh uuid per run and Harbor's + # JobConfig comparison can therefore match (AALGO-430). + force_rerun=(self._config.force_rerun or stale), ) # Fingerprint the inputs *before* running and confirm they are # unchanged afterwards. Stamping only the post-run state would label @@ -736,15 +761,52 @@ async def _create_and_run(agent: Any) -> None: datasets=[DatasetConfig(path=dataset_path, task_names=list(task_names) if task_names else None)], **timeout_kwargs, ) - job = await Job.create(job_config) - await job.run() + + async def _attempt() -> None: + job = await Job.create(job_config) + await job.run() + + try: + await _attempt() + except FileExistsError as exc: + # Harbor refuses to resume a job dir whose persisted `config.json` or + # `lock.json` differs from this run's — and it refuses by raising, not + # by re-running. Its comparison is deliberately stricter than the SDK + # cache stamp: `quiet`, `n_concurrent_trials` and the `task_names` + # filter all change the JobConfig without changing the results, so the + # stamp excludes them (a full cache hit must not pay for a concurrency + # tweak) while Harbor still rejects the directory. Honour the intent of + # the rerun rather than propagating a crash. + # + # Identify the refusal positively before deleting anything. Both of + # Harbor's refusals fire before any trial executes, so discarding costs + # only completed work — but that reasoning holds *only* for those two. + # An ordinary "file exists" raised from inside a trial, a hook, or an + # environment build must not be mistaken for drift and answered by + # destroying the directory. + if not (job_dir.exists() and _is_harbor_resume_refusal(exc, job_dir)): + raise + drift = _describe_job_config_drift(job_dir, job_config) + logger.warning( + "Harbor refused to resume job dir %s, so it is being discarded and re-run from scratch: %s%s", + job_dir, + exc, + f" Differing config: {drift}." if drift else "", + ) + shutil.rmtree(job_dir) + await _attempt() if config.agent_import_path is None: await _create_and_run(AgentConfig(name=config.agent_name or "oracle", model_name=config.agent_model_name)) elif config.agent_dir is not None: - # Loose wrapper file: make its directory importable for the run. + # Loose wrapper file: make its directory importable for the run. The + # jobs_dir exclusion must match _cache_stamp's, or a jobs_dir nested under + # agent_dir would shift the package name as results accumulate. agent_dir = config.agent_dir.expanduser().resolve() - with scoped_harbor_agent_import(agent_dir, config.agent_import_path) as scoped_import: + excluded_roots = frozenset({config.jobs_dir.expanduser().resolve()}) + with scoped_harbor_agent_import( + agent_dir, config.agent_import_path, exclude=excluded_roots + ) as scoped_import: await _create_and_run(AgentConfig(import_path=scoped_import, model_name=config.agent_model_name)) else: # Already-importable module (installed package): let Harbor import it directly. @@ -753,13 +815,73 @@ async def _create_and_run(agent: Any) -> None: return job_dir, run_job +def _is_harbor_resume_refusal(exc: FileExistsError, job_dir: Path) -> bool: + """Return True when ``exc`` is Harbor declining to resume ``job_dir``. + + Separates Harbor's refusal — the one case where deleting the directory is the + right answer — from an ordinary "file exists" surfacing from a trial, a hook or an + environment build, where deleting it would destroy completed work to no purpose. + + Two signals, both required. Harbor constructs its refusals with a bare message, so + ``errno`` is unset, while an OS-level ``EEXIST`` always carries one; and both + refusals name the job directory and end in a known phrase. + """ + if exc.errno is not None: + return False + message = str(exc) + return str(job_dir) in message and any(phrase in message for phrase in _HARBOR_RESUME_REFUSALS) + + +def _describe_job_config_drift(job_dir: Path, job_config: Any) -> str: + """Name the fields that differ between ``job_dir``'s persisted JobConfig and this run's. + + Harbor reports *that* an existing config differs, never *which* field, which + leaves the resulting discard looking arbitrary. This reproduces enough of its + comparison to say — turning "Harbor refused" into "n_concurrent_trials: 10 -> 4". + + Best-effort by construction. Returns ``""`` when the difference cannot be + located: no ``config.json``, unparseable, or a refusal that came from + ``lock.json`` instead, which has no JobConfig difference to report. Diagnostics + must never mask the failure they explain, so every error here is swallowed. + """ + try: + stored_text = (job_dir / _HARBOR_JOB_CONFIG_FILENAME).read_text(encoding="utf-8") + # Harbor persists with exclude_defaults=True, so the JSON omits every field + # left at its default and comparing it raw would report phantom differences. + # Round-tripping through the model refills them, which is what Harbor itself + # compares after re-validating the stored config. + stored = type(job_config).model_validate_json(stored_text).model_dump() + current = job_config.model_dump() + return ", ".join( + f"{field}: {_truncated_repr(stored.get(field))} -> {_truncated_repr(value)}" + for field, value in current.items() + if field not in _HARBOR_EQ_IGNORED_FIELDS and stored.get(field) != value + ) + except Exception: + return "" + + +def _truncated_repr(value: Any) -> str: + """Render ``value`` for a log line, bounded so a nested config can't flood it.""" + text = repr(value) + return text if len(text) <= _DRIFT_VALUE_CHARS else f"{text[:_DRIFT_VALUE_CHARS]}..." + + @contextlib.contextmanager -def scoped_harbor_agent_import(agent_dir: Path, import_path: str) -> Iterator[str]: - """Make ``agent_dir`` importable under a unique synthetic package for the block. +def scoped_harbor_agent_import( + agent_dir: Path, import_path: str, *, exclude: frozenset[Path] = frozenset() +) -> Iterator[str]: + """Make ``agent_dir`` importable under a content-addressed package for the block. Args: agent_dir: directory containing the module referenced by ``import_path``. import_path: Harbor agent path, ``"module"`` or ``"module:attribute"``. + exclude: resolved directories to leave out of the content digest. Pass the + same set :func:`_cache_stamp` uses — in practice ``jobs_dir``, which is + caller-chosen and may sit *under* ``agent_dir``. Omitting it lets the + growing results tree feed the package name, so the import path would + change on every run and the resume this function exists to enable would + never happen. See :func:`_digest_directory`. Yields: str: the rewritten import path Harbor should load (the module rooted under @@ -768,9 +890,33 @@ def scoped_harbor_agent_import(agent_dir: Path, import_path: str) -> Iterator[st Raises: ValueError: if ``import_path`` has no module component. - On exit the injected ``sys.modules`` entries are removed. The mutation is - guarded by a process-wide lock so concurrent runs don't corrupt import state; - each run gets its own uniquely-named package so distinct agents never collide. + **The package name is derived from the directory's contents, not a random + UUID, and that is load-bearing.** This string becomes ``AgentConfig.import_path`` + and therefore part of Harbor's ``JobConfig``, which Harbor compares field-by-field + when deciding whether an existing job directory may be resumed. A random suffix + made that comparison fail on every rerun, so Harbor raised ``FileExistsError`` + instead of resuming and its per-trial resume was unreachable for any caller that + sets ``agent_dir`` (AALGO-430). Content-addressing keeps distinct agents isolated + while letting an unchanged agent resume — and makes an *edited* agent invalidate + the job dir on Harbor's own terms. + + Identical contents therefore share a package name, so overlapping scopes are + refcounted: the injected ``sys.modules`` entries are removed when the last + scope exits, not the first (see :func:`_uninstall_agent_package`). The mutation + is guarded by a process-wide lock. ``sys.modules`` is per-process, so concurrent + *processes* were never at risk here. + + The name is ``_``, so it tracks the directory's *location* as + well as its contents — deliberately, because an opaque hash makes every traceback + and import error unreadable. That is a narrow, knowing divergence from the cache + stamp, which excludes ``agent_dir`` so a relocated but identical agent still hits + (see :func:`_cache_stamp`). Relocating an agent while pinning the same + ``job_name`` therefore leaves the stamp valid but changes this string, and Harbor + declines to resume; :func:`_build_native_job` absorbs that into a clean re-run. + The results stay correct — it costs one repeated job. Callers that rebuild agents + under changing directory names (the Experimentalist does) are unaffected, because + the agent name feeds their ``job_name`` too, so a rename lands in a different job + dir with nothing to resume. Only ``agent_dir`` (not ``sys.path``) is made importable, so a loose wrapper must be self-contained: a single module, or one that reaches siblings via @@ -782,7 +928,10 @@ def scoped_harbor_agent_import(agent_dir: Path, import_path: str) -> Iterator[st module_name = module_name.strip().lstrip(".") if not module_name: raise ValueError("import_path must be 'module' or 'module:attribute'") - package = f"{_AGENT_IMPORT_ROOT}.{_safe_identifier(agent_dir.name)}_{uuid4().hex[:8]}" + # Hashed here rather than reused from the cache stamp: this must describe the tree + # as it is about to be imported, and the extra walk is noise next to Docker. + suffix = _digest_directory(agent_dir, exclude=exclude)[:_IMPORT_DIGEST_CHARS] + package = f"{_AGENT_IMPORT_ROOT}.{_safe_identifier(agent_dir.name)}_{suffix}" with _IMPORT_LOCK: _install_agent_package(package, agent_dir) try: @@ -802,7 +951,11 @@ def _safe_identifier(value: str) -> str: def _install_agent_package(package: str, agent_dir: Path) -> None: - """Register ``package`` (and its parents) in ``sys.modules`` rooted at ``agent_dir``.""" + """Register ``package`` (and its parents) in ``sys.modules`` rooted at ``agent_dir``. + + Refcounted: package names are content-addressed, so two overlapping scopes on the + same agent directory legitimately share one. Callers must hold ``_IMPORT_LOCK``. + """ parts = package.split(".") for idx in range(1, len(parts) + 1): name = ".".join(parts[:idx]) @@ -814,10 +967,26 @@ def _install_agent_package(package: str, agent_dir: Path) -> None: if idx > 1: setattr(sys.modules[".".join(parts[: idx - 1])], parts[idx - 1], module) sys.modules[package].__path__ = [str(agent_dir)] + # Counted only once the injection it guards has succeeded. Incrementing first + # would strand the count above zero if any step above raised — the scope never + # opens, so nothing ever decrements it, and the package could never be torn down + # again for the life of the process. + _AGENT_PACKAGE_REFCOUNTS[package] = _AGENT_PACKAGE_REFCOUNTS.get(package, 0) + 1 def _uninstall_agent_package(package: str) -> None: - """Remove ``package`` and any submodules imported through it from ``sys.modules``.""" + """Remove ``package`` and its submodules from ``sys.modules`` on the last exit. + + Tearing down on the *first* exit would break a still-open scope sharing the same + content-addressed name, so the removal waits for the refcount to reach zero. + Callers must hold ``_IMPORT_LOCK``. + """ + remaining = _AGENT_PACKAGE_REFCOUNTS.get(package, 0) - 1 + if remaining > 0: + _AGENT_PACKAGE_REFCOUNTS[package] = remaining + return + _AGENT_PACKAGE_REFCOUNTS.pop(package, None) + for name in [n for n in sys.modules if n == package or n.startswith(f"{package}.")]: sys.modules.pop(name, None) parent, _, child = package.rpartition(".") From cbe2b56bd7873350abf7e2a16fe5e5b2cfc912a7 Mon Sep 17 00:00:00 2001 From: Nick Goncharenko Date: Wed, 29 Jul 2026 16:11:14 -0700 Subject: [PATCH 3/3] test(evaluator-sdk): cover the Harbor trial-adapter edge cases (A2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two of the four A2 cases were already covered (multi-attempt dir naming, non-`reward` metric keys). These close the other two, which the smoke run never exercised: - Trial dir with no result.json — Harbor still creates the dir when a trial dies before the verifier. Assert it is skipped rather than raising, and that its task is still reported missing instead of silently vanishing. Same for a truncated result.json. - exception_info shapes beyond a bare string. Harbor writes a mapping; only the bare-string path was tested. Parametrized over exception_type/type/name/class, a mapping with none of those (-> UnknownException), an empty mapping, a bare string, and a non-string scalar — plus the absent case, which gives the rest their meaning. Resolving any of these to None would promote a crashed trial to COMPLETED and let it score. Verified the parametrization is load-bearing: disabling the Mapping branch of _exception_type fails 6 of its 8 cases. No production code changed; `make vendor` produces no diff. Co-Authored-By: Claude Opus 5 Signed-off-by: Nick Goncharenko --- .../tests/agent_eval/test_harbor_runtime.py | 105 +++++++++++++++++- 1 file changed, 104 insertions(+), 1 deletion(-) diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime.py index 882f37f438..6179862f5f 100644 --- a/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime.py +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_harbor_runtime.py @@ -35,8 +35,13 @@ def _write_trial( - job_dir: Path, trial_name: str, task_name: str, *, reward: float | None, exception: str | None = None + job_dir: Path, trial_name: str, task_name: str, *, reward: float | None, exception: object | None = None ) -> None: + """Write one Harbor trial dir. ``exception`` is stored verbatim as ``exception_info``. + + Typed loosely on purpose: Harbor writes a mapping there, older runs wrote a bare + string, so the adapter has to cope with both. + """ trial_dir = job_dir / trial_name (trial_dir / "agent").mkdir(parents=True) (trial_dir / "verifier").mkdir(parents=True) @@ -1156,3 +1161,101 @@ def test_every_harbor_job_config_field_is_classified_against_the_sdk_stamp() -> assert not classified - actual, ( f"{sorted(classified - actual)} no longer exist on Harbor's JobConfig; drop them from the classification." ) + + +def test_trial_dir_without_result_json_is_skipped_and_its_task_reported_missing( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A trial that died before the verifier leaves a dir but no result.json. + + Harbor still creates the trial dir (and often an exception.txt), so the adapter + has to tolerate the absence rather than raise, while the task it belonged to + must not silently vanish from the run — it is reported as having no result. + """ + job_dir = tmp_path / "job" + job_dir.mkdir() + _write_trial(job_dir, "ok-task__aaa", "ok-task", reward=1.0) + crashed = job_dir / "crashed-task__bbb" + (crashed / "agent").mkdir(parents=True) + (crashed / "exception.txt").write_text("Traceback (most recent call last): ...") + + tasks = [ + AgentEvalTask(id="ok-task", intent="x", inputs={"instruction": "p"}, metrics=[HarborRewardMetric()]), + AgentEvalTask(id="crashed-task", intent="y", inputs={"instruction": "q"}, metrics=[HarborRewardMetric()]), + ] + with caplog.at_level(logging.WARNING): + trials = build_trials_from_job_dir(job_dir, tasks) + + assert [trial.task_id for trial in trials] == ["ok-task"] + assert "crashed-task" in caplog.text + + +def test_unreadable_result_json_is_skipped_without_raising(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """A truncated result.json must not take the whole job's adaptation down.""" + job_dir = tmp_path / "job" + job_dir.mkdir() + _write_trial(job_dir, "ok-task__aaa", "ok-task", reward=1.0) + broken = job_dir / "broken-task__bbb" + broken.mkdir() + (broken / "result.json").write_text("{not json") + + tasks = [ + AgentEvalTask(id="ok-task", intent="x", inputs={"instruction": "p"}, metrics=[HarborRewardMetric()]), + AgentEvalTask(id="broken-task", intent="y", inputs={"instruction": "q"}, metrics=[HarborRewardMetric()]), + ] + with caplog.at_level(logging.WARNING): + trials = build_trials_from_job_dir(job_dir, tasks) + + assert [trial.task_id for trial in trials] == ["ok-task"] + assert "broken-task" in caplog.text + + +@pytest.mark.parametrize( + ("exception_info", "expected_type"), + [ + ({"exception_type": "TimeoutError"}, "TimeoutError"), + ({"type": "TimeoutError"}, "TimeoutError"), + ({"name": "TimeoutError"}, "TimeoutError"), + ({"class": "TimeoutError"}, "TimeoutError"), + # A mapping Harbor filled with something unexpected still counts as failed. + ({"message": "boom"}, "UnknownException"), + ({}, "UnknownException"), + # Older/other writers put a bare value there. + ("NonZeroAgentExitCodeError", "NonZeroAgentExitCodeError"), + (17, "17"), + ], +) +def test_exception_info_shapes_all_resolve_to_a_type( + tmp_path: Path, exception_info: object, expected_type: str +) -> None: + """Any non-null exception_info must mark the trial failed, whatever its shape. + + Only a bare string was covered before, so a mapping — which is what Harbor + actually writes — went untested. Resolving to None here would silently promote + a crashed trial to COMPLETED and let it score. + """ + job_dir = tmp_path / "job" + job_dir.mkdir() + _write_trial(job_dir, "t__aaa", "t", reward=1.0, exception=exception_info) + + trials = build_trials_from_job_dir( + job_dir, [AgentEvalTask(id="t", intent="x", inputs={"instruction": "p"}, metrics=[HarborRewardMetric()])] + ) + + assert len(trials) == 1 + assert trials[0].metadata["exception_type"] == expected_type + assert trials[0].status is AgentEvalTrialStatus.PARTIAL + + +def test_absent_exception_info_leaves_the_trial_completed(tmp_path: Path) -> None: + """The negative case that gives the parametrization above its meaning.""" + job_dir = tmp_path / "job" + job_dir.mkdir() + _write_trial(job_dir, "t__aaa", "t", reward=1.0, exception=None) + + trials = build_trials_from_job_dir( + job_dir, [AgentEvalTask(id="t", intent="x", inputs={"instruction": "p"}, metrics=[HarborRewardMetric()])] + ) + + assert "exception_type" not in trials[0].metadata + assert trials[0].status is AgentEvalTrialStatus.COMPLETED