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]