From 229a2fea2e3e5219a1494e6fd3adbfd556f4937a Mon Sep 17 00:00:00 2001 From: Sandy Chapman Date: Fri, 3 Jul 2026 13:58:36 -0300 Subject: [PATCH] feat(evaluator): add NeMo Fabric agent-eval runner with ATIF trajectory capture Add a `fabric` agent-eval runner kind alongside `codex`: FabricAgentRuntime drives an agent harness through the NeMo Fabric SDK, maps each RunResult to an AgentEvalTrial, and captures the agent trajectory as ATIF (NeMo Relay file exporter) into trial evidence under the standard trace key so graders can consume it. - FabricRunnerTarget(kind="fabric") + widened AgentRunnerTarget union + _resolve_target wiring - ATIF trajectory capture -> EVIDENCE_TRACE(format=atif); telemetry uri/trace_id preserved - unit tests + hermetic CI integration test + gated live fabric->codex e2e - regenerated plugin openapi + vendored SDK mirror Co-Authored-By: Claude Opus 4.8 Signed-off-by: Sandy Chapman --- .../agent_eval/runtimes/fabric/runtime.py | 366 ++++++++++++++++++ .../agent_eval/test_fabric_integration.py | 245 ++++++++++++ .../tests/agent_eval/test_fabric_runtime.py | 325 ++++++++++++++++ plugins/nemo-evaluator/openapi/openapi.yaml | 57 +++ .../src/nemo_evaluator/jobs/agent_evaluate.py | 12 + .../src/nemo_evaluator/jobs/agent_spec.py | 36 +- .../tests/test_agent_evaluate.py | 24 ++ script/dev-install-fabric.sh | 83 ++++ .../agent_eval/runtimes/fabric/runtime.py | 366 ++++++++++++++++++ 9 files changed, 1512 insertions(+), 2 deletions(-) create mode 100644 packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/runtime.py create mode 100644 packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_integration.py create mode 100644 packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_runtime.py create mode 100755 script/dev-install-fabric.sh create mode 100644 sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/runtime.py diff --git a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/runtime.py b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/runtime.py new file mode 100644 index 00000000000..44e120d0a30 --- /dev/null +++ b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/runtime.py @@ -0,0 +1,366 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""NeMo Fabric-backed agent-eval runtime. + +``FabricAgentRuntime`` drives an agent harness (Codex, Hermes, ...) through the +NeMo Fabric Python SDK and adapts each normalized Fabric ``RunResult`` into an +:class:`AgentEvalTrial`. The harness is chosen by the supplied Fabric config's +``harness.adapter_id`` (never inferred from a model); an optional ``model`` slug +is applied as a final profile overlay, mirroring Fabric's own Harbor integration. + +``nemo_fabric`` is an optional native dependency: its types are imported for +annotations under ``TYPE_CHECKING`` and the package is loaded lazily at runtime, +so this module stays importable without it. +""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask +from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrial, AgentEvalTrialStatus, AgentOutput +from nemo_evaluator_sdk.values.evidence import ( + EVIDENCE_FORMAT_ATIF, + EVIDENCE_TRACE, + CandidateEvidence, + EvidenceDescriptor, +) + +if TYPE_CHECKING: + # Annotations use nemo_fabric's real types (single source of truth). nemo_fabric is an optional + # native package not yet in our locked dependency set, so it is imported for typing only and + # loaded lazily at runtime (see ``run_tasks``). Drop the ty:ignore once nemo-fabric is a + # resolvable dependency and the checker can see it. + from nemo_fabric import ( # ty: ignore[unresolved-import] + FabricClient, + FabricConfig, + FabricProfileConfig, + RunResult, + ) + +DEFAULT_FABRIC_TIMEOUT_S = 600 +_RUNTIME_NAME = "fabric" +_MISSING_FABRIC_MSG = "FabricAgentRuntime requires the `nemo-fabric` package (native NeMo Fabric SDK)." +_MISSING_RELAY_MSG = ( + "FabricAgentRuntime trajectory capture requires the `nemo-relay` package " + "(install `nemo-fabric[relay]`), or set capture_trajectory=False." +) + +# Evidence-dir layout for trajectory capture. These subdir names are our own local layout — we create +# them and hand them to Fabric/Relay, so they are not derived from either library. +_RELAY_SUBDIR = "relay" +_ARTIFACTS_SUBDIR = "artifacts" +# Trajectory profile identity + the file-exporter output names we choose (Relay accepts these as inputs). +_TRAJECTORY_PROFILE_NAME = "eval_trajectory" +_ATIF_FILENAME_TEMPLATE = "trajectory-{session_id}.atif.json" +_ATOF_FILENAME = "events.atof.jsonl" +# Fabric telemetry-profile selectors (file exporter, no OTLP endpoint). +_TELEMETRY_PROVIDER = "relay" +_TELEMETRY_MODE = "sdk" +# ``kind`` Fabric stamps on the promoted Relay ATIF artifact; used to surface it as trace evidence. +_ATIF_ARTIFACT_KIND = "atif" + + +class FabricAgentRuntime: + """AgentTaskRunner that generates trials by running tasks through NeMo Fabric.""" + + def __init__( + self, + *, + config: Mapping[str, Any], + profiles: Sequence[Mapping[str, Any]] | None = None, + model: str | None = None, + base_dir: str | Path | None = None, + work_root: str | Path | None = None, + timeout_s: int = DEFAULT_FABRIC_TIMEOUT_S, + capture_trajectory: bool = True, + runtime_name: str = _RUNTIME_NAME, + ) -> None: + self._config = config + self._profiles = list(profiles or []) + self._model = model + self._base_dir = Path(base_dir).expanduser() if base_dir is not None else None + self._work_root = Path(work_root).expanduser() if work_root is not None else None + self._timeout_s = timeout_s + self._capture_trajectory = capture_trajectory + self._runtime_name = runtime_name + + async def run_tasks( + self, + tasks: Sequence[AgentEvalTask], + config: AgentEvalRunConfig | None = None, + ) -> Sequence[AgentEvalTrial]: + try: + # nemo_fabric ships a native (pyo3) core and is an optional dependency, so it is imported + # lazily here rather than at module load. + from nemo_fabric import FabricClient, FabricConfig, FabricProfileConfig # ty: ignore[unresolved-import] + except ImportError as exc: + raise RuntimeError(_MISSING_FABRIC_MSG) from exc + + resolved_config = config or AgentEvalRunConfig() + agent_config = FabricConfig.from_mapping(self._config) + base_profiles = self._build_profiles(FabricProfileConfig) + semaphore = asyncio.Semaphore(resolved_config.parallelism) + + async with FabricClient() as client: + + async def run_one(index: int, task: AgentEvalTask) -> AgentEvalTrial: + async with semaphore: + return await self._run_task( + client, agent_config, base_profiles, FabricProfileConfig, index, task, resolved_config + ) + + return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) + + async def _run_task( + self, + client: FabricClient, + agent_config: FabricConfig, + base_profiles: list[FabricProfileConfig], + profile_cls: type[FabricProfileConfig], + index: int, + task: AgentEvalTask, + config: AgentEvalRunConfig, + ) -> AgentEvalTrial: + evidence_dir = self._evidence_dir(index, task, config) + evidence_dir.mkdir(parents=True, exist_ok=True) + + profiles = list(base_profiles) + if self._capture_trajectory: + # Enable Relay's ATIF file exporter, writing the trajectory under this task's durable + # evidence dir; Fabric promotes the resulting file into RunResult.artifacts. Both the + # Fabric artifact root and the relay output dir must exist and be durable. + relay_dir = evidence_dir / _RELAY_SUBDIR + artifacts_dir = evidence_dir / _ARTIFACTS_SUBDIR + relay_dir.mkdir(parents=True, exist_ok=True) + artifacts_dir.mkdir(parents=True, exist_ok=True) + profiles.append(self._trajectory_profile(profile_cls, relay_dir=relay_dir, artifacts_dir=artifacts_dir)) + + try: + result = await asyncio.wait_for( + client.run( + agent_config, + profiles=profiles, + input=_fabric_input(task), + request_id=task.id, + base_dir=self._base_dir, + ), + timeout=self._timeout_s, + ) + except TimeoutError as exc: + return self._failed_trial(task, evidence_dir, exc) + except Exception as exc: # noqa: BLE001 - a task failure must not abort the whole run + return self._failed_trial(task, evidence_dir, exc) + + return self._to_trial(task, result, evidence_dir) + + def _to_trial(self, task: AgentEvalTask, result: RunResult, evidence_dir: Path) -> AgentEvalTrial: + # Persist the full normalized Fabric result so graders (and debugging) can see the raw + # envelope, and expose it as an evidence descriptor. + result_path = evidence_dir / "fabric_result.json" + result_path.write_text(json.dumps(result.to_mapping(), indent=2, default=str), encoding="utf-8") + + base_metadata: dict[str, Any] = { + "runtime": self._runtime_name, + "harness": result.harness, + "adapter_id": result.adapter_id, + "adapter_kind": result.adapter_kind, + "invocation_id": result.invocation_id, + "agent_model": self._model, + } + + if result.status != "succeeded": + return self._failed_trial(task, evidence_dir, _result_error(result), extra_metadata=base_metadata) + + return AgentEvalTrial( + id=f"{task.id}:fabric", + task_id=task.id, + status=AgentEvalTrialStatus.COMPLETED, + output=AgentOutput( + output_text=_extract_output_text(result.output), + response=result.output, + metadata={**base_metadata, "evidence_dir": str(evidence_dir)}, + ), + evidence=self._evidence(result, result_path), + metadata={**base_metadata, "generated": True}, + ) + + def _evidence(self, result: RunResult, result_path: Path) -> CandidateEvidence: + descriptors: dict[str, EvidenceDescriptor] = { + "result": EvidenceDescriptor(kind="json", format="json", ref=str(result_path)), + } + for artifact in result.artifacts.artifacts: + descriptors[artifact.name] = EvidenceDescriptor( + kind=artifact.kind or "file", + ref=str(artifact.path), + metadata={"media_type": artifact.media_type}, + ) + # Surface the Relay ATIF trajectory under the standard trace evidence key so graders + # that consume a normalized trajectory find it. + if artifact.kind == _ATIF_ARTIFACT_KIND: + descriptors[EVIDENCE_TRACE] = EvidenceDescriptor( + kind=EVIDENCE_TRACE, + format=EVIDENCE_FORMAT_ATIF, + ref=str(artifact.path), + ) + return CandidateEvidence( + descriptors=descriptors, + metadata={ + "runtime": self._runtime_name, + "harness": result.harness, + "telemetry": [ + {"provider": ref.provider, "kind": ref.kind, "uri": ref.uri, "trace_id": ref.trace_id} + for ref in result.telemetry + ], + "events": [{"kind": event.kind, "message": event.message} for event in result.events], + }, + ) + + def _failed_trial( + self, + task: AgentEvalTask, + evidence_dir: Path, + error: Exception | Mapping[str, Any], + *, + extra_metadata: Mapping[str, Any] | None = None, + ) -> AgentEvalTrial: + if isinstance(error, Mapping): + error_type = str(error.get("code") or error.get("stage") or "FabricError") + error_message = str(error.get("message") or error) + else: + error_type = error.__class__.__name__ + error_message = str(error) + error_path = evidence_dir / "error.json" + error_path.write_text(json.dumps({"error_type": error_type, "error": error_message}) + "\n", encoding="utf-8") + return AgentEvalTrial( + id=f"{task.id}:fabric", + task_id=task.id, + status=AgentEvalTrialStatus.FAILED, + output=None, + evidence=CandidateEvidence( + descriptors={"error": EvidenceDescriptor(kind="error", format="json", ref=str(error_path))}, + metadata={"runtime": self._runtime_name}, + ), + metadata={ + **(dict(extra_metadata) if extra_metadata else {}), + "runtime": self._runtime_name, + "error_type": error_type, + "error": error_message, + }, + ) + + def _build_profiles(self, profile_cls: type[FabricProfileConfig]) -> list[FabricProfileConfig]: + profiles = [profile_cls.from_mapping(profile) for profile in self._profiles] + if self._model: + # Apply the model as a final profile overlay (mirrors nemo_fabric.integrations.harbor). + provider = self._model.split("/", maxsplit=1)[0] if "/" in self._model else "openai" + profiles.append( + profile_cls( + name="eval_model", + models={"default": {"provider": provider, "model": self._model}}, + ) + ) + return profiles + + def _trajectory_profile( + self, profile_cls: type[FabricProfileConfig], *, relay_dir: Path, artifacts_dir: Path + ) -> FabricProfileConfig: + # Relay ATIF/ATOF file exporter (mode=sdk): the harness emits its trajectory to a local + # nemo-relay gateway, which writes ``trajectory-*.atif.json`` under ``relay_dir``. No OTLP + # collector endpoint is involved. Requires the ``nemo-relay`` gateway on PATH in the runtime. + # The Fabric artifact root is pinned to a durable dir so the promoted trajectory persists. + # + # The observability component is built from nemo_relay's own typed config objects so Relay owns + # its schema (no hand-maintained dict that silently drifts when Relay changes it); imported + # lazily since nemo-relay, like nemo-fabric, is an optional native dependency. ``schema_version`` + # is omitted — ``FabricProfileConfig`` defaults it. + try: + from nemo_relay.observability import ( # ty: ignore[unresolved-import] + AtifConfig, + AtofConfig, + ComponentSpec, + ObservabilityConfig, + ) + except ImportError as exc: + raise RuntimeError(_MISSING_RELAY_MSG) from exc + + relay_dir_str = str(relay_dir) + artifacts_dir_str = str(artifacts_dir) + observability = ComponentSpec( + config=ObservabilityConfig( + atif=AtifConfig( + enabled=True, + output_directory=relay_dir_str, + filename_template=_ATIF_FILENAME_TEMPLATE, + agent_name=self._runtime_name, + agent_version="fabric", + ), + atof=AtofConfig( + enabled=True, + output_directory=relay_dir_str, + filename=_ATOF_FILENAME, + mode="overwrite", + ), + ) + ) + return profile_cls.from_mapping( + { + "name": _TRAJECTORY_PROFILE_NAME, + "description": "Capture the agent trajectory as ATIF via the NeMo Relay file exporter.", + "runtime": {"artifacts": artifacts_dir_str}, + "environment": {"artifacts": artifacts_dir_str}, + "telemetry": { + "enabled": True, + "provider": _TELEMETRY_PROVIDER, + "mode": _TELEMETRY_MODE, + "output_dir": relay_dir_str, + "config": {"version": 1, "components": [observability.to_dict()]}, + }, + } + ) + + def _evidence_dir(self, index: int, task: AgentEvalTask, config: AgentEvalRunConfig) -> Path: + root = self._work_root + if root is None: + root = (config.output_dir or Path.cwd()) / "evidence" / "fabric" + safe_task_id = _safe_path_name(task.id) + task_dir = f"{index:06d}-{safe_task_id}" if safe_task_id else f"task-{index:06d}" + return Path(root) / task_dir + + +def _fabric_input(task: AgentEvalTask) -> str: + return f"Task id: {task.id}\nIntent: {task.intent}\nInputs: {task.inputs}\n" + + +def _extract_output_text(output: object) -> str | None: + """Pull the user-visible message out of a Fabric ``RunResult.output`` (JSON-shaped). + + Harness outputs vary; adapters commonly nest the final message under ``response`` (the codex-cli + adapter does). Prefer a string ``response``/``output_text``, else stringify the whole value. + """ + if output is None: + return None + if isinstance(output, str): + return output + if isinstance(output, Mapping): + for key in ("response", "output_text", "text", "message"): + value = output.get(key) + if isinstance(value, str): + return value + return json.dumps(output, default=str) + + +def _result_error(result: RunResult) -> Mapping[str, Any]: + error = result.error + if error is None: + return {"code": result.status, "message": "Fabric run did not succeed"} + return {"stage": error.stage, "code": error.code, "message": error.message} + + +def _safe_path_name(value: str) -> str: + return "".join(char if char.isalnum() or char in "._-" else "-" for char in value).strip(".-")[:120] diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_integration.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_integration.py new file mode 100644 index 00000000000..39870b6020a --- /dev/null +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_integration.py @@ -0,0 +1,245 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Integration tests: a Fabric-driven agent eval end-to-end, scored by a metric that consumes the +captured trajectory evidence. + +- ``test_fabric_runner_eval_exposes_trajectory_to_metric`` is hermetic (fake ``nemo_fabric``) and runs + in CI: it proves the runner -> evaluator -> metric -> evidence chain, i.e. the metric receives and + reads the trajectory (ATIF) evidence for the task. +- ``test_fabric_codex_live_eval_captures_atif_trajectory`` is the real fabric->codex->Relay run, gated + behind the required binaries/checkout so CI skips it; run it locally after + ``script/dev-install-fabric.sh``. +""" + +from __future__ import annotations + +import importlib.util +import json +import os +import shutil +import sys +import types +from pathlib import Path +from typing import Any + +import pytest +from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator +from nemo_evaluator_sdk.agent_eval.runtimes.fabric import runtime as fabric_runtime +from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask +from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult +from nemo_evaluator_sdk.values.evidence import EVIDENCE_FORMAT_ATIF, EVIDENCE_TRACE + + +class _TrajectoryEvidenceMetric: + """Scores 1.0 iff the trial exposes a readable ATIF trajectory with at least one step. + + Exercises exactly the concern under test: a grader receives the candidate evidence and can locate + + read the captured trajectory (the ``trace`` descriptor) for the task. + """ + + @property + def type(self) -> str: + return "has-trajectory" + + def output_spec(self) -> list[MetricOutputSpec]: + return [MetricOutputSpec.boolean("has_trajectory")] + + async def compute_scores(self, input: MetricInput) -> MetricResult: # noqa: A002 - matches protocol + steps = 0 + evidence = input.candidate.evidence + if evidence is not None: + descriptor = evidence.get(EVIDENCE_TRACE) + if descriptor is not None and descriptor.ref: + payload = json.loads(Path(descriptor.ref).read_text(encoding="utf-8")) + steps = len(payload.get("steps") or []) + return MetricResult(outputs=[MetricOutput(name="has_trajectory", value=steps > 0)]) + + +def _task() -> AgentEvalTask: + return AgentEvalTask( + id="say-done", + intent="Agent follows a trivial instruction and exits cleanly.", + inputs={"instruction": "Reply with the single word DONE and nothing else."}, + metrics=[_TrajectoryEvidenceMetric()], + ) + + +# --- hermetic: fake nemo_fabric so CI exercises the runner+evaluator+metric+evidence chain ---------- + + +class _FakeConfig: + @classmethod + def from_mapping(cls, mapping: dict[str, Any]) -> _FakeConfig: + return cls() + + +class _FakeProfile: + def __init__(self, **kwargs: Any) -> None: + self.kwargs = kwargs + + @classmethod + def from_mapping(cls, mapping: dict[str, Any]) -> _FakeProfile: + return cls(**mapping) + + +class _FakeArtifact: + def __init__(self, name: str, kind: str, path: Path) -> None: + self.name = name + self.kind = kind + self.path = path + self.media_type = "application/json" if kind == "atif" else "text/plain" + self.metadata: dict[str, Any] = {} + + +class _FakeManifest: + def __init__(self, artifacts: list[_FakeArtifact]) -> None: + self.root: Path | None = None + self.artifacts = artifacts + + +class _FakeResult: + def __init__(self, artifacts: list[_FakeArtifact]) -> None: + self.status = "succeeded" + self.output = {"adapter": "cli", "response": "DONE"} + self.error = None + self.harness = "codex" + self.adapter_id = "nvidia.fabric.codex.cli" + self.adapter_kind = "process" + self.invocation_id = "inv-1" + self.artifacts = _FakeManifest(artifacts) + self.telemetry: list[Any] = [] + self.events: list[Any] = [] + + def to_mapping(self) -> dict[str, Any]: + return {"status": self.status, "output": self.output} + + +@pytest.mark.asyncio +async def test_fabric_runner_eval_exposes_trajectory_to_metric(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # A real (fake-backed) ATIF file the promoted artifact points at. + atif_path = tmp_path / "trajectory.atif.json" + atif_path.write_text(json.dumps({"schema_version": "atif/v1", "steps": [{"kind": "message"}]}), encoding="utf-8") + artifacts = [_FakeArtifact("relay_atif", "atif", atif_path), _FakeArtifact("stdout", "log", tmp_path / "out.txt")] + (tmp_path / "out.txt").write_text("DONE\n", encoding="utf-8") + + class _FakeClient: + async def __aenter__(self) -> _FakeClient: + return self + + async def __aexit__(self, *exc: Any) -> bool: + return False + + async def run(self, agent: Any, **kwargs: Any) -> _FakeResult: + return _FakeResult(artifacts) + + module = types.ModuleType("nemo_fabric") + module.FabricClient = _FakeClient # type: ignore[attr-defined] + module.FabricConfig = _FakeConfig # type: ignore[attr-defined] + module.FabricProfileConfig = _FakeProfile # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "nemo_fabric", module) + + # Trajectory capture builds the profile from nemo_relay's typed config objects (lazy import); stub + # the optional package so the hermetic chain resolves without the native dependency. + class _FakeRelayObj: + def __init__(self, **kwargs: Any) -> None: + self.kwargs = kwargs + + def to_dict(self) -> dict[str, Any]: + return {k: (v.to_dict() if hasattr(v, "to_dict") else v) for k, v in self.kwargs.items()} + + class _FakeComponentSpec: + def __init__(self, *, config: Any, enabled: bool = True) -> None: + self.config = config + self.enabled = enabled + + def to_dict(self) -> dict[str, Any]: + return {"kind": "observability", "enabled": self.enabled, "config": self.config.to_dict()} + + relay_mod = types.ModuleType("nemo_relay") + observability_mod = types.ModuleType("nemo_relay.observability") + observability_mod.AtifConfig = _FakeRelayObj # type: ignore[attr-defined] + observability_mod.AtofConfig = _FakeRelayObj # type: ignore[attr-defined] + observability_mod.ObservabilityConfig = _FakeRelayObj # type: ignore[attr-defined] + observability_mod.ComponentSpec = _FakeComponentSpec # type: ignore[attr-defined] + relay_mod.observability = observability_mod # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "nemo_relay", relay_mod) + monkeypatch.setitem(sys.modules, "nemo_relay.observability", observability_mod) + + runtime = fabric_runtime.FabricAgentRuntime( + config={"metadata": {"name": "a"}, "harness": {"adapter_id": "nvidia.fabric.codex.cli"}}, + work_root=tmp_path / "fabric", + ) + result = AgentEvaluator().run_sync( + tasks=[_task()], + target=runtime, + config=AgentEvalRunConfig(output_dir=tmp_path / "out", parallelism=1, write_dashboard=False), + ) + + trial = result.trials[0] + assert trial.status == "completed" + # The trajectory is exposed under the standard trace key, as an existing ATIF file. + trace = trial.evidence.descriptors[EVIDENCE_TRACE] + assert trace.format == EVIDENCE_FORMAT_ATIF + assert Path(trace.ref).exists() + # The metric received the evidence and scored from the trajectory content. + scores = [s for s in result.scores if s.metric_type == "has-trajectory"] + assert scores and scores[0].trial_id == trial.id + assert scores[0].outputs[0].name == "has_trajectory" + assert scores[0].outputs[0].value in (True, 1.0) + + +# --- gated live: real fabric -> codex -> Relay ATIF ------------------------------------------------ + +_FABRIC_REPO = os.environ.get("NEMO_FABRIC_REPO", "") +_LIVE_READY = bool( + _FABRIC_REPO + and shutil.which("codex") + and shutil.which("nemo-relay") + and importlib.util.find_spec("nemo_fabric") is not None +) +requires_live_fabric = pytest.mark.skipif( + not _LIVE_READY, + reason="needs NEMO_FABRIC_REPO + codex + nemo-relay gateway + nemo_fabric (run script/dev-install-fabric.sh)", +) + + +@requires_live_fabric +@pytest.mark.timeout(300) +def test_fabric_codex_live_eval_captures_atif_trajectory(tmp_path: Path) -> None: + codex_config = { + "schema_version": "fabric.agent/v1alpha1", + "metadata": {"name": "eval-fabric-live"}, + "harness": { + "adapter_id": "nvidia.fabric.codex.cli", + "resolution": "preinstalled", + "settings": {"sandbox": "workspace-write", "skip_git_repo_check": True, "timeout_seconds": 180}, + }, + "runtime": {"mode": "oneshot", "transport": "cli", "input_schema": "text", "output_schema": "message"}, + "environment": {"provider": "local", "workspace": str(tmp_path / "ws")}, + "telemetry": {"enabled": False}, + } + (tmp_path / "ws").mkdir(parents=True, exist_ok=True) + runtime = fabric_runtime.FabricAgentRuntime( + config=codex_config, + base_dir=Path(_FABRIC_REPO), # so the adapter registry resolves adapters/codex-cli + work_root=tmp_path / "fabric", + capture_trajectory=True, + ) + + result = AgentEvaluator().run_sync( + tasks=[_task()], + target=runtime, + config=AgentEvalRunConfig(output_dir=tmp_path / "out", parallelism=1, write_dashboard=False), + ) + + trial = result.trials[0] + assert trial.status == "completed", trial.metadata + trace = trial.evidence.descriptors[EVIDENCE_TRACE] + assert trace.format == EVIDENCE_FORMAT_ATIF + atif = Path(trace.ref) + assert atif.exists() and atif.stat().st_size > 0 + assert "steps" in json.loads(atif.read_text(encoding="utf-8")) + # The metric read the real trajectory and scored on it. + scores = [s for s in result.scores if s.metric_type == "has-trajectory"] + assert scores and scores[0].outputs[0].value in (True, 1.0) diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_runtime.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_runtime.py new file mode 100644 index 00000000000..dd53d5828d6 --- /dev/null +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_runtime.py @@ -0,0 +1,325 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for FabricAgentRuntime using a fake nemo_fabric SDK (the native package is optional).""" + +from __future__ import annotations + +import json +import sys +import types +from pathlib import Path +from typing import Any + +import pytest +from nemo_evaluator_sdk.agent_eval.runtimes.fabric import runtime as fabric_runtime +from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalTask +from nemo_evaluator_sdk.values.evidence import EVIDENCE_FORMAT_ATIF, EVIDENCE_TRACE + + +class _FakeConfig: + def __init__(self, mapping: dict[str, Any]) -> None: + self.mapping = mapping + + @classmethod + def from_mapping(cls, mapping: dict[str, Any]) -> _FakeConfig: + return cls(mapping) + + +class _FakeProfile: + def __init__(self, *, name: str | None = None, models: Any = None, mapping: Any = None) -> None: + self.name = name + self.models = models + self.mapping = mapping + + @classmethod + def from_mapping(cls, mapping: dict[str, Any]) -> _FakeProfile: + return cls(name=mapping.get("name"), mapping=mapping) + + +class _FakeRelayConfig: + """Stand-in for nemo_relay.observability's typed config objects (AtifConfig/AtofConfig/...).""" + + def __init__(self, **kwargs: Any) -> None: + self.kwargs = kwargs + + def to_dict(self) -> dict[str, Any]: + return {key: (value.to_dict() if hasattr(value, "to_dict") else value) for key, value in self.kwargs.items()} + + +class _FakeComponentSpec: + def __init__(self, *, config: Any, enabled: bool = True) -> None: + self.config = config + self.enabled = enabled + + def to_dict(self) -> dict[str, Any]: + return {"kind": "observability", "enabled": self.enabled, "config": self.config.to_dict()} + + +class _FakeArtifact: + def __init__(self, name: str, kind: str, path: Path, media_type: str | None = None) -> None: + self.name = name + self.kind = kind + self.path = path + self.media_type = media_type + + +class _FakeManifest: + def __init__(self, artifacts: list[_FakeArtifact]) -> None: + self.root: Path | None = None + self.artifacts = artifacts + + +class _FakeError: + def __init__(self, stage: str, code: str, message: str) -> None: + self.stage = stage + self.code = code + self.message = message + + +class _FakeTelemetry: + def __init__(self, *, provider: str, kind: str, uri: str | None, trace_id: str | None) -> None: + self.provider = provider + self.kind = kind + self.uri = uri + self.trace_id = trace_id + + +class _FakeEvent: + def __init__(self, kind: str, message: str) -> None: + self.kind = kind + self.message = message + + +class _FakeResult: + def __init__( + self, + *, + status: str, + output: Any = None, + error: _FakeError | None = None, + artifacts: list[_FakeArtifact] | None = None, + ) -> None: + self.status = status + self.output = output + self.error = error + self.harness = "codex" + self.adapter_id = "nvidia.fabric.codex.cli" + self.adapter_kind = "process" + self.invocation_id = "inv-1" + self.artifacts = _FakeManifest(artifacts or []) + self.telemetry = [_FakeTelemetry(provider="relay", kind="trace", uri="file:///relay", trace_id="tid-1")] + self.events = [_FakeEvent("runtime_start", "started")] + + def to_mapping(self) -> dict[str, Any]: + return {"status": self.status, "output": self.output, "harness": self.harness} + + +def _install_fake_fabric(monkeypatch: pytest.MonkeyPatch, handler: Any) -> type: + """Inject a fake ``nemo_fabric`` module (the runtime imports it lazily); return the client class.""" + + class _FakeClient: + recorded: list[dict[str, Any]] = [] + + async def __aenter__(self) -> _FakeClient: + return self + + async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> bool: + return False + + async def run(self, agent: Any, **kwargs: Any) -> Any: + _FakeClient.recorded.append({"agent": agent, **kwargs}) + return handler(agent, kwargs) + + _FakeClient.recorded = [] + module = types.ModuleType("nemo_fabric") + module.FabricClient = _FakeClient # type: ignore[attr-defined] + module.FabricConfig = _FakeConfig # type: ignore[attr-defined] + module.FabricProfileConfig = _FakeProfile # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "nemo_fabric", module) + + # The runtime builds the trajectory profile from nemo_relay's typed config objects (lazy import); + # stub the optional package so trajectory-capture paths resolve without the native dependency. + relay_mod = types.ModuleType("nemo_relay") + observability_mod = types.ModuleType("nemo_relay.observability") + observability_mod.AtifConfig = _FakeRelayConfig # type: ignore[attr-defined] + observability_mod.AtofConfig = _FakeRelayConfig # type: ignore[attr-defined] + observability_mod.ObservabilityConfig = _FakeRelayConfig # type: ignore[attr-defined] + observability_mod.ComponentSpec = _FakeComponentSpec # type: ignore[attr-defined] + relay_mod.observability = observability_mod # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "nemo_relay", relay_mod) + monkeypatch.setitem(sys.modules, "nemo_relay.observability", observability_mod) + return _FakeClient + + +_TASK = AgentEvalTask(id="task/1", intent="Answer.", inputs={"prompt": "Ping?"}) +_CONFIG = {"metadata": {"name": "a"}, "harness": {"adapter_id": "nvidia.fabric.codex.cli"}} + + +@pytest.mark.asyncio +async def test_fabric_runtime_maps_succeeded_result_to_completed_trial( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + artifact = _FakeArtifact("stdout", "log", tmp_path / "stdout.txt") + + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult( + status="succeeded", + output={"adapter": "cli", "response": "PONG", "returncode": 0}, + artifacts=[artifact], + ) + + client_cls = _install_fake_fabric(monkeypatch, handler) + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, model="openai/gpt-5.4", work_root=tmp_path / "fabric") + + trials = await runtime.run_tasks([_TASK]) + + trial = trials[0] + assert trial.status == "completed" + assert trial.output is not None + assert trial.output.output_text == "PONG" # extracted from the adapter envelope's `response` + assert trial.output.response == {"adapter": "cli", "response": "PONG", "returncode": 0} + assert trial.metadata["harness"] == "codex" + assert trial.metadata["adapter_id"] == "nvidia.fabric.codex.cli" + assert trial.metadata["generated"] is True + # Evidence: the persisted result envelope + each Fabric artifact by name. + assert trial.evidence is not None + assert trial.evidence.descriptors["result"].ref.endswith("fabric_result.json") + assert trial.evidence.descriptors["stdout"].ref == str(tmp_path / "stdout.txt") + result_file = tmp_path / "fabric" / "000000-task-1" / "fabric_result.json" + assert json.loads(result_file.read_text(encoding="utf-8"))["status"] == "succeeded" + # Model applied as a profile overlay (Harbor pattern) + the Relay ATIF trajectory overlay. + profiles = client_cls.recorded[0]["profiles"] + names = [p.name for p in profiles] + assert "eval_model" in names + assert "eval_trajectory" in names # capture_trajectory defaults on + model_profile = next(p for p in profiles if p.name == "eval_model") + assert model_profile.models == {"default": {"provider": "openai", "model": "openai/gpt-5.4"}} + assert client_cls.recorded[0]["request_id"] == "task/1" + # Telemetry reference is preserved end-to-end (uri + trace_id), not just provider/kind. + assert trial.evidence.metadata["telemetry"][0]["uri"] == "file:///relay" + assert trial.evidence.metadata["telemetry"][0]["trace_id"] == "tid-1" + + +@pytest.mark.asyncio +async def test_fabric_runtime_maps_atif_artifact_to_trace_evidence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + atif = _FakeArtifact("relay_atif", "atif", tmp_path / "trajectory.atif.json", "application/json") + + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output={"response": "ok"}, artifacts=[atif]) + + _install_fake_fabric(monkeypatch, handler) + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, work_root=tmp_path / "fabric") + + trials = await runtime.run_tasks([_TASK]) + + evidence = trials[0].evidence + assert evidence is not None + # The ATIF artifact is exposed both under its own name and the standard trace key. + assert evidence.descriptors["relay_atif"].ref == str(tmp_path / "trajectory.atif.json") + trace = evidence.descriptors[EVIDENCE_TRACE] + assert trace.format == EVIDENCE_FORMAT_ATIF + assert trace.ref == str(tmp_path / "trajectory.atif.json") + + +@pytest.mark.asyncio +async def test_fabric_runtime_capture_trajectory_false_skips_relay_overlay( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output="ok") + + client_cls = _install_fake_fabric(monkeypatch, handler) + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, work_root=tmp_path / "fabric", capture_trajectory=False) + + await runtime.run_tasks([_TASK]) + + names = [p.name for p in client_cls.recorded[0]["profiles"]] + assert "eval_trajectory" not in names + + +@pytest.mark.asyncio +async def test_fabric_runtime_maps_failed_result_to_failed_trial( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult( + status="failed", + error=_FakeError(stage="invoke", code="process_exit_nonzero", message="boom"), + ) + + _install_fake_fabric(monkeypatch, handler) + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, work_root=tmp_path / "fabric") + + trials = await runtime.run_tasks([_TASK]) + + trial = trials[0] + assert trial.status == "failed" + assert trial.output is None + assert trial.metadata["error_type"] == "process_exit_nonzero" + assert trial.metadata["error"] == "boom" + assert trial.evidence is not None + assert "error" in trial.evidence.descriptors + + +@pytest.mark.asyncio +async def test_fabric_runtime_maps_timeout_to_failed_trial(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output="never") + + _install_fake_fabric(monkeypatch, handler) + + async def fake_wait_for(awaitable: Any, timeout: float) -> Any: + awaitable.close() + raise TimeoutError + + monkeypatch.setattr(fabric_runtime.asyncio, "wait_for", fake_wait_for) + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, work_root=tmp_path / "fabric") + + trials = await runtime.run_tasks([_TASK]) + + assert trials[0].status == "failed" + assert trials[0].metadata["error_type"] == "TimeoutError" + + +@pytest.mark.asyncio +async def test_fabric_runtime_captures_run_exception_as_failed_trial( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + raise RuntimeError("client blew up") + + _install_fake_fabric(monkeypatch, handler) + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, work_root=tmp_path / "fabric") + + trials = await runtime.run_tasks([_TASK]) + + assert trials[0].status == "failed" + assert trials[0].metadata["error_type"] == "RuntimeError" + assert trials[0].metadata["error"] == "client blew up" + + +@pytest.mark.asyncio +async def test_fabric_runtime_raises_without_nemo_fabric(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + # run_tasks surfaces a clear error when the optional native package can't be imported. + monkeypatch.setitem(sys.modules, "nemo_fabric", None) # forces ImportError on `import nemo_fabric` + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, work_root=tmp_path / "fabric") + with pytest.raises(RuntimeError, match="requires the `nemo-fabric` package"): + await runtime.run_tasks([_TASK]) + + +@pytest.mark.asyncio +async def test_fabric_runtime_trajectory_capture_requires_nemo_relay( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + # Trajectory capture builds the profile from nemo_relay; surface a clear error when it is absent. + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output="ok") + + _install_fake_fabric(monkeypatch, handler) # installs fabric + relay stubs... + monkeypatch.setitem(sys.modules, "nemo_relay.observability", None) # ...then force ImportError on relay + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, work_root=tmp_path / "fabric") + with pytest.raises(RuntimeError, match="nemo-relay"): + await runtime.run_tasks([_TASK]) diff --git a/plugins/nemo-evaluator/openapi/openapi.yaml b/plugins/nemo-evaluator/openapi/openapi.yaml index e957f3fdddf..174c5117eb4 100644 --- a/plugins/nemo-evaluator/openapi/openapi.yaml +++ b/plugins/nemo-evaluator/openapi/openapi.yaml @@ -1451,6 +1451,7 @@ components: - $ref: '#/components/schemas/ModelTargetInput' - $ref: '#/components/schemas/AgentTargetInput' - $ref: '#/components/schemas/CodexRunnerTarget' + - $ref: '#/components/schemas/FabricRunnerTarget' title: Target description: 'What generates trials online: a Model or Agent endpoint, or an agent runner (e.g. Codex CLI). Endpoint targets carry their own request @@ -1602,6 +1603,7 @@ components: - $ref: '#/components/schemas/ModelTargetOutput' - $ref: '#/components/schemas/AgentTargetOutput' - $ref: '#/components/schemas/CodexRunnerTarget' + - $ref: '#/components/schemas/FabricRunnerTarget' title: Target description: 'What generates trials online: a Model or Agent endpoint, or an agent runner (e.g. Codex CLI). Endpoint targets carry their own request @@ -2751,6 +2753,61 @@ components: - kind title: EvidenceDescriptor description: Descriptor for a candidate trace, source, or artifact. + FabricRunnerTarget: + properties: + kind: + type: string + const: fabric + title: Kind + default: fabric + config: + additionalProperties: true + type: object + title: Config + description: Inline NeMo Fabric agent config (an ``agent.yaml`` as a JSON-shaped + mapping). Its ``harness.adapter_id`` selects the harness, e.g. ``nvidia.fabric.codex.cli`` + for Codex. + profiles: + items: + additionalProperties: true + type: object + type: array + title: Profiles + description: Ordered Fabric profile overlays applied after the base config, + before ``model``. + model: + title: Model + description: Optional ``provider/model`` slug applied as a final profile + overlay; the harness default is used when omitted. + type: string + timeout_s: + type: integer + minimum: 1.0 + title: Timeout S + description: Per-task timeout for the Fabric run, in seconds. + default: 600 + capture_trajectory: + type: boolean + title: Capture Trajectory + description: Capture the agent trajectory as ATIF via NeMo Relay and attach + it to trial evidence. Requires the NeMo Relay gateway in the run environment. + default: true + additionalProperties: false + type: object + required: + - config + title: FabricRunnerTarget + description: 'Generate trials by driving an agent harness through the NeMo Fabric + runtime. + + + Fabric is harness-agnostic: the harness (Codex, Hermes, ...) is selected by + the supplied + + config''s ``harness.adapter_id`` and is never inferred from ``model``. ``model`` + is applied as a + + final profile overlay when given.' FieldMapping: properties: input: diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py index b2c715234db..6ea9e7fd444 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_evaluate.py @@ -31,6 +31,7 @@ AgentEvalTaskSpec, AgentTarget, CodexRunnerTarget, + FabricRunnerTarget, ModelTarget, Target, ) @@ -41,6 +42,7 @@ from nemo_evaluator_sdk.agent_eval.persistence import persist_run from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult from nemo_evaluator_sdk.agent_eval.runtimes.codex.runtime import CodexCliAgentRuntime +from nemo_evaluator_sdk.agent_eval.runtimes.fabric.runtime import FabricAgentRuntime from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTarget from nemo_evaluator_sdk.metrics.protocol import Metric @@ -248,6 +250,16 @@ def _resolve_target( work_root=ctx.storage.persistent / "codex", ) return runtime, None, None + if isinstance(target, FabricRunnerTarget): + fabric_runtime = FabricAgentRuntime( + config=target.config, + profiles=target.profiles, + model=target.model, + timeout_s=target.timeout_s, + capture_trajectory=target.capture_trajectory, + work_root=ctx.storage.persistent / "fabric", + ) + return fabric_runtime, None, None return None, None, None @staticmethod diff --git a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py index 034860a712a..b2a5bdfa886 100644 --- a/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py +++ b/plugins/nemo-evaluator/src/nemo_evaluator/jobs/agent_spec.py @@ -76,9 +76,41 @@ class CodexRunnerTarget(BaseModel): timeout_s: int = Field(default=600, ge=1, description="Per-task timeout for the Codex CLI, in seconds.") +class FabricRunnerTarget(BaseModel): + """Generate trials by driving an agent harness through the NeMo Fabric runtime. + + Fabric is harness-agnostic: the harness (Codex, Hermes, ...) is selected by the supplied + config's ``harness.adapter_id`` and is never inferred from ``model``. ``model`` is applied as a + final profile overlay when given. + """ + + model_config = ConfigDict(extra="forbid") + + kind: Literal["fabric"] = "fabric" + config: dict[str, Any] = Field( + description="Inline NeMo Fabric agent config (an ``agent.yaml`` as a JSON-shaped mapping). Its " + "``harness.adapter_id`` selects the harness, e.g. ``nvidia.fabric.codex.cli`` for Codex.", + ) + profiles: list[dict[str, Any]] = Field( + default_factory=list, + description="Ordered Fabric profile overlays applied after the base config, before ``model``.", + ) + model: str | None = Field( + default=None, + description="Optional ``provider/model`` slug applied as a final profile overlay; the harness " + "default is used when omitted.", + ) + timeout_s: int = Field(default=600, ge=1, description="Per-task timeout for the Fabric run, in seconds.") + capture_trajectory: bool = Field( + default=True, + description="Capture the agent trajectory as ATIF via NeMo Relay and attach it to trial evidence. " + "Requires the NeMo Relay gateway in the run environment.", + ) + + #: The agent-runner slot of the target union — the spec-side mirror of ``AgentTaskRunner``, resolved -#: to a runtime at run time. One member today; widen to a ``kind``-union as more runners land. -AgentRunnerTarget: TypeAlias = CodexRunnerTarget +#: to a runtime at run time. ``kind``-discriminated; widen with more members as runners land. +AgentRunnerTarget: TypeAlias = CodexRunnerTarget | FabricRunnerTarget #: What generates trials: a Model or Agent endpoint, or an agent runner. ``kind``-discriminated, and #: the spec-level analog of the SDK's runtime ``AgentEvalTarget`` (Model | Agent | AgentTaskRunner). diff --git a/plugins/nemo-evaluator/tests/test_agent_evaluate.py b/plugins/nemo-evaluator/tests/test_agent_evaluate.py index a9d1a096041..fa5dfc503b5 100644 --- a/plugins/nemo-evaluator/tests/test_agent_evaluate.py +++ b/plugins/nemo-evaluator/tests/test_agent_evaluate.py @@ -25,6 +25,7 @@ AgentEvalTaskSpec, AgentTarget, CodexRunnerTarget, + FabricRunnerTarget, ModelTarget, Target, ) @@ -35,6 +36,7 @@ from nemo_evaluator.tasks.runner import SDK_INITIALIZATION_EXIT_CODE from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult, AgentEvalSummary from nemo_evaluator_sdk.agent_eval.runtimes.codex.runtime import CodexCliAgentRuntime +from nemo_evaluator_sdk.agent_eval.runtimes.fabric.runtime import FabricAgentRuntime from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask from nemo_evaluator_sdk.agent_eval.trials import ( AgentEvalTarget, @@ -192,6 +194,21 @@ def test_resolve_target_builds_codex_runtime_from_runner_target(tmp_path: Path) assert params is None +def test_resolve_target_builds_fabric_runtime_from_runner_target(tmp_path: Path) -> None: + ctx = _job_context(tmp_path) + fabric_target = FabricRunnerTarget( + config={"metadata": {"name": "a"}, "harness": {"adapter_id": "nvidia.fabric.codex.cli"}}, + model="openai/gpt-5.4", + ) + target, prompt_template, params = AgentEvalJob._resolve_target(fabric_target, ctx) + assert isinstance(target, FabricAgentRuntime) + assert target._model == "openai/gpt-5.4" + assert target._work_root == ctx.storage.persistent / "fabric" + # A runner shapes its own request, so it contributes no prompt template or inference params. + assert prompt_template is None + assert params is None + + def test_resolve_target_unpacks_model_target_request_config(tmp_path: Path) -> None: ctx = _job_context(tmp_path) model = Model(url="http://model.test/v1/chat/completions", name="m") @@ -342,6 +359,13 @@ def _assert_agent_eval_step_entrypoint(job_spec: PlatformJobSpec) -> None: ("target", "expected_kind", "expected_endpoint_name"), [ (CodexRunnerTarget(model="gpt-5.5"), "codex", None), + ( + FabricRunnerTarget( + config={"metadata": {"name": "a"}, "harness": {"adapter_id": "nvidia.fabric.codex.cli"}} + ), + "fabric", + None, + ), ( ModelTarget( model=Model(url="http://model.test/v1/chat/completions", name="test-model"), diff --git a/script/dev-install-fabric.sh b/script/dev-install-fabric.sh new file mode 100755 index 00000000000..cb205098720 --- /dev/null +++ b/script/dev-install-fabric.sh @@ -0,0 +1,83 @@ +#!/usr/bin/env bash +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Dev-only: set up the local dependencies the Fabric eval runner needs so the type checker and a +# live FabricAgentRuntime run work end-to-end: +# 1. the native `nemo-fabric` SDK (with the codex + relay extras) into the project venv, and +# 2. the `nemo-relay` gateway binary (required for ATIF trajectory capture on the codex harness). +# This is an imperative install — it does NOT touch uv.lock, and CI intentionally runs without it +# (the `# ty: ignore[unresolved-import]` in agent_eval/runtimes/fabric/runtime.py covers the CI case). +# +# nemo-fabric and the nemo-relay gateway are private/native builds with no published wheel/binary in +# our index, so they can't be locked dependencies yet (see +# plugins/nemo-evaluator/docs/design/fabric-runner-integration.md, Tier 3). A live codex run also +# needs the `codex` CLI + `codex login` auth. +# +# Usage: +# script/dev-install-fabric.sh # NeMo-Fabric+NeMo-Relay under $HOME/workspace +# NEMO_FABRIC_REPO=... NEMO_RELAY_REPO=... script/dev-install-fabric.sh +# script/dev-install-fabric.sh --uninstall # restore the CI-equivalent (no nemo-fabric) state +set -euo pipefail + +VENV_PY=".venv/bin/python" +if [ ! -x "$VENV_PY" ]; then + echo "Project venv not found at $VENV_PY. Run 'make bootstrap-python' (see SETUP.md) first." >&2 + exit 1 +fi + +if [ "${1:-}" = "--uninstall" ]; then + uv pip uninstall --python "$VENV_PY" nemo-fabric + echo "Removed nemo-fabric; venv is back to the lock-consistent / CI-equivalent state." + echo "(The nemo-relay gateway binary, if installed, is left in place — remove it from ~/.cargo/bin manually if desired.)" + exit 0 +fi + +# nemo-fabric builds a Rust/pyo3 extension via maturin and the relay gateway is a Rust CLI, so cargo +# must be on PATH. +if ! command -v cargo >/dev/null 2>&1 && [ -f "$HOME/.cargo/env" ]; then + # shellcheck disable=SC1091 + . "$HOME/.cargo/env" +fi +if ! command -v cargo >/dev/null 2>&1; then + echo "cargo (Rust toolchain) not found; install it to build the native components: https://rustup.rs" >&2 + exit 1 +fi + +# 1. nemo-fabric SDK (+ codex and relay extras) into the project venv. +FABRIC_REPO="${NEMO_FABRIC_REPO:-$HOME/workspace/NeMo-Fabric}" +if [ ! -d "$FABRIC_REPO" ]; then + echo "NeMo-Fabric checkout not found at: $FABRIC_REPO" >&2 + echo "Clone it (gh repo clone NVIDIA/NeMo-Fabric) or set NEMO_FABRIC_REPO=/path/to/NeMo-Fabric." >&2 + exit 1 +fi +echo "Building + installing nemo-fabric[codex,relay] from $FABRIC_REPO into $VENV_PY ..." +uv pip install --python "$VENV_PY" "${FABRIC_REPO}[codex,relay]" +"$VENV_PY" -c "import nemo_fabric; from nemo_fabric import FabricClient, RunResult; print('nemo_fabric OK:', nemo_fabric.__file__)" + +# 2. nemo-relay gateway binary (codex -> OTLP -> gateway -> trajectory-*.atif.json). Required for +# trajectory capture; the pip `nemo-relay` package does NOT ship this executable. +if command -v nemo-relay >/dev/null 2>&1; then + echo "nemo-relay gateway already on PATH: $(command -v nemo-relay) ($(nemo-relay --version 2>/dev/null || echo '?'))" +else + RELAY_REPO="${NEMO_RELAY_REPO:-$HOME/workspace/NeMo-Relay}" + if [ ! -d "$RELAY_REPO" ]; then + echo "nemo-relay gateway not on PATH and NeMo-Relay checkout not found at: $RELAY_REPO" >&2 + echo "Clone it (gh repo clone NVIDIA/NeMo-Relay) or set NEMO_RELAY_REPO=/path/to/NeMo-Relay." >&2 + echo "Trajectory (ATIF) capture will fail without the nemo-relay gateway." >&2 + exit 1 + fi + echo "Building + installing the nemo-relay gateway from $RELAY_REPO (cargo install) ..." + cargo install --path "$RELAY_REPO/crates/cli" --locked + echo "nemo-relay gateway installed: $(command -v nemo-relay) ($(nemo-relay --version 2>/dev/null || echo '?'))" +fi + +cat <<'EOF' + +Done. Fabric's real types now resolve (ty enforces them; you'll see 2 harmless "unused ty: ignore" +warnings while nemo-fabric is installed), and live FabricAgentRuntime runs can capture ATIF +trajectories (needs the `codex` CLI + `codex login` for a codex-harness run). + +Restore the CI-equivalent state with: + script/dev-install-fabric.sh --uninstall +EOF diff --git a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/runtime.py b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/runtime.py new file mode 100644 index 00000000000..8aad5d1e4af --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/runtime.py @@ -0,0 +1,366 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""NeMo Fabric-backed agent-eval runtime. + +``FabricAgentRuntime`` drives an agent harness (Codex, Hermes, ...) through the +NeMo Fabric Python SDK and adapts each normalized Fabric ``RunResult`` into an +:class:`AgentEvalTrial`. The harness is chosen by the supplied Fabric config's +``harness.adapter_id`` (never inferred from a model); an optional ``model`` slug +is applied as a final profile overlay, mirroring Fabric's own Harbor integration. + +``nemo_fabric`` is an optional native dependency: its types are imported for +annotations under ``TYPE_CHECKING`` and the package is loaded lazily at runtime, +so this module stays importable without it. +""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from nemo_platform.beta.evaluator.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask +from nemo_platform.beta.evaluator.agent_eval.trials import AgentEvalTrial, AgentEvalTrialStatus, AgentOutput +from nemo_platform.beta.evaluator.values.evidence import ( + EVIDENCE_FORMAT_ATIF, + EVIDENCE_TRACE, + CandidateEvidence, + EvidenceDescriptor, +) + +if TYPE_CHECKING: + # Annotations use nemo_fabric's real types (single source of truth). nemo_fabric is an optional + # native package not yet in our locked dependency set, so it is imported for typing only and + # loaded lazily at runtime (see ``run_tasks``). Drop the ty:ignore once nemo-fabric is a + # resolvable dependency and the checker can see it. + from nemo_fabric import ( # ty: ignore[unresolved-import] + FabricClient, + FabricConfig, + FabricProfileConfig, + RunResult, + ) + +DEFAULT_FABRIC_TIMEOUT_S = 600 +_RUNTIME_NAME = "fabric" +_MISSING_FABRIC_MSG = "FabricAgentRuntime requires the `nemo-fabric` package (native NeMo Fabric SDK)." +_MISSING_RELAY_MSG = ( + "FabricAgentRuntime trajectory capture requires the `nemo-relay` package " + "(install `nemo-fabric[relay]`), or set capture_trajectory=False." +) + +# Evidence-dir layout for trajectory capture. These subdir names are our own local layout — we create +# them and hand them to Fabric/Relay, so they are not derived from either library. +_RELAY_SUBDIR = "relay" +_ARTIFACTS_SUBDIR = "artifacts" +# Trajectory profile identity + the file-exporter output names we choose (Relay accepts these as inputs). +_TRAJECTORY_PROFILE_NAME = "eval_trajectory" +_ATIF_FILENAME_TEMPLATE = "trajectory-{session_id}.atif.json" +_ATOF_FILENAME = "events.atof.jsonl" +# Fabric telemetry-profile selectors (file exporter, no OTLP endpoint). +_TELEMETRY_PROVIDER = "relay" +_TELEMETRY_MODE = "sdk" +# ``kind`` Fabric stamps on the promoted Relay ATIF artifact; used to surface it as trace evidence. +_ATIF_ARTIFACT_KIND = "atif" + + +class FabricAgentRuntime: + """AgentTaskRunner that generates trials by running tasks through NeMo Fabric.""" + + def __init__( + self, + *, + config: Mapping[str, Any], + profiles: Sequence[Mapping[str, Any]] | None = None, + model: str | None = None, + base_dir: str | Path | None = None, + work_root: str | Path | None = None, + timeout_s: int = DEFAULT_FABRIC_TIMEOUT_S, + capture_trajectory: bool = True, + runtime_name: str = _RUNTIME_NAME, + ) -> None: + self._config = config + self._profiles = list(profiles or []) + self._model = model + self._base_dir = Path(base_dir).expanduser() if base_dir is not None else None + self._work_root = Path(work_root).expanduser() if work_root is not None else None + self._timeout_s = timeout_s + self._capture_trajectory = capture_trajectory + self._runtime_name = runtime_name + + async def run_tasks( + self, + tasks: Sequence[AgentEvalTask], + config: AgentEvalRunConfig | None = None, + ) -> Sequence[AgentEvalTrial]: + try: + # nemo_fabric ships a native (pyo3) core and is an optional dependency, so it is imported + # lazily here rather than at module load. + from nemo_fabric import FabricClient, FabricConfig, FabricProfileConfig # ty: ignore[unresolved-import] + except ImportError as exc: + raise RuntimeError(_MISSING_FABRIC_MSG) from exc + + resolved_config = config or AgentEvalRunConfig() + agent_config = FabricConfig.from_mapping(self._config) + base_profiles = self._build_profiles(FabricProfileConfig) + semaphore = asyncio.Semaphore(resolved_config.parallelism) + + async with FabricClient() as client: + + async def run_one(index: int, task: AgentEvalTask) -> AgentEvalTrial: + async with semaphore: + return await self._run_task( + client, agent_config, base_profiles, FabricProfileConfig, index, task, resolved_config + ) + + return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) + + async def _run_task( + self, + client: FabricClient, + agent_config: FabricConfig, + base_profiles: list[FabricProfileConfig], + profile_cls: type[FabricProfileConfig], + index: int, + task: AgentEvalTask, + config: AgentEvalRunConfig, + ) -> AgentEvalTrial: + evidence_dir = self._evidence_dir(index, task, config) + evidence_dir.mkdir(parents=True, exist_ok=True) + + profiles = list(base_profiles) + if self._capture_trajectory: + # Enable Relay's ATIF file exporter, writing the trajectory under this task's durable + # evidence dir; Fabric promotes the resulting file into RunResult.artifacts. Both the + # Fabric artifact root and the relay output dir must exist and be durable. + relay_dir = evidence_dir / _RELAY_SUBDIR + artifacts_dir = evidence_dir / _ARTIFACTS_SUBDIR + relay_dir.mkdir(parents=True, exist_ok=True) + artifacts_dir.mkdir(parents=True, exist_ok=True) + profiles.append(self._trajectory_profile(profile_cls, relay_dir=relay_dir, artifacts_dir=artifacts_dir)) + + try: + result = await asyncio.wait_for( + client.run( + agent_config, + profiles=profiles, + input=_fabric_input(task), + request_id=task.id, + base_dir=self._base_dir, + ), + timeout=self._timeout_s, + ) + except TimeoutError as exc: + return self._failed_trial(task, evidence_dir, exc) + except Exception as exc: # noqa: BLE001 - a task failure must not abort the whole run + return self._failed_trial(task, evidence_dir, exc) + + return self._to_trial(task, result, evidence_dir) + + def _to_trial(self, task: AgentEvalTask, result: RunResult, evidence_dir: Path) -> AgentEvalTrial: + # Persist the full normalized Fabric result so graders (and debugging) can see the raw + # envelope, and expose it as an evidence descriptor. + result_path = evidence_dir / "fabric_result.json" + result_path.write_text(json.dumps(result.to_mapping(), indent=2, default=str), encoding="utf-8") + + base_metadata: dict[str, Any] = { + "runtime": self._runtime_name, + "harness": result.harness, + "adapter_id": result.adapter_id, + "adapter_kind": result.adapter_kind, + "invocation_id": result.invocation_id, + "agent_model": self._model, + } + + if result.status != "succeeded": + return self._failed_trial(task, evidence_dir, _result_error(result), extra_metadata=base_metadata) + + return AgentEvalTrial( + id=f"{task.id}:fabric", + task_id=task.id, + status=AgentEvalTrialStatus.COMPLETED, + output=AgentOutput( + output_text=_extract_output_text(result.output), + response=result.output, + metadata={**base_metadata, "evidence_dir": str(evidence_dir)}, + ), + evidence=self._evidence(result, result_path), + metadata={**base_metadata, "generated": True}, + ) + + def _evidence(self, result: RunResult, result_path: Path) -> CandidateEvidence: + descriptors: dict[str, EvidenceDescriptor] = { + "result": EvidenceDescriptor(kind="json", format="json", ref=str(result_path)), + } + for artifact in result.artifacts.artifacts: + descriptors[artifact.name] = EvidenceDescriptor( + kind=artifact.kind or "file", + ref=str(artifact.path), + metadata={"media_type": artifact.media_type}, + ) + # Surface the Relay ATIF trajectory under the standard trace evidence key so graders + # that consume a normalized trajectory find it. + if artifact.kind == _ATIF_ARTIFACT_KIND: + descriptors[EVIDENCE_TRACE] = EvidenceDescriptor( + kind=EVIDENCE_TRACE, + format=EVIDENCE_FORMAT_ATIF, + ref=str(artifact.path), + ) + return CandidateEvidence( + descriptors=descriptors, + metadata={ + "runtime": self._runtime_name, + "harness": result.harness, + "telemetry": [ + {"provider": ref.provider, "kind": ref.kind, "uri": ref.uri, "trace_id": ref.trace_id} + for ref in result.telemetry + ], + "events": [{"kind": event.kind, "message": event.message} for event in result.events], + }, + ) + + def _failed_trial( + self, + task: AgentEvalTask, + evidence_dir: Path, + error: Exception | Mapping[str, Any], + *, + extra_metadata: Mapping[str, Any] | None = None, + ) -> AgentEvalTrial: + if isinstance(error, Mapping): + error_type = str(error.get("code") or error.get("stage") or "FabricError") + error_message = str(error.get("message") or error) + else: + error_type = error.__class__.__name__ + error_message = str(error) + error_path = evidence_dir / "error.json" + error_path.write_text(json.dumps({"error_type": error_type, "error": error_message}) + "\n", encoding="utf-8") + return AgentEvalTrial( + id=f"{task.id}:fabric", + task_id=task.id, + status=AgentEvalTrialStatus.FAILED, + output=None, + evidence=CandidateEvidence( + descriptors={"error": EvidenceDescriptor(kind="error", format="json", ref=str(error_path))}, + metadata={"runtime": self._runtime_name}, + ), + metadata={ + **(dict(extra_metadata) if extra_metadata else {}), + "runtime": self._runtime_name, + "error_type": error_type, + "error": error_message, + }, + ) + + def _build_profiles(self, profile_cls: type[FabricProfileConfig]) -> list[FabricProfileConfig]: + profiles = [profile_cls.from_mapping(profile) for profile in self._profiles] + if self._model: + # Apply the model as a final profile overlay (mirrors nemo_fabric.integrations.harbor). + provider = self._model.split("/", maxsplit=1)[0] if "/" in self._model else "openai" + profiles.append( + profile_cls( + name="eval_model", + models={"default": {"provider": provider, "model": self._model}}, + ) + ) + return profiles + + def _trajectory_profile( + self, profile_cls: type[FabricProfileConfig], *, relay_dir: Path, artifacts_dir: Path + ) -> FabricProfileConfig: + # Relay ATIF/ATOF file exporter (mode=sdk): the harness emits its trajectory to a local + # nemo-relay gateway, which writes ``trajectory-*.atif.json`` under ``relay_dir``. No OTLP + # collector endpoint is involved. Requires the ``nemo-relay`` gateway on PATH in the runtime. + # The Fabric artifact root is pinned to a durable dir so the promoted trajectory persists. + # + # The observability component is built from nemo_relay's own typed config objects so Relay owns + # its schema (no hand-maintained dict that silently drifts when Relay changes it); imported + # lazily since nemo-relay, like nemo-fabric, is an optional native dependency. ``schema_version`` + # is omitted — ``FabricProfileConfig`` defaults it. + try: + from nemo_relay.observability import ( # ty: ignore[unresolved-import] + AtifConfig, + AtofConfig, + ComponentSpec, + ObservabilityConfig, + ) + except ImportError as exc: + raise RuntimeError(_MISSING_RELAY_MSG) from exc + + relay_dir_str = str(relay_dir) + artifacts_dir_str = str(artifacts_dir) + observability = ComponentSpec( + config=ObservabilityConfig( + atif=AtifConfig( + enabled=True, + output_directory=relay_dir_str, + filename_template=_ATIF_FILENAME_TEMPLATE, + agent_name=self._runtime_name, + agent_version="fabric", + ), + atof=AtofConfig( + enabled=True, + output_directory=relay_dir_str, + filename=_ATOF_FILENAME, + mode="overwrite", + ), + ) + ) + return profile_cls.from_mapping( + { + "name": _TRAJECTORY_PROFILE_NAME, + "description": "Capture the agent trajectory as ATIF via the NeMo Relay file exporter.", + "runtime": {"artifacts": artifacts_dir_str}, + "environment": {"artifacts": artifacts_dir_str}, + "telemetry": { + "enabled": True, + "provider": _TELEMETRY_PROVIDER, + "mode": _TELEMETRY_MODE, + "output_dir": relay_dir_str, + "config": {"version": 1, "components": [observability.to_dict()]}, + }, + } + ) + + def _evidence_dir(self, index: int, task: AgentEvalTask, config: AgentEvalRunConfig) -> Path: + root = self._work_root + if root is None: + root = (config.output_dir or Path.cwd()) / "evidence" / "fabric" + safe_task_id = _safe_path_name(task.id) + task_dir = f"{index:06d}-{safe_task_id}" if safe_task_id else f"task-{index:06d}" + return Path(root) / task_dir + + +def _fabric_input(task: AgentEvalTask) -> str: + return f"Task id: {task.id}\nIntent: {task.intent}\nInputs: {task.inputs}\n" + + +def _extract_output_text(output: object) -> str | None: + """Pull the user-visible message out of a Fabric ``RunResult.output`` (JSON-shaped). + + Harness outputs vary; adapters commonly nest the final message under ``response`` (the codex-cli + adapter does). Prefer a string ``response``/``output_text``, else stringify the whole value. + """ + if output is None: + return None + if isinstance(output, str): + return output + if isinstance(output, Mapping): + for key in ("response", "output_text", "text", "message"): + value = output.get(key) + if isinstance(value, str): + return value + return json.dumps(output, default=str) + + +def _result_error(result: RunResult) -> Mapping[str, Any]: + error = result.error + if error is None: + return {"code": result.status, "message": "Fabric run did not succeed"} + return {"stage": error.stage, "code": error.code, "message": error.message} + + +def _safe_path_name(value: str) -> str: + return "".join(char if char.isalnum() or char in "._-" else "-" for char in value).strip(".-")[:120]