From 3f6fb01685cba189ab8d8d94b33a9228084cd0a3 Mon Sep 17 00:00:00 2001 From: Nick Goncharenko Date: Mon, 3 Aug 2026 09:35:05 -0700 Subject: [PATCH] feat(evaluator-sdk): add content-aware Harbor cache Signed-off-by: Nick Goncharenko --- .../agent_eval/runtimes/harbor_runtime.py | 667 +++++++++- .../tests/agent_eval/test_harbor_runtime.py | 1127 ++++++++++++++++- .../agent_eval/test_harbor_runtime_e2e.py | 86 ++ .../agent_eval/runtimes/harbor_runtime.py | 667 +++++++++- 4 files changed, 2477 insertions(+), 70 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 cfd3b3da5a..374f2defa6 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 @@ -32,9 +32,11 @@ from __future__ import annotations import contextlib +import hashlib import importlib.machinery import json import logging +import os import re import shutil import sys @@ -45,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 @@ -71,6 +72,42 @@ _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" +# Public so downstreams can assert the SDK is new enough to own cache staleness. +CACHE_STAMP_VERSION = 1 +# Excluded from the cache fingerprint — see :func:`_cache_stamp` for why each one. +_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 +# deliverable — the Harbor wrapper does not upload them into the task container. +_DIGEST_SKIP_DIRS = frozenset({".git", "__pycache__", ".venv", ".uv", ".mypy_cache", ".pytest_cache"}) +_DIGEST_CHUNK_BYTES = 1 << 20 RunJob = Callable[[], Awaitable[None]] @@ -209,17 +246,78 @@ async def run_tasks( In native mode the dataset directory is recovered from the tasks (each carries ``metadata['harbor_dataset_path']`` from :func:`discover_harbor_tasks`) unless a ``dataset_path`` override was given, - so callers don't repeat it. ``job_dir`` doubles as a cache: the Harbor run - is skipped and results are simply re-adapted when every requested task - already has ``n_attempts`` completed, non-errored results there (unless - ``force_rerun`` is set). The cache only engages when the config pins a - stable ``job_name``; the default timestamped name never hits it. + so callers don't repeat it. + + ``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. + + Assumes a **single writer per job directory**. Neither this runtime nor + Harbor locks it, so two processes sharing a pinned ``job_name`` on a shared + volume will race. """ if self._config is not None: dataset_path = self._dataset_path or _dataset_path_from_tasks(tasks) - job_dir, run_job = _build_native_job(self._config, dataset_path, self._task_names) - if self._config.force_rerun or not _all_tasks_cached(job_dir, tasks, n_attempts=self._config.n_attempts): + job_name, job_dir = _resolve_job_dir(self._config) + + # Only fingerprint when the answer can depend on it: an unpinned job name + # can never hit, and force_rerun/a missing dir already decided. This keeps + # the digest I/O off every run of callers that don't pin a job name. + stamp: dict[str, Any] | None = None + if self._config.job_name is None or self._config.force_rerun or not job_dir.is_dir(): + stale = True + else: + stamp = _cache_stamp(self._config, dataset_path, tasks) + stale = _cache_is_stale(job_dir, stamp) + + if stale or not _all_tasks_cached(job_dir, tasks, n_attempts=self._config.n_attempts): + job_dir, run_job = _build_native_job( + self._config, + dataset_path, + self._task_names, + job_name=job_name, + # 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 + # results produced from the old sources with the new fingerprint, so + # a later run would happily serve them. Covers what Harbor actually + # ran: with no task_names filter that is the whole dataset, and + # recording only the requested subset would make the next full-set + # run look stale and re-run a complete job dir. + coverage = _stamp_coverage(dataset_path, tasks, self._task_names) + before = _cache_stamp(self._config, dataset_path, coverage) await run_job() + if self._config.job_name is not None: + after = _cache_stamp(self._config, dataset_path, coverage) + if after == before: + _write_cache_stamp(job_dir, after) + else: + # Leaving it unstamped re-runs next time, which is the safe + # direction: we cannot say which inputs produced these results. + logger.warning( + "Agent or task contents changed while Harbor job %s was running; leaving it unstamped " + "so the next run re-executes rather than trusting these results.", + job_dir, + ) return build_trials_from_job_dir(job_dir, tasks, reward_key=self._reward_key) if self._job_dir is None: # unreachable: __init__ guarantees config or job_dir @@ -268,10 +366,352 @@ def _all_tasks_cached(job_dir: Path, tasks: Sequence[AgentEvalTask], *, n_attemp return all(counts.get(task.id, 0) >= n_attempts for task in tasks) +def _feed(digest: "hashlib._Hash", label: bytes, payload: bytes) -> None: + """Append a length-framed field to ``digest``. + + Framing matters: concatenating ``name \0 content \0`` is ambiguous because file + *contents* may contain NUL, so two different trees can produce an identical byte + stream. Prefixing every variable-length field with its length makes the encoding + injective, which is what stops a collision from being read as "unchanged". + """ + digest.update(label) + digest.update(len(payload).to_bytes(8, "big")) + digest.update(payload) + + +def _safe_resolve(path: Path) -> Path: + """``Path.resolve()`` that degrades instead of raising. + + The digest is a best-effort guard, not a reason to fail a run that would + otherwise succeed, so fall back to the unresolved absolute path. + + ``RuntimeError`` is caught alongside ``OSError`` and is the case that actually + fires: on CPython 3.12 — the floor this package targets — a **symlink loop** + surfaces as ``RuntimeError("Symlink loop from ...")``, because ``resolve()`` + translates ``ELOOP`` before re-raising. It is not an ``OSError``, so catching + only that would let a loop under a task directory kill the run. A loop raises + deterministically, not as a race. ``OSError`` covers the narrower case of a + symlink that disappears mid-walk, since ``resolve()`` calls ``os.readlink``. + + Both are 3.12/3.13 behaviours: 3.14 resolves a loop without raising at all. + """ + try: + return path.resolve() + except (OSError, RuntimeError): + return path.absolute() + + +def _is_executable(path: Path) -> bool: + """Whether the owner-execute bit is set, following symlinks. + + Only the execute bit, mirroring what git tracks: read/write bits vary with umask + and would evict the cache for nothing, but flipping +x on ``tests/test.sh`` or an + agent entrypoint genuinely changes what Harbor does. + """ + try: + return bool(path.stat().st_mode & 0o100) + except OSError: + return False + + +def _digest_directory(root: Path, *, exclude: frozenset[Path] = frozenset()) -> str: + """Content-hash a directory tree, skipping build/VCS noise and excluded roots. + + Contents rather than mtimes: callers routinely materialize the directory with + ``copytree`` (the optimizer does, per candidate), which rewrites every mtime and + would defeat the cache entirely. + + ``exclude`` takes *resolved* directories to skip wholesale. It exists because + ``jobs_dir`` is caller-chosen and may sit **under** the dataset or agent + directory; without excluding it the digest would hash the growing results tree + it is meant to validate, and would never stabilize. + + Symlinks are **followed**, not skipped. Skipping them silently defeats the whole + guard: a directory assembled out of links to shared sources would hash to the + empty digest, so edits behind those links would never invalidate the cache. The + link target is folded in alongside the contents, so re-pointing a link is a + change even when both targets happen to hold identical bytes. + + Unreadable or vanished files are folded in as a marker rather than raised: a + transient read failure must not kill a run that would otherwise have succeeded. + """ + digest = hashlib.sha256() + if not root.is_dir(): + return digest.hexdigest() + # Keep only exclusions strictly *inside* this tree. An excluded root that is an + # ancestor of (or equal to) `root` would otherwise match every entry and yield + # an empty digest — silently disabling invalidation for the whole directory, + # which is exactly the failure this function exists to prevent. jobs_dir being + # a parent of the agent/task dir is a legitimate layout, not a reason to stop + # hashing it. + root_resolved = _safe_resolve(root) + excluded = { + resolved + for resolved in (_safe_resolve(path) for path in exclude) + if resolved != root_resolved and resolved.is_relative_to(root_resolved) + } + # Resolved dirs already walked, so a symlink cycle terminates instead of hanging. + visited: set[Path] = set() + + def walk(directory: Path) -> None: + resolved_dir = _safe_resolve(directory) + if resolved_dir in visited: + return + visited.add(resolved_dir) + try: + entries = sorted(directory.iterdir()) + except OSError as exc: + logger.warning("Could not list %s while fingerprinting %s: %s", directory, root, exc) + _feed(digest, b"unlistable", b"") + return + for path in entries: + if path.name in _DIGEST_SKIP_DIRS: + continue + resolved = _safe_resolve(path) + if any(resolved == item or item in resolved.parents for item in excluded): + continue + + _feed(digest, b"name", path.relative_to(root).as_posix().encode("utf-8")) + _feed(digest, b"mode", b"x" if _is_executable(path) else b"-") + if path.is_symlink(): + # Record where the link points, so retargeting counts as a change. + try: + target = os.readlink(path).encode("utf-8") + except OSError as exc: + # The link vanished mid-walk. Same contract as an unreadable + # file: degrade to a marker rather than kill the run. + logger.warning("Could not read link %s while fingerprinting %s: %s", path, root, exc) + target = b"" + _feed(digest, b"symlink", target) + if path.is_dir(): + _feed(digest, b"dir", b"") + walk(path) + continue + if not path.is_file(): + # Broken link, socket, fifo: nothing to hash, but its presence counts. + _feed(digest, b"not-a-file", b"") + continue + content = hashlib.sha256() + try: + with path.open("rb") as handle: + # Streamed: Harbor datasets may ship large seeds or build contexts. + for chunk in iter(lambda: handle.read(_DIGEST_CHUNK_BYTES), b""): + content.update(chunk) + except OSError as exc: + logger.warning("Could not read %s while fingerprinting %s: %s", path, root, exc) + content.update(b"") + # The sub-digest is fixed width, so file bytes can never be confused with + # the framing around them. + _feed(digest, b"file", content.digest()) + + walk(root) + return digest.hexdigest() + + +def _task_dirs_for(dataset_path: Path, tasks: Sequence[AgentEvalTask]) -> dict[str, Path | None]: + """Resolve each task's on-disk directory, falling back to re-discovery. + + :func:`discover_harbor_tasks` stamps ``metadata['harbor_task_dir']``, but callers + may build :class:`AgentEvalTask` objects by hand (the Evaluator plugin builds + them from a job spec), so the metadata is not guaranteed. + + A task that cannot be resolved maps to ``None`` — the caller must treat that as + un-cacheable rather than silently omitting it from the fingerprint, which would + be a stale-cache hole. + """ + # `_safe_resolve`, not bare `resolve()`: this walk is a best-effort cache guard, so + # a symlink that vanishes mid-run must degrade to an unresolved absolute path + # rather than raise out of a job that would otherwise succeed. + dataset_root = _safe_resolve(dataset_path) + resolved: dict[str, Path | None] = {} + for task in tasks: + stamped = task.metadata.get("harbor_task_dir") + candidate = Path(stamped) if isinstance(stamped, str) and stamped else None + # The stamp records where a task was *discovered*, which is not necessarily + # where this run executes it: `dataset_path` can be overridden on the runner. + # Trusting a stale or foreign path would fingerprint one dataset while Harbor + # runs another, so anything missing or outside the active dataset is dropped + # and re-discovered below. + if candidate is not None: + candidate_resolved = _safe_resolve(candidate) + if not candidate.is_dir() or not candidate_resolved.is_relative_to(dataset_root): + logger.debug( + "Ignoring stamped harbor_task_dir %s for task %r: not a directory under the active dataset %s", + candidate, + task.id, + dataset_root, + ) + candidate = None + resolved[task.id] = candidate + if all(path is not None for path in resolved.values()): + return resolved + + try: + discovered = { + task.id: Path(str(task.metadata["harbor_task_dir"])) for task in discover_harbor_tasks(dataset_path) + } + except (OSError, ValueError) as exc: + # discover_harbor_tasks raises on ANY malformed task.toml in the dataset. + # Refusing the cache is the safe reading; failing the run is not, since this + # path previously never read those files. + logger.warning( + "Could not resolve Harbor task dirs under %s; treating the cache as stale: %s", dataset_path, exc + ) + return dict.fromkeys(resolved, None) + return {task_id: path or discovered.get(task_id) for task_id, path in resolved.items()} + + +def _stamp_coverage( + dataset_path: Path, + tasks: Sequence[AgentEvalTask], + task_names: Sequence[str] | None, +) -> Sequence[AgentEvalTask]: + """Tasks a written stamp must cover: everything Harbor was asked to run. + + ``task_names`` is the filter handed to Harbor's ``DatasetConfig``. When it is + ``None`` Harbor runs every task in the dataset, which can be a superset of the + tasks this call was asked to score — and a stamp that recorded only the smaller + set would report the larger one as stale on the next run. + """ + if task_names is not None: + return tasks + try: + discovered = discover_harbor_tasks(dataset_path) + except (OSError, ValueError): + # Same reasoning as _task_dirs_for: a malformed sibling task must not fail a + # run. Recording only the requested tasks just costs a re-run later. + return tasks + covered = {task.id: task for task in discovered} + covered.update({task.id: task for task in tasks}) + return list(covered.values()) + + +def _cache_stamp( + config: HarborRuntimeConfig, + dataset_path: Path, + tasks: Sequence[AgentEvalTask], +) -> dict[str, Any]: + """Fingerprint the inputs that decide whether a job dir can be reused. + + Covers the result-affecting options, the contents of ``agent_dir``, and the + contents of every task directory. Two gaps are deliberate and worth knowing + before trusting a hit: when ``agent_dir`` is ``None`` the agent is an already + importable module, so only its *import path* is fingerprinted and edits to that + installed package are invisible; and a task whose directory cannot be resolved + is recorded as ````, which always forces a re-run. + + Recorded per task rather than as one job-wide hash so that evaluating a + **subset** of a previously-cached job still hits: staleness is decided only over + the tasks actually requested. + + Excluded from the option hash: presentation and placement knobs (``quiet``, + ``n_concurrent_trials``, ``jobs_dir``, ``job_name``, ``force_rerun``), which + change nothing about the results; ``agent_dir``, an absolute path whose + *content* is hashed separately, so a relocated but identical agent still hits; + and ``reward_key``, which only selects which reward + :func:`build_trials_from_job_dir` reads back and must not cost a Docker re-run. + """ + options = config.model_dump(exclude=set(_CACHE_IRRELEVANT_OPTIONS), mode="json") + # `_safe_resolve` throughout, matching `_task_dirs_for`: fingerprinting is + # best-effort, so a symlink loop or a vanished link under any of these must + # degrade to an unresolved path rather than raise out of `run_tasks` and fail a + # run that would otherwise succeed. + excluded_roots = frozenset({_safe_resolve(config.jobs_dir.expanduser())}) + + agent_digest = "" + if config.agent_dir is not None: + agent_digest = _digest_directory(_safe_resolve(config.agent_dir.expanduser()), exclude=excluded_roots) + + task_digests: dict[str, str] = {} + for task_id, task_dir in sorted(_task_dirs_for(dataset_path, tasks).items()): + task_digests[task_id] = ( + "" if task_dir is None else _digest_directory(_safe_resolve(task_dir), exclude=excluded_roots) + ) + + return { + "version": CACHE_STAMP_VERSION, + "options": hashlib.sha256(json.dumps(options, sort_keys=True, default=str).encode("utf-8")).hexdigest(), + "agent": agent_digest, + "tasks": task_digests, + } + + +def _cache_is_stale(job_dir: Path, stamp: Mapping[str, Any]) -> bool: + """Return True when ``job_dir`` was not produced by the inputs in ``stamp``. + + A directory with no stamp is stale: it predates this check, or was written by + plain Harbor, and re-running is the safe reading. An ```` task + digest is likewise always stale — we could not prove the inputs match. A + directory that does not exist is stale too: there is nothing there to reuse, and + answering "not stale" would be an invitation to serve zero trials. + """ + if not job_dir.is_dir(): + return True + try: + stored = json.loads((job_dir / CACHE_STAMP_FILENAME).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + stored = None + + reason: str | None = None + if not isinstance(stored, Mapping): + reason = "no usable cache stamp" + elif stored.get("version") != stamp["version"]: + reason = "cache stamp version changed" + elif stored.get("options") != stamp["options"]: + reason = "a result-affecting option changed" + elif stored.get("agent") != stamp["agent"]: + reason = "the agent directory changed" + else: + stored_tasks = stored.get("tasks") + stored_tasks = stored_tasks if isinstance(stored_tasks, Mapping) else {} + for task_id, digest in stamp["tasks"].items(): + if digest == "": + reason = f"task {task_id!r} could not be resolved on disk" + break + if stored_tasks.get(task_id) != digest: + reason = f"task {task_id!r} changed or was not part of the cached run" + break + + if reason is None: + return False + logger.info("Re-running Harbor job %s instead of serving it from cache: %s.", job_dir, reason) + return True + + +def _write_cache_stamp(job_dir: Path, stamp: Mapping[str, Any]) -> None: + """Record the inputs a completed job dir was produced from. + + Best-effort: a job dir that could not be stamped simply re-runs next time, which + is the safe direction. Written as a *file* deliberately — Harbor deletes any + stray *directory* in a job dir that lacks ``result.json``. + """ + try: + (job_dir / CACHE_STAMP_FILENAME).write_text( + json.dumps(stamp, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + except OSError as exc: + logger.warning("Could not stamp Harbor job dir %s with its cache key: %s", job_dir, exc) + + +def _resolve_job_dir(config: HarborRuntimeConfig) -> tuple[str, Path]: + """Resolve ``(job_name, job_dir)`` without importing Harbor. + + Split out from :func:`_build_native_job` because the caller must know the job + directory *before* deciding whether to run: an unpinned ``job_name`` is a + timestamp with microsecond precision, so resolving it twice would yield two + different directories and the cache decision would be made about the wrong one. + """ + job_name = config.job_name or datetime.now(timezone.utc).strftime("%Y-%m-%d__%H-%M-%S__%f") + return job_name, config.jobs_dir / job_name + + def _build_native_job( config: HarborRuntimeConfig, dataset_path: Path, task_names: Sequence[str] | None, + *, + job_name: str | None = None, + force_rerun: bool | None = None, ) -> tuple[Path, RunJob]: """Build a Harbor ``JobConfig`` from ``config`` and return ``(job_dir, run_job)``. @@ -280,9 +720,18 @@ def _build_native_job( without importing Harbor. When ``agent_import_path`` is set, ``run_job`` scopes the user's agent package into ``sys.modules`` for the run and removes it afterwards (see :func:`scoped_harbor_agent_import`). + + Args: + job_name: Pre-resolved job name from :func:`_resolve_job_dir`. Pass it when + the caller already resolved the directory, so an unpinned name is not + re-generated into a different timestamp. + force_rerun: Overrides ``config.force_rerun`` for this build. Passed rather + than applied via ``model_copy`` so the caller's config is never mutated + and the job name stays fixed. """ - job_name = config.job_name or datetime.now(timezone.utc).strftime("%Y-%m-%d__%H-%M-%S__%f") - job_dir = config.jobs_dir / job_name + resolved_name = job_name if job_name is not None else _resolve_job_dir(config)[0] + job_dir = config.jobs_dir / resolved_name + effective_force_rerun = config.force_rerun if force_rerun is None else force_rerun async def run_job() -> None: try: @@ -295,7 +744,7 @@ async def run_job() -> None: '(it requires Python >=3.12). Install it separately: uv pip install "harbor>=0.16.1"' ) from exc - if config.force_rerun and job_dir.exists(): + if effective_force_rerun and job_dir.exists(): shutil.rmtree(job_dir) artifacts: list[str | ArtifactConfig] = list(config.artifacts) @@ -316,7 +765,7 @@ async def run_job() -> None: async def _create_and_run(agent: Any) -> None: job_config = JobConfig( - job_name=job_name, + job_name=resolved_name, jobs_dir=config.jobs_dir, n_attempts=config.n_attempts, n_concurrent_trials=config.n_concurrent_trials, @@ -327,15 +776,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. @@ -344,13 +830,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 @@ -359,9 +905,41 @@ 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. + + **That last sentence only holds if the caller derives its job name from the + *resolved* directory, as this function does.** Deriving it from the caller's + spelling instead lets the two disagree: a symlink keeps its own name while + resolving elsewhere, so flipping it at a fixed ``job_name`` would reuse one job + dir for two different agents, caught only by Harbor's refusal rather than by + design. The Experimentalist resolves first for exactly this reason + (``resolve_harbor_run_inputs``). 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 @@ -373,7 +951,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: @@ -393,7 +974,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]) @@ -404,11 +989,41 @@ def _install_agent_package(package: str, agent_dir: Path) -> None: sys.modules[name] = module if idx > 1: setattr(sys.modules[".".join(parts[: idx - 1])], parts[idx - 1], module) - sys.modules[package].__path__ = [str(agent_dir)] + installed = sys.modules[package] + if not installed.__path__: + installed.__path__ = [str(agent_dir)] + elif installed.__path__ != [str(agent_dir)]: + # Two directories sharing this package name share a content digest, so their + # trees are byte-identical and the path already installed is exactly as + # correct as this one — the excluded content (`.git`, `__pycache__`, the + # env dirs, `jobs_dir`) is not importable. Repointing would swap the + # directory out from under a scope that is still open, for no gain. + logger.debug( + "Agent package %s is already installed from %s; keeping it for the identical tree at %s", + package, + installed.__path__[0], + 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(".") @@ -731,6 +1346,8 @@ def reward_payload_from_result( __all__ = [ + "CACHE_STAMP_FILENAME", + "CACHE_STAMP_VERSION", "DEFAULT_REWARD_KEY", "HarborAgentTaskRunner", "HarborRewardMetric", 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 b87917b2ef..7b53c1c6c8 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 @@ -2,11 +2,16 @@ # SPDX-License-Identifier: Apache-2.0 import asyncio +import builtins +import hashlib import importlib import json 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 @@ -15,6 +20,7 @@ HarborRewardMetric, HarborRuntimeConfig, HarborTasksetLoader, + _build_native_job, build_trials_from_job_dir, discover_harbor_tasks, reward_payload_from_result, @@ -23,14 +29,19 @@ 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" 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) @@ -164,31 +175,265 @@ def test_runtime_config_defaults_and_runner_requires_a_source() -> None: asyncio.run(runner.run_tasks([AgentEvalTask(id="t", intent="x", inputs={})])) -@pytest.mark.asyncio -async def test_native_runner_uses_job_dir_as_cache(tmp_path: Path) -> None: - # A native run whose job_dir already covers every requested task is re-adapted, - # not re-run: run_job is never awaited, so Harbor is never imported here. This - # also exercises recovering the dataset dir from task metadata (no dataset_path). - jobs_dir = tmp_path / "jobs" +def _cached_task(dataset_path: Path, task_dir: Path, task_id: str = "t") -> AgentEvalTask: + """A task whose dataset and on-disk directory the cache stamp can resolve.""" + return AgentEvalTask( + id=task_id, + intent="x", + inputs={"instruction": "x"}, + metrics=[HarborRewardMetric()], + metadata={"harbor_dataset_path": str(dataset_path), "harbor_task_dir": str(task_dir)}, + ) + + +def _seed_cached_job(tmp_path: Path, *, task_id: str = "t") -> tuple[HarborRuntimeConfig, Path, AgentEvalTask]: + """A complete job dir plus the config/task that produced it, stamped as a real run would.""" + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _cache_stamp, _write_cache_stamp + + dataset_path = tmp_path / "dataset" + task_dir = dataset_path / task_id + task_dir.mkdir(parents=True) + (task_dir / "task.toml").write_text(f'[task]\nname = "{task_id}"\n') + + # jobs_dir deliberately nested under the dataset dir: the digest must exclude it, + # or it would hash its own growing results tree and never stabilize. + jobs_dir = dataset_path / "jobs" job_dir = jobs_dir / "cached-job" job_dir.mkdir(parents=True) - _write_trial(job_dir, "t__aaa", "t", reward=1.0) + _write_trial(job_dir, f"{task_id}__aaa", task_id, reward=1.0) config = HarborRuntimeConfig(jobs_dir=jobs_dir, job_name="cached-job") - runner = HarborAgentTaskRunner(config=config) - tasks = [ - AgentEvalTask( - id="t", - intent="x", - inputs={"instruction": "x"}, - metrics=[HarborRewardMetric()], - metadata={"harbor_dataset_path": str(tmp_path)}, - ) - ] + task = _cached_task(dataset_path, task_dir, task_id) + _write_cache_stamp(job_dir, _cache_stamp(config, dataset_path, [task])) + return config, job_dir, task + + +@pytest.mark.asyncio +async def test_native_runner_uses_job_dir_as_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # A native run whose job_dir covers every requested task AND carries a matching + # cache stamp is re-adapted, not re-run: run_job is never awaited, so Harbor is + # never imported here (which is why this test needs no harbor install). + config, _job_dir, task = _seed_cached_job(tmp_path) + # Watch the lazy import directly instead of mutating sys.modules: popping only + # "harbor" would leave already-imported harbor.* submodules parentless and + # corrupt the module identity other suites monkeypatch. + imported: list[str] = [] + real_import = builtins.__import__ + + def recording_import(name, *args, **kwargs): + if name == "harbor" or name.startswith("harbor."): + imported.append(name) + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", recording_import) + trials = await HarborAgentTaskRunner(config=config).run_tasks([task]) + monkeypatch.undo() - trials = await runner.run_tasks(tasks) assert [trial.task_id for trial in trials] == ["t"] assert trials[0].metadata["reward"] == 1.0 + assert imported == [], f"a cache hit must not import harbor, but imported {imported}" + + +def _stamp_for(config: HarborRuntimeConfig, task: AgentEvalTask, job_dir: Path) -> None: + """Stamp ``job_dir`` as though ``config`` had just produced it.""" + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _cache_stamp, _write_cache_stamp + + dataset_path = Path(str(task.metadata["harbor_dataset_path"])) + _write_cache_stamp(job_dir, _cache_stamp(config, dataset_path, [task])) + + +def _spy_on_run_job(monkeypatch: pytest.MonkeyPatch, calls: list[bool]) -> None: + """Replace the native job build so run_tasks is observable without Harbor. + + Records whether the run was attempted and what force_rerun it was built with. + """ + from nemo_evaluator_sdk.agent_eval.runtimes import harbor_runtime + + def fake(config, _dataset_path, _task_names, *, job_name=None, force_rerun=None): + async def run_job() -> None: + calls.append(bool(force_rerun)) + + return config.jobs_dir / (job_name or "job"), run_job + + monkeypatch.setattr(harbor_runtime, "_build_native_job", fake) + + +@pytest.mark.asyncio +async def test_unstamped_job_dir_is_not_trusted(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # A complete job dir with no stamp predates this check, or was written by plain + # Harbor. Re-running is the safe reading. + config, job_dir, task = _seed_cached_job(tmp_path) + (job_dir / ".nemo-eval-harbor-cache.json").unlink() + calls: list[bool] = [] + _spy_on_run_job(monkeypatch, calls) + + await HarborAgentTaskRunner(config=config).run_tasks([task]) + + assert calls == [True], "an unstamped dir must be re-run, and discarded rather than resumed" + + +@pytest.mark.asyncio +async def test_changed_inputs_discard_the_job_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # A stamp mismatch means the surviving results were produced by different + # inputs, so they must be deleted rather than resumed onto. + config, _job_dir, task = _seed_cached_job(tmp_path) + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + (agent_dir / "wrapper.py").write_text("x = 1\n") + config = config.model_copy(update={"agent_import_path": "wrapper:Agent", "agent_dir": agent_dir}) + calls: list[bool] = [] + _spy_on_run_job(monkeypatch, calls) + + await HarborAgentTaskRunner(config=config).run_tasks([task]) + + assert calls == [True], "changed inputs must discard, not resume" + + +@pytest.mark.asyncio +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") + config, job_dir, task = _seed_cached_job(tmp_path) + config = config.model_copy(update={"agent_import_path": "wrapper:Agent", "agent_dir": agent_dir, "n_attempts": 2}) + _stamp_for(config, task, job_dir) # stamp matches; only coverage is short + calls: list[bool] = [] + _spy_on_run_job(monkeypatch, calls) + + await HarborAgentTaskRunner(config=config).run_tasks([task]) + + 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, 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 + calls: list[bool] = [] + _spy_on_run_job(monkeypatch, calls) + + await HarborAgentTaskRunner(config=config).run_tasks([task]) + + assert calls == [False], "a resumable miss must not delete completed trials" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mutation", ["agent", "task", "option"]) +async def test_changed_inputs_invalidate_the_cache(tmp_path: Path, mutation: str) -> None: + # Each of these changes what a run would produce, so the stamped dir must not be + # served. Reaching run_job (and failing there) is the observable signal. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _cache_is_stale, _cache_stamp + + config, job_dir, task = _seed_cached_job(tmp_path) + dataset_path = Path(str(task.metadata["harbor_dataset_path"])) + + if mutation == "agent": + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + (agent_dir / "wrapper.py").write_text("x = 1\n") + config = config.model_copy( + update={"agent_import_path": "wrapper:Agent", "agent_dir": agent_dir}, + ) + elif mutation == "task": + (dataset_path / "t" / "task.toml").write_text('[task]\nname = "t"\nchanged = true\n') + else: + config = config.model_copy(update={"n_attempts": 2}) + + assert _cache_is_stale(job_dir, _cache_stamp(config, dataset_path, [task])) is True + + +@pytest.mark.asyncio +async def test_cosmetic_options_do_not_evict_the_cache(tmp_path: Path) -> None: + # Presentation and placement knobs change nothing about the results; evicting on + # them would cost a full Docker re-run for nothing. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _cache_is_stale, _cache_stamp + + config, job_dir, task = _seed_cached_job(tmp_path) + dataset_path = Path(str(task.metadata["harbor_dataset_path"])) + relaxed = config.model_copy(update={"quiet": False, "n_concurrent_trials": 1, "reward_key": "other"}) + + assert _cache_is_stale(job_dir, _cache_stamp(relaxed, dataset_path, [task])) is False + + +@pytest.mark.asyncio +async def test_task_subset_of_a_cached_run_still_hits(tmp_path: Path) -> None: + # Stamping per task (not one job-wide hash) means evaluating a subset of a + # previously cached job is still a hit. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import ( + _cache_is_stale, + _cache_stamp, + _write_cache_stamp, + ) + + config, job_dir, task_a = _seed_cached_job(tmp_path, task_id="t") + dataset_path = Path(str(task_a.metadata["harbor_dataset_path"])) + task_b_dir = dataset_path / "u" + task_b_dir.mkdir() + (task_b_dir / "task.toml").write_text('[task]\nname = "u"\n') + task_b = _cached_task(dataset_path, task_b_dir, "u") + + _write_cache_stamp(job_dir, _cache_stamp(config, dataset_path, [task_a, task_b])) + + assert _cache_is_stale(job_dir, _cache_stamp(config, dataset_path, [task_a])) is False + + +def test_unpinned_job_name_writes_no_stamp_and_reads_no_files(tmp_path: Path) -> None: + # The default timestamped job name can never hit the cache, so the fingerprint + # must not be computed at all — this is the path plugins/nemo-evaluator takes. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _resolve_job_dir + + config = HarborRuntimeConfig(jobs_dir=tmp_path / "jobs") + first = _resolve_job_dir(config)[1] + + assert config.job_name is None + assert first.parent == tmp_path / "jobs" + assert not list((tmp_path / "jobs").glob("**/.nemo-eval-harbor-cache.json")) + + +def test_cache_stamp_survives_harbors_stray_directory_sweep(tmp_path: Path) -> None: + # Harbor rmtree's any *directory* in a job dir lacking result.json. The stamp must + # therefore be a file, or it would be silently deleted on the next Harbor run. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import CACHE_STAMP_FILENAME + + _config, job_dir, _task = _seed_cached_job(tmp_path) + stamp = job_dir / CACHE_STAMP_FILENAME + + assert stamp.is_file() + assert not stamp.is_dir() + + +def test_cache_stamp_handles_a_missing_agent_dir(tmp_path: Path) -> None: + # agent_dir is None for every built-in-agent caller (including nemo-evaluator). + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _cache_stamp + + config, _job_dir, task = _seed_cached_job(tmp_path) + stamp = _cache_stamp(config, Path(str(task.metadata["harbor_dataset_path"])), [task]) + + assert config.agent_dir is None + assert stamp["agent"] == "" + + +def test_unresolvable_task_dir_is_always_stale(tmp_path: Path) -> None: + # A task we cannot locate on disk must never be silently omitted from the + # fingerprint — that would be a stale-cache hole. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _cache_is_stale, _cache_stamp + + config, job_dir, _task = _seed_cached_job(tmp_path) + orphan = AgentEvalTask(id="ghost", intent="x", inputs={"instruction": "x"}, metrics=[HarborRewardMetric()]) + stamp = _cache_stamp(config, tmp_path / "nonexistent-dataset", [orphan]) + + assert stamp["tasks"]["ghost"] == "" + assert _cache_is_stale(job_dir, stamp) is True def test_multiple_attempts_map_to_one_trial_each(tmp_path: Path) -> None: @@ -247,3 +492,845 @@ def test_scoped_agent_import_makes_wrapper_importable_then_cleans_up(tmp_path: P # On exit the injected module and its synthetic package are gone from sys.modules. assert module_name not in sys.modules assert package not in sys.modules + + +def test_digest_ignores_an_exclusion_that_contains_the_whole_tree(tmp_path: Path) -> None: + # jobs_dir sitting *above* the agent/task dir is a legitimate layout. Applying the + # exclusion there would match every entry and yield an empty digest, silently + # disabling invalidation for the entire directory. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _digest_directory + + work = tmp_path / "work" + (work / "agent").mkdir(parents=True) + (work / "agent" / "a.py").write_text("v1") + + before = _digest_directory(work / "agent", exclude=frozenset({work})) + assert before != hashlib.sha256().hexdigest(), "an ancestor exclusion must not empty the digest" + + (work / "agent" / "a.py").write_text("v2-DIFFERENT") + assert _digest_directory(work / "agent", exclude=frozenset({work})) != before + + +def test_digest_still_ignores_a_jobs_dir_nested_inside_the_tree(tmp_path: Path) -> None: + # The case the exclusion actually exists for: results written under the hashed + # tree must not make the fingerprint move on every run. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _digest_directory + + agent = tmp_path / "agent" + (agent / "jobs").mkdir(parents=True) + (agent / "a.py").write_text("src") + (agent / "jobs" / "result.json").write_text("{}") + + before = _digest_directory(agent, exclude=frozenset({agent / "jobs"})) + (agent / "jobs" / "result.json").write_text('{"more": "output"}') + assert _digest_directory(agent, exclude=frozenset({agent / "jobs"})) == before + + +def test_task_dir_outside_the_active_dataset_is_rediscovered(tmp_path: Path) -> None: + # `dataset_path` can be overridden on the runner, so a task stamped during + # discovery under dataset A must not be fingerprinted when Harbor runs dataset B. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _task_dirs_for + + dataset_a, dataset_b = tmp_path / "dsA", tmp_path / "dsB" + for dataset in (dataset_a, dataset_b): + (dataset / "t").mkdir(parents=True) + (dataset / "t" / "task.toml").write_text('[task]\nname = "t"\n') + + task = AgentEvalTask( + id="t", + intent="x", + inputs={"instruction": "x"}, + metadata={"harbor_dataset_path": str(dataset_a), "harbor_task_dir": str(dataset_a / "t")}, + ) + + resolved = _task_dirs_for(dataset_b, [task])["t"] + assert resolved is not None + assert resolved.resolve().is_relative_to(dataset_b.resolve()), "must fingerprint the dataset Harbor runs" + + +def test_vanished_task_dir_is_rediscovered(tmp_path: Path) -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _task_dirs_for + + dataset = tmp_path / "ds" + (dataset / "t").mkdir(parents=True) + (dataset / "t" / "task.toml").write_text('[task]\nname = "t"\n') + task = AgentEvalTask( + id="t", + intent="x", + inputs={"instruction": "x"}, + metadata={"harbor_task_dir": str(tmp_path / "gone" / "t")}, + ) + + assert _task_dirs_for(dataset, [task])["t"] == dataset / "t" + + +def test_symlink_loop_degrades_the_stamp_instead_of_killing_the_run(tmp_path: Path) -> None: + """A loop under a task dir must not take down a run over a best-effort fingerprint. + + Deliberately a *loop*, not a dangling or vanished link. On CPython 3.12 — the + floor this package targets — `Path.resolve()` translates `ELOOP` into + ``RuntimeError``, which is **not** an ``OSError``, so catching only ``OSError`` + leaves this crashing. It also raises deterministically rather than as a race, so + the failure is reproducible rather than occasional. + + What this pins is :func:`_safe_resolve`'s exception set. The loop is reached + through :func:`_digest_directory`'s walk, which already routes every path through + ``_safe_resolve``. It does **not** exercise ``_cache_stamp``'s own resolve calls: + ``_task_dirs_for`` filters candidates with ``is_dir()``, which returns ``False`` + for a loop, so a looping path never reaches them. Those calls use + ``_safe_resolve`` for consistency, not because a live crash was demonstrated. + """ + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import HarborRuntimeConfig, _cache_stamp, _safe_resolve + + dataset = tmp_path / "ds" + task_dir = dataset / "t" + task_dir.mkdir(parents=True) + (task_dir / "task.toml").write_text('[task]\nname = "t"\n') + loop = task_dir / "loop" + loop.symlink_to(loop) + + # Pin the premise: if this stops raising, the guard below is no longer load-bearing. + with pytest.raises(RuntimeError): + loop.resolve() + assert _safe_resolve(loop) == loop.absolute(), "_safe_resolve must swallow the loop, not just OSError" + + task = AgentEvalTask( + id="t", + intent="x", + inputs={"instruction": "x"}, + metadata={"harbor_task_dir": str(task_dir)}, + ) + config = HarborRuntimeConfig(jobs_dir=tmp_path / "jobs", job_name="pinned") + + stamp = _cache_stamp(config, dataset, [task]) + + assert set(stamp["tasks"]) == {"t"}, "the task must still be fingerprinted, not dropped" + + +@pytest.mark.asyncio +async def test_inputs_changing_mid_run_leaves_the_job_unstamped( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # If the candidate is edited while Harbor is running, the results came from the + # OLD sources. Stamping the new fingerprint onto them would let a later run serve + # them as if they matched — so the job dir is deliberately left unstamped. + from nemo_evaluator_sdk.agent_eval.runtimes import harbor_runtime + + config, job_dir, task = _seed_cached_job(tmp_path) + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + (agent_dir / "wrapper.py").write_text("v1\n") + config = config.model_copy(update={"agent_import_path": "wrapper:Agent", "agent_dir": agent_dir}) + (job_dir / harbor_runtime.CACHE_STAMP_FILENAME).unlink() + + def fake(cfg, _dataset_path, _task_names, *, job_name=None, force_rerun=None): + async def run_job() -> None: + (agent_dir / "wrapper.py").write_text("v2-EDITED-MID-RUN\n") + + return cfg.jobs_dir / (job_name or "job"), run_job + + monkeypatch.setattr(harbor_runtime, "_build_native_job", fake) + + await HarborAgentTaskRunner(config=config).run_tasks([task]) + + assert not (job_dir / harbor_runtime.CACHE_STAMP_FILENAME).exists(), ( + "results produced from pre-edit sources must not be stamped with the post-edit fingerprint" + ) + + +def test_digest_is_injective_over_separator_bearing_contents(tmp_path: Path) -> None: + """Distinct trees must never share a digest, even when contents embed the framing. + + A collision here fails CLOSED - the digest matches, so stale results are served. + The historical bug was concatenating ``name \\0 content \\0`` with no length + framing: because file *contents* may contain NUL, ``{a: b"", b: b"Z"}`` and + ``{a: b"\\0b\\0Z"}`` produced an identical byte stream. + + Rather than pin one hand-built pair to one encoding, this asserts the property: + every tree below is structurally different, so every digest must differ. The + contents are chosen to embed the separators an unframed encoding would rely on. + """ + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _digest_directory + + trees: dict[str, dict[str, bytes]] = { + "two_files_empty_then_z": {"a": b"", "b": b"Z"}, + "one_file_absorbing_nul": {"a": b"\0b\0Z"}, + "one_file_absorbing_nul_and_mode": {"a": b"\0b\0-\0Z"}, + "three_files": {"a": b"", "b": b"", "c": b"Z"}, + "two_files_swapped": {"a": b"Z", "b": b""}, + "one_file_named_b": {"b": b"Z"}, + } + + digests: dict[str, str] = {} + for name, files in trees.items(): + root = tmp_path / name + root.mkdir() + for filename, content in files.items(): + (root / filename).write_bytes(content) + digests[name] = _digest_directory(root) + + collisions = { + (left, right) for left in digests for right in digests if left < right and digests[left] == digests[right] + } + assert not collisions, f"distinct trees produced identical digests: {sorted(collisions)}" + + +def test_digest_tracks_the_execute_bit(tmp_path: Path) -> None: + # Harbor discovers and runs tests/test.sh; flipping +x changes what happens + # without changing a single byte of content. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _digest_directory + + task = tmp_path / "task" + task.mkdir() + script = task / "test.sh" + script.write_text("#!/bin/sh\necho hi\n") + + script.chmod(0o644) + non_executable = _digest_directory(task) + script.chmod(0o755) + assert _digest_directory(task) != non_executable, "+x must invalidate" + + +def test_digest_ignores_read_write_permission_noise(tmp_path: Path) -> None: + # Only the execute bit is tracked, mirroring git: umask differences between two + # checkouts of the same sources must not evict a usable cache. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _digest_directory + + task = tmp_path / "task" + task.mkdir() + source = task / "a.py" + source.write_text("x = 1\n") + + source.chmod(0o644) + before = _digest_directory(task) + source.chmod(0o600) + assert _digest_directory(task) == before + + +def test_digest_covers_vendored_dependencies_but_not_the_environment(tmp_path: Path) -> None: + # node_modules ships with the agent and changes what it does, so it counts. + # .venv is environment the Harbor wrapper never uploads, so it does not. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _digest_directory + + agent = tmp_path / "agent" + (agent / "node_modules" / "lib").mkdir(parents=True) + (agent / ".venv").mkdir() + (agent / "main.js").write_text("x") + (agent / "node_modules" / "lib" / "index.js").write_text("v1") + (agent / ".venv" / "marker").write_text("1") + + before = _digest_directory(agent) + (agent / "node_modules" / "lib" / "index.js").write_text("v2-DIFFERENT") + assert _digest_directory(agent) != before, "a vendored dependency change must invalidate" + + after_dep = _digest_directory(agent) + (agent / ".venv" / "marker").write_text("2") + assert _digest_directory(agent) == after_dep, ".venv churn must not evict the cache" + + +def test_digest_survives_a_dangling_symlink(tmp_path: Path) -> None: + # A link whose *target* is missing: is_dir()/is_file() are both False, so it is + # recorded as a marker. (readlink still succeeds here - see the test below for + # the case where readlink itself fails.) + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _digest_directory + + agent = tmp_path / "agent" + agent.mkdir() + (agent / "real.py").write_text("x") + (agent / "dangling").symlink_to(tmp_path / "does-not-exist") + + assert _digest_directory(agent) # no raise + + +def test_digest_distinguishes_a_file_from_a_directory_of_the_same_name(tmp_path: Path) -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _digest_directory + + as_file = tmp_path / "as_file" + as_file.mkdir() + (as_file / "thing").write_text("") + + as_dir = tmp_path / "as_dir" + as_dir.mkdir() + (as_dir / "thing").mkdir() + + assert _digest_directory(as_file) != _digest_directory(as_dir) + + +def test_digest_survives_readlink_failing_mid_walk(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # The link itself disappearing between is_symlink() and readlink is a real race + # against any process cleaning up the tree. It must degrade to a marker rather + # than raise out of run_tasks and kill an otherwise-good evaluation. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _digest_directory + + agent = tmp_path / "agent" + agent.mkdir() + (agent / "real.py").write_text("x") + (agent / "link").symlink_to(agent / "real.py") + + def exploding_readlink(*_args: object, **_kwargs: object) -> str: + raise OSError(2, "No such file or directory") + + 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_same_named_identical_agents_do_not_repoint_an_open_scope(tmp_path: Path) -> None: + # Content-addressing means two directories with the same basename and identical + # contents share a package name. The second install must not swap `__path__` out + # from under the first, still-open scope. + # + # Deliberately NOT fixed by hashing the resolved path into the package name: that + # would make the name location-dependent, so the same agent evaluated from a + # different path would produce a different JobConfig and Harbor would refuse to + # resume — reintroducing AALGO-430. The trees are byte-identical here, so keeping + # the first path is correct. + from nemo_evaluator_sdk.agent_eval.runtimes.harbor_runtime import _AGENT_PACKAGE_REFCOUNTS + + first = tmp_path / "one" / "agent" + second = tmp_path / "two" / "agent" + for agent_dir in (first, second): + agent_dir.mkdir(parents=True) + (agent_dir / "wrapper.py").write_text("VALUE = 1\n") + + with scoped_harbor_agent_import(first, "wrapper:Agent") as outer: + package = outer.rsplit(".", 1)[0] + assert sys.modules[package].__path__ == [str(first)] + with scoped_harbor_agent_import(second, "wrapper:Agent") as inner: + assert inner == outer, "identical contents and basename must share one package" + assert sys.modules[package].__path__ == [str(first)], ( + "the second install must not repoint a scope that is still open" + ) + assert sys.modules[package].__path__ == [str(first)], "the inner exit must not tear down the outer scope" + + assert package not in sys.modules + assert _AGENT_PACKAGE_REFCOUNTS == {} + + +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." + ) + + +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 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..92425b1256 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,68 @@ 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. + + Runs two attempts and drops one, so "resumed" is distinguishable from "discarded + and re-run from scratch" — with a single attempt the two are observationally + identical and the test would pass either way. + """ + 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, + # Two attempts so one can be dropped and one kept. With a single attempt, + # discarding the whole job dir and re-running is observationally identical to + # resuming, and the assertions below could not tell them apart. + n_attempts=2, + ) + + first = await run_harbor_eval(config, _DATASET_DIR) + assert [trial.status for trial in first.trials] == [AgentEvalTrialStatus.COMPLETED] * 2 + + job_dir = jobs_dir / "resume-job" + trial_dirs = sorted(path.parent for path in job_dir.glob("*/result.json")) + assert len(trial_dirs) == 2, "the first run must have written both attempts" + survivor, dropped = trial_dirs + survivor_result = (survivor / "result.json").read_text(encoding="utf-8") + + # Drop one attempt'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 FileExistsError. + (dropped / "result.json").unlink() + + second = await run_harbor_eval(config, _DATASET_DIR) + + # The point of the test: the completed attempt was *resumed*, not re-run. Harbor + # suffixes each trial dir with a shortuuid, so a discarded job dir would come back + # under a different name, and a re-executed trial would rewrite result.json. + assert survivor.is_dir(), "the completed attempt's trial dir must survive the rerun" + assert (survivor / "result.json").read_text(encoding="utf-8") == survivor_result, ( + "the completed attempt must be reused untouched, not re-executed" + ) + assert not dropped.is_dir(), "the result-less attempt must be cleared and re-run" + + assert [trial.status for trial in second.trials] == [AgentEvalTrialStatus.COMPLETED] * 2 + assert {trial.task_id for trial in second.trials} == {_TASK_NAME} + assert [trial.metadata["reward"] for trial in second.trials] == [1.0, 1.0] 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 e04dd9b35d..20d56349d3 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 @@ -32,9 +32,11 @@ from __future__ import annotations import contextlib +import hashlib import importlib.machinery import json import logging +import os import re import shutil import sys @@ -45,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 @@ -71,6 +72,42 @@ _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" +# Public so downstreams can assert the SDK is new enough to own cache staleness. +CACHE_STAMP_VERSION = 1 +# Excluded from the cache fingerprint — see :func:`_cache_stamp` for why each one. +_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 +# deliverable — the Harbor wrapper does not upload them into the task container. +_DIGEST_SKIP_DIRS = frozenset({".git", "__pycache__", ".venv", ".uv", ".mypy_cache", ".pytest_cache"}) +_DIGEST_CHUNK_BYTES = 1 << 20 RunJob = Callable[[], Awaitable[None]] @@ -209,17 +246,78 @@ async def run_tasks( In native mode the dataset directory is recovered from the tasks (each carries ``metadata['harbor_dataset_path']`` from :func:`discover_harbor_tasks`) unless a ``dataset_path`` override was given, - so callers don't repeat it. ``job_dir`` doubles as a cache: the Harbor run - is skipped and results are simply re-adapted when every requested task - already has ``n_attempts`` completed, non-errored results there (unless - ``force_rerun`` is set). The cache only engages when the config pins a - stable ``job_name``; the default timestamped name never hits it. + so callers don't repeat it. + + ``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. + + Assumes a **single writer per job directory**. Neither this runtime nor + Harbor locks it, so two processes sharing a pinned ``job_name`` on a shared + volume will race. """ if self._config is not None: dataset_path = self._dataset_path or _dataset_path_from_tasks(tasks) - job_dir, run_job = _build_native_job(self._config, dataset_path, self._task_names) - if self._config.force_rerun or not _all_tasks_cached(job_dir, tasks, n_attempts=self._config.n_attempts): + job_name, job_dir = _resolve_job_dir(self._config) + + # Only fingerprint when the answer can depend on it: an unpinned job name + # can never hit, and force_rerun/a missing dir already decided. This keeps + # the digest I/O off every run of callers that don't pin a job name. + stamp: dict[str, Any] | None = None + if self._config.job_name is None or self._config.force_rerun or not job_dir.is_dir(): + stale = True + else: + stamp = _cache_stamp(self._config, dataset_path, tasks) + stale = _cache_is_stale(job_dir, stamp) + + if stale or not _all_tasks_cached(job_dir, tasks, n_attempts=self._config.n_attempts): + job_dir, run_job = _build_native_job( + self._config, + dataset_path, + self._task_names, + job_name=job_name, + # 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 + # results produced from the old sources with the new fingerprint, so + # a later run would happily serve them. Covers what Harbor actually + # ran: with no task_names filter that is the whole dataset, and + # recording only the requested subset would make the next full-set + # run look stale and re-run a complete job dir. + coverage = _stamp_coverage(dataset_path, tasks, self._task_names) + before = _cache_stamp(self._config, dataset_path, coverage) await run_job() + if self._config.job_name is not None: + after = _cache_stamp(self._config, dataset_path, coverage) + if after == before: + _write_cache_stamp(job_dir, after) + else: + # Leaving it unstamped re-runs next time, which is the safe + # direction: we cannot say which inputs produced these results. + logger.warning( + "Agent or task contents changed while Harbor job %s was running; leaving it unstamped " + "so the next run re-executes rather than trusting these results.", + job_dir, + ) return build_trials_from_job_dir(job_dir, tasks, reward_key=self._reward_key) if self._job_dir is None: # unreachable: __init__ guarantees config or job_dir @@ -268,10 +366,352 @@ def _all_tasks_cached(job_dir: Path, tasks: Sequence[AgentEvalTask], *, n_attemp return all(counts.get(task.id, 0) >= n_attempts for task in tasks) +def _feed(digest: "hashlib._Hash", label: bytes, payload: bytes) -> None: + """Append a length-framed field to ``digest``. + + Framing matters: concatenating ``name \0 content \0`` is ambiguous because file + *contents* may contain NUL, so two different trees can produce an identical byte + stream. Prefixing every variable-length field with its length makes the encoding + injective, which is what stops a collision from being read as "unchanged". + """ + digest.update(label) + digest.update(len(payload).to_bytes(8, "big")) + digest.update(payload) + + +def _safe_resolve(path: Path) -> Path: + """``Path.resolve()`` that degrades instead of raising. + + The digest is a best-effort guard, not a reason to fail a run that would + otherwise succeed, so fall back to the unresolved absolute path. + + ``RuntimeError`` is caught alongside ``OSError`` and is the case that actually + fires: on CPython 3.12 — the floor this package targets — a **symlink loop** + surfaces as ``RuntimeError("Symlink loop from ...")``, because ``resolve()`` + translates ``ELOOP`` before re-raising. It is not an ``OSError``, so catching + only that would let a loop under a task directory kill the run. A loop raises + deterministically, not as a race. ``OSError`` covers the narrower case of a + symlink that disappears mid-walk, since ``resolve()`` calls ``os.readlink``. + + Both are 3.12/3.13 behaviours: 3.14 resolves a loop without raising at all. + """ + try: + return path.resolve() + except (OSError, RuntimeError): + return path.absolute() + + +def _is_executable(path: Path) -> bool: + """Whether the owner-execute bit is set, following symlinks. + + Only the execute bit, mirroring what git tracks: read/write bits vary with umask + and would evict the cache for nothing, but flipping +x on ``tests/test.sh`` or an + agent entrypoint genuinely changes what Harbor does. + """ + try: + return bool(path.stat().st_mode & 0o100) + except OSError: + return False + + +def _digest_directory(root: Path, *, exclude: frozenset[Path] = frozenset()) -> str: + """Content-hash a directory tree, skipping build/VCS noise and excluded roots. + + Contents rather than mtimes: callers routinely materialize the directory with + ``copytree`` (the optimizer does, per candidate), which rewrites every mtime and + would defeat the cache entirely. + + ``exclude`` takes *resolved* directories to skip wholesale. It exists because + ``jobs_dir`` is caller-chosen and may sit **under** the dataset or agent + directory; without excluding it the digest would hash the growing results tree + it is meant to validate, and would never stabilize. + + Symlinks are **followed**, not skipped. Skipping them silently defeats the whole + guard: a directory assembled out of links to shared sources would hash to the + empty digest, so edits behind those links would never invalidate the cache. The + link target is folded in alongside the contents, so re-pointing a link is a + change even when both targets happen to hold identical bytes. + + Unreadable or vanished files are folded in as a marker rather than raised: a + transient read failure must not kill a run that would otherwise have succeeded. + """ + digest = hashlib.sha256() + if not root.is_dir(): + return digest.hexdigest() + # Keep only exclusions strictly *inside* this tree. An excluded root that is an + # ancestor of (or equal to) `root` would otherwise match every entry and yield + # an empty digest — silently disabling invalidation for the whole directory, + # which is exactly the failure this function exists to prevent. jobs_dir being + # a parent of the agent/task dir is a legitimate layout, not a reason to stop + # hashing it. + root_resolved = _safe_resolve(root) + excluded = { + resolved + for resolved in (_safe_resolve(path) for path in exclude) + if resolved != root_resolved and resolved.is_relative_to(root_resolved) + } + # Resolved dirs already walked, so a symlink cycle terminates instead of hanging. + visited: set[Path] = set() + + def walk(directory: Path) -> None: + resolved_dir = _safe_resolve(directory) + if resolved_dir in visited: + return + visited.add(resolved_dir) + try: + entries = sorted(directory.iterdir()) + except OSError as exc: + logger.warning("Could not list %s while fingerprinting %s: %s", directory, root, exc) + _feed(digest, b"unlistable", b"") + return + for path in entries: + if path.name in _DIGEST_SKIP_DIRS: + continue + resolved = _safe_resolve(path) + if any(resolved == item or item in resolved.parents for item in excluded): + continue + + _feed(digest, b"name", path.relative_to(root).as_posix().encode("utf-8")) + _feed(digest, b"mode", b"x" if _is_executable(path) else b"-") + if path.is_symlink(): + # Record where the link points, so retargeting counts as a change. + try: + target = os.readlink(path).encode("utf-8") + except OSError as exc: + # The link vanished mid-walk. Same contract as an unreadable + # file: degrade to a marker rather than kill the run. + logger.warning("Could not read link %s while fingerprinting %s: %s", path, root, exc) + target = b"" + _feed(digest, b"symlink", target) + if path.is_dir(): + _feed(digest, b"dir", b"") + walk(path) + continue + if not path.is_file(): + # Broken link, socket, fifo: nothing to hash, but its presence counts. + _feed(digest, b"not-a-file", b"") + continue + content = hashlib.sha256() + try: + with path.open("rb") as handle: + # Streamed: Harbor datasets may ship large seeds or build contexts. + for chunk in iter(lambda: handle.read(_DIGEST_CHUNK_BYTES), b""): + content.update(chunk) + except OSError as exc: + logger.warning("Could not read %s while fingerprinting %s: %s", path, root, exc) + content.update(b"") + # The sub-digest is fixed width, so file bytes can never be confused with + # the framing around them. + _feed(digest, b"file", content.digest()) + + walk(root) + return digest.hexdigest() + + +def _task_dirs_for(dataset_path: Path, tasks: Sequence[AgentEvalTask]) -> dict[str, Path | None]: + """Resolve each task's on-disk directory, falling back to re-discovery. + + :func:`discover_harbor_tasks` stamps ``metadata['harbor_task_dir']``, but callers + may build :class:`AgentEvalTask` objects by hand (the Evaluator plugin builds + them from a job spec), so the metadata is not guaranteed. + + A task that cannot be resolved maps to ``None`` — the caller must treat that as + un-cacheable rather than silently omitting it from the fingerprint, which would + be a stale-cache hole. + """ + # `_safe_resolve`, not bare `resolve()`: this walk is a best-effort cache guard, so + # a symlink that vanishes mid-run must degrade to an unresolved absolute path + # rather than raise out of a job that would otherwise succeed. + dataset_root = _safe_resolve(dataset_path) + resolved: dict[str, Path | None] = {} + for task in tasks: + stamped = task.metadata.get("harbor_task_dir") + candidate = Path(stamped) if isinstance(stamped, str) and stamped else None + # The stamp records where a task was *discovered*, which is not necessarily + # where this run executes it: `dataset_path` can be overridden on the runner. + # Trusting a stale or foreign path would fingerprint one dataset while Harbor + # runs another, so anything missing or outside the active dataset is dropped + # and re-discovered below. + if candidate is not None: + candidate_resolved = _safe_resolve(candidate) + if not candidate.is_dir() or not candidate_resolved.is_relative_to(dataset_root): + logger.debug( + "Ignoring stamped harbor_task_dir %s for task %r: not a directory under the active dataset %s", + candidate, + task.id, + dataset_root, + ) + candidate = None + resolved[task.id] = candidate + if all(path is not None for path in resolved.values()): + return resolved + + try: + discovered = { + task.id: Path(str(task.metadata["harbor_task_dir"])) for task in discover_harbor_tasks(dataset_path) + } + except (OSError, ValueError) as exc: + # discover_harbor_tasks raises on ANY malformed task.toml in the dataset. + # Refusing the cache is the safe reading; failing the run is not, since this + # path previously never read those files. + logger.warning( + "Could not resolve Harbor task dirs under %s; treating the cache as stale: %s", dataset_path, exc + ) + return dict.fromkeys(resolved, None) + return {task_id: path or discovered.get(task_id) for task_id, path in resolved.items()} + + +def _stamp_coverage( + dataset_path: Path, + tasks: Sequence[AgentEvalTask], + task_names: Sequence[str] | None, +) -> Sequence[AgentEvalTask]: + """Tasks a written stamp must cover: everything Harbor was asked to run. + + ``task_names`` is the filter handed to Harbor's ``DatasetConfig``. When it is + ``None`` Harbor runs every task in the dataset, which can be a superset of the + tasks this call was asked to score — and a stamp that recorded only the smaller + set would report the larger one as stale on the next run. + """ + if task_names is not None: + return tasks + try: + discovered = discover_harbor_tasks(dataset_path) + except (OSError, ValueError): + # Same reasoning as _task_dirs_for: a malformed sibling task must not fail a + # run. Recording only the requested tasks just costs a re-run later. + return tasks + covered = {task.id: task for task in discovered} + covered.update({task.id: task for task in tasks}) + return list(covered.values()) + + +def _cache_stamp( + config: HarborRuntimeConfig, + dataset_path: Path, + tasks: Sequence[AgentEvalTask], +) -> dict[str, Any]: + """Fingerprint the inputs that decide whether a job dir can be reused. + + Covers the result-affecting options, the contents of ``agent_dir``, and the + contents of every task directory. Two gaps are deliberate and worth knowing + before trusting a hit: when ``agent_dir`` is ``None`` the agent is an already + importable module, so only its *import path* is fingerprinted and edits to that + installed package are invisible; and a task whose directory cannot be resolved + is recorded as ````, which always forces a re-run. + + Recorded per task rather than as one job-wide hash so that evaluating a + **subset** of a previously-cached job still hits: staleness is decided only over + the tasks actually requested. + + Excluded from the option hash: presentation and placement knobs (``quiet``, + ``n_concurrent_trials``, ``jobs_dir``, ``job_name``, ``force_rerun``), which + change nothing about the results; ``agent_dir``, an absolute path whose + *content* is hashed separately, so a relocated but identical agent still hits; + and ``reward_key``, which only selects which reward + :func:`build_trials_from_job_dir` reads back and must not cost a Docker re-run. + """ + options = config.model_dump(exclude=set(_CACHE_IRRELEVANT_OPTIONS), mode="json") + # `_safe_resolve` throughout, matching `_task_dirs_for`: fingerprinting is + # best-effort, so a symlink loop or a vanished link under any of these must + # degrade to an unresolved path rather than raise out of `run_tasks` and fail a + # run that would otherwise succeed. + excluded_roots = frozenset({_safe_resolve(config.jobs_dir.expanduser())}) + + agent_digest = "" + if config.agent_dir is not None: + agent_digest = _digest_directory(_safe_resolve(config.agent_dir.expanduser()), exclude=excluded_roots) + + task_digests: dict[str, str] = {} + for task_id, task_dir in sorted(_task_dirs_for(dataset_path, tasks).items()): + task_digests[task_id] = ( + "" if task_dir is None else _digest_directory(_safe_resolve(task_dir), exclude=excluded_roots) + ) + + return { + "version": CACHE_STAMP_VERSION, + "options": hashlib.sha256(json.dumps(options, sort_keys=True, default=str).encode("utf-8")).hexdigest(), + "agent": agent_digest, + "tasks": task_digests, + } + + +def _cache_is_stale(job_dir: Path, stamp: Mapping[str, Any]) -> bool: + """Return True when ``job_dir`` was not produced by the inputs in ``stamp``. + + A directory with no stamp is stale: it predates this check, or was written by + plain Harbor, and re-running is the safe reading. An ```` task + digest is likewise always stale — we could not prove the inputs match. A + directory that does not exist is stale too: there is nothing there to reuse, and + answering "not stale" would be an invitation to serve zero trials. + """ + if not job_dir.is_dir(): + return True + try: + stored = json.loads((job_dir / CACHE_STAMP_FILENAME).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + stored = None + + reason: str | None = None + if not isinstance(stored, Mapping): + reason = "no usable cache stamp" + elif stored.get("version") != stamp["version"]: + reason = "cache stamp version changed" + elif stored.get("options") != stamp["options"]: + reason = "a result-affecting option changed" + elif stored.get("agent") != stamp["agent"]: + reason = "the agent directory changed" + else: + stored_tasks = stored.get("tasks") + stored_tasks = stored_tasks if isinstance(stored_tasks, Mapping) else {} + for task_id, digest in stamp["tasks"].items(): + if digest == "": + reason = f"task {task_id!r} could not be resolved on disk" + break + if stored_tasks.get(task_id) != digest: + reason = f"task {task_id!r} changed or was not part of the cached run" + break + + if reason is None: + return False + logger.info("Re-running Harbor job %s instead of serving it from cache: %s.", job_dir, reason) + return True + + +def _write_cache_stamp(job_dir: Path, stamp: Mapping[str, Any]) -> None: + """Record the inputs a completed job dir was produced from. + + Best-effort: a job dir that could not be stamped simply re-runs next time, which + is the safe direction. Written as a *file* deliberately — Harbor deletes any + stray *directory* in a job dir that lacks ``result.json``. + """ + try: + (job_dir / CACHE_STAMP_FILENAME).write_text( + json.dumps(stamp, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + except OSError as exc: + logger.warning("Could not stamp Harbor job dir %s with its cache key: %s", job_dir, exc) + + +def _resolve_job_dir(config: HarborRuntimeConfig) -> tuple[str, Path]: + """Resolve ``(job_name, job_dir)`` without importing Harbor. + + Split out from :func:`_build_native_job` because the caller must know the job + directory *before* deciding whether to run: an unpinned ``job_name`` is a + timestamp with microsecond precision, so resolving it twice would yield two + different directories and the cache decision would be made about the wrong one. + """ + job_name = config.job_name or datetime.now(timezone.utc).strftime("%Y-%m-%d__%H-%M-%S__%f") + return job_name, config.jobs_dir / job_name + + def _build_native_job( config: HarborRuntimeConfig, dataset_path: Path, task_names: Sequence[str] | None, + *, + job_name: str | None = None, + force_rerun: bool | None = None, ) -> tuple[Path, RunJob]: """Build a Harbor ``JobConfig`` from ``config`` and return ``(job_dir, run_job)``. @@ -280,9 +720,18 @@ def _build_native_job( without importing Harbor. When ``agent_import_path`` is set, ``run_job`` scopes the user's agent package into ``sys.modules`` for the run and removes it afterwards (see :func:`scoped_harbor_agent_import`). + + Args: + job_name: Pre-resolved job name from :func:`_resolve_job_dir`. Pass it when + the caller already resolved the directory, so an unpinned name is not + re-generated into a different timestamp. + force_rerun: Overrides ``config.force_rerun`` for this build. Passed rather + than applied via ``model_copy`` so the caller's config is never mutated + and the job name stays fixed. """ - job_name = config.job_name or datetime.now(timezone.utc).strftime("%Y-%m-%d__%H-%M-%S__%f") - job_dir = config.jobs_dir / job_name + resolved_name = job_name if job_name is not None else _resolve_job_dir(config)[0] + job_dir = config.jobs_dir / resolved_name + effective_force_rerun = config.force_rerun if force_rerun is None else force_rerun async def run_job() -> None: try: @@ -295,7 +744,7 @@ async def run_job() -> None: '(it requires Python >=3.12). Install it separately: uv pip install "harbor>=0.16.1"' ) from exc - if config.force_rerun and job_dir.exists(): + if effective_force_rerun and job_dir.exists(): shutil.rmtree(job_dir) artifacts: list[str | ArtifactConfig] = list(config.artifacts) @@ -316,7 +765,7 @@ async def run_job() -> None: async def _create_and_run(agent: Any) -> None: job_config = JobConfig( - job_name=job_name, + job_name=resolved_name, jobs_dir=config.jobs_dir, n_attempts=config.n_attempts, n_concurrent_trials=config.n_concurrent_trials, @@ -327,15 +776,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. @@ -344,13 +830,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 @@ -359,9 +905,41 @@ 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. + + **That last sentence only holds if the caller derives its job name from the + *resolved* directory, as this function does.** Deriving it from the caller's + spelling instead lets the two disagree: a symlink keeps its own name while + resolving elsewhere, so flipping it at a fixed ``job_name`` would reuse one job + dir for two different agents, caught only by Harbor's refusal rather than by + design. The Experimentalist resolves first for exactly this reason + (``resolve_harbor_run_inputs``). 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 @@ -373,7 +951,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: @@ -393,7 +974,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]) @@ -404,11 +989,41 @@ def _install_agent_package(package: str, agent_dir: Path) -> None: sys.modules[name] = module if idx > 1: setattr(sys.modules[".".join(parts[: idx - 1])], parts[idx - 1], module) - sys.modules[package].__path__ = [str(agent_dir)] + installed = sys.modules[package] + if not installed.__path__: + installed.__path__ = [str(agent_dir)] + elif installed.__path__ != [str(agent_dir)]: + # Two directories sharing this package name share a content digest, so their + # trees are byte-identical and the path already installed is exactly as + # correct as this one — the excluded content (`.git`, `__pycache__`, the + # env dirs, `jobs_dir`) is not importable. Repointing would swap the + # directory out from under a scope that is still open, for no gain. + logger.debug( + "Agent package %s is already installed from %s; keeping it for the identical tree at %s", + package, + installed.__path__[0], + 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(".") @@ -731,6 +1346,8 @@ def reward_payload_from_result( __all__ = [ + "CACHE_STAMP_FILENAME", + "CACHE_STAMP_VERSION", "DEFAULT_REWARD_KEY", "HarborAgentTaskRunner", "HarborRewardMetric",