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 index cc41c5d99d..c300e215c6 100644 --- 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 @@ -7,7 +7,11 @@ 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. +is applied as the config's default model, mirroring Fabric's own Harbor integration. + +Per-task settings (workspace, model, trajectory capture) are composed directly onto +a copy of the supplied config via the SDK's config helpers (``model_copy`` + +``enable_relay`` + ``environment``), rather than layered as profile overlays. Every task runs in its own fresh workspace: the runtime seeds it from ``inputs['files']`` (a no-op when there are none), runs the harness in it (via @@ -38,6 +42,7 @@ CandidateEvidence, EvidenceDescriptor, ) +from pydantic import JsonValue if TYPE_CHECKING: # Annotations use nemo_fabric's real types (single source of truth). nemo_fabric is an optional @@ -45,9 +50,10 @@ # 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, + Fabric, FabricConfig, FabricProfileConfig, + RunOutput, RunResult, ) @@ -66,18 +72,18 @@ # Per-task workspace: where seed files are staged and where the harness reads/writes. We # create it, point Fabric's ``environment.workspace`` at it, and expose it as ``workspace`` evidence. _WORKSPACE_SUBDIR = "workspace" -_WORKSPACE_PROFILE_NAME = "eval_workspace" # Evidence key + descriptor kind for the staged workspace, consumed by the # workspace-reading metrics. _WORKSPACE_EVIDENCE_KEY = "workspace" _WORKSPACE_EVIDENCE_KIND = "filesystem" -# Trajectory profile identity + the file-exporter output names we choose (Relay accepts these as inputs). -_TRAJECTORY_PROFILE_NAME = "eval_trajectory" +# File-exporter output names we choose for the Relay ATIF/ATOF trajectory (Relay accepts these as inputs). _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" +# Names for the trailing overlays that re-assert the evaluator-owned per-task settings (see +# ``_eval_lock_profiles``): Fabric applies caller profiles over the config, so these must trail them. +_WORKSPACE_PROFILE_NAME = "eval_workspace" +_MODEL_PROFILE_NAME = "eval_model" +_ARTIFACTS_PROFILE_NAME = "eval_artifacts" # ``kind`` Fabric stamps on the promoted Relay ATIF artifact; used to surface it as trace evidence. _ATIF_ARTIFACT_KIND = "atif" @@ -121,69 +127,75 @@ async def run_tasks( 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] + from nemo_fabric import Fabric, 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) + # Fail fast (once) if trajectory capture is requested but the nemo-relay gateway isn't + # importable, rather than failing every task the same way inside the per-task guard. + if self._capture_trajectory: + try: + import nemo_relay.observability # noqa: F401 # ty: ignore[unresolved-import] + except ImportError as exc: + raise RuntimeError(_MISSING_RELAY_MSG) from exc + # Caller-supplied profile overlays pass through as-is; this runtime's per-task workspace, model, + # and trajectory settings are composed directly onto a copy of the config (config-first), not + # layered as profiles. + base_profiles = [FabricProfileConfig.from_mapping(profile) for profile in self._profiles] semaphore = asyncio.Semaphore(resolved_config.parallelism) - async with FabricClient() as client: + # ``Fabric`` (formerly ``FabricClient``) is a lightweight, reusable facade — not a lifecycle + # context manager — so it is created once and reused across tasks with no cleanup. + client = Fabric() - 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 - ) + async def run_one(index: int, task: AgentEvalTask) -> AgentEvalTrial: + async with semaphore: + return await self._run_task(client, agent_config, base_profiles, index, task, resolved_config) - return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) + return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) async def _run_task( self, - client: FabricClient, + client: Fabric, agent_config: FabricConfig, base_profiles: list[FabricProfileConfig], - profile_cls: type[FabricProfileConfig], index: int, task: AgentEvalTask, config: AgentEvalRunConfig, ) -> AgentEvalTrial: + # nemo_fabric is already imported+validated in ``run_tasks``; this is a cached sys.modules + # lookup, not a re-load, so the type is used where it's constructed instead of threaded down. + from nemo_fabric import FabricProfileConfig, RunRequest # ty: ignore[unresolved-import] + 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)) - # Every task runs in its own fresh workspace: seed any ``inputs['files']`` into it (a no-op when - # there are none), point the harness at it via ``environment.workspace``, and expose it as - # ``workspace`` filesystem evidence — a uniform per-task dir that maps cleanly onto a per-task - # container volume later. Seeding runs inside the guarded block so a bad seed (a path escaping - # the workspace, an unresolvable fileset) fails just this task, not the whole run; it is - # synchronous and may block (a fileset handler downloads), so it is offloaded off the shared - # event loop. + # there are none), point the harness at it, and expose it as ``workspace`` filesystem evidence — + # a uniform per-task dir that maps cleanly onto a per-task container volume later. Seeding runs + # inside the guarded block so a bad seed (a path escaping the workspace, an unresolvable fileset) + # fails just this task, not the whole run; it is synchronous and may block (a fileset handler + # downloads), so it is offloaded off the shared event loop. workspace_dir = evidence_dir / _WORKSPACE_SUBDIR workspace_dir.mkdir(parents=True, exist_ok=True) try: seeded_files = await asyncio.to_thread(seed_workspace, workspace_dir, task.inputs.get(SEED_FILES_INPUT_KEY)) - profiles.append(self._workspace_profile(profile_cls, workspace_dir=workspace_dir)) + task_config = self._compose_config(agent_config, evidence_dir, workspace_dir) + # Caller ``base_profiles`` are applied by Fabric over the config; the evaluator-owned + # settings are re-asserted as trailing overlays so they win over any caller profile. + lock_profiles = self._eval_lock_profiles( + FabricProfileConfig, workspace_dir=workspace_dir, evidence_dir=evidence_dir + ) result = await asyncio.wait_for( + # ``Fabric.run`` folds the per-invocation input + request id into a ``RunRequest``. client.run( - agent_config, - profiles=profiles, - input=_fabric_input(task, seeded_files), - request_id=task.id, + task_config, + profiles=[*base_profiles, *lock_profiles], base_dir=self._base_dir, + request=RunRequest(input=_fabric_input(task, seeded_files), request_id=task.id), ), timeout=self._timeout_s, ) @@ -214,13 +226,17 @@ def _to_trial( if result.status != "succeeded": return self._failed_trial(task, evidence_dir, _result_error(result), extra_metadata=base_metadata) + # Fabric wraps the output in a ``RunOutput`` mapping (RunOutput response contract, #52), + # which is not itself a JSON value; normalize it to a plain mapping so it round-trips through the + # trial's ``JsonValue``-typed response. + output = _normalize_output(result.output) 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, + output_text=_extract_output_text(output), + response=output, metadata={**base_metadata, "evidence_dir": str(evidence_dir)}, ), evidence=self._evidence(result, result_path, workspace_dir), @@ -297,31 +313,90 @@ def _failed_trial( }, ) - def _build_profiles(self, profile_cls: type[FabricProfileConfig]) -> list[FabricProfileConfig]: - profiles = [profile_cls.from_mapping(profile) for profile in self._profiles] + def _compose_config( + self, + agent_config: FabricConfig, + evidence_dir: Path, + workspace_dir: Path, + ) -> FabricConfig: + # nemo_fabric is already imported+validated in ``run_tasks``; this is a cached sys.modules + # lookup, not a re-load, so the type is used where it's constructed instead of threaded down. + from nemo_fabric import EnvironmentConfig # ty: ignore[unresolved-import] + + # Config-first composition (the SDK's recommended in-memory pattern): copy the base config and + # apply this task's workspace, model, and trajectory settings directly onto it, rather than + # layering FabricProfileConfig overlays. + cfg = agent_config.model_copy(deep=True) + + # Point the harness at this task's staged workspace (the codex-cli adapter resolves its cwd from + # it). ``provider="local"`` is required by the native planner. Any config-supplied + # environment.workspace is overridden per task. + environment = cfg.environment or EnvironmentConfig(provider="local") + environment.provider = environment.provider or "local" + environment.workspace = str(workspace_dir) + cfg.environment = environment + + # Apply the model as the config's default (mirrors nemo_fabric.integrations.harbor). 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}}, + cfg.models["default"] = {"provider": provider, "model": self._model} + + if self._capture_trajectory: + # Enable Relay's ATIF/ATOF file exporter under this task's durable evidence dir, and pin the + # Fabric artifact root so the promoted ``trajectory-*.atif.json`` persists. Requires the + # ``nemo-relay`` gateway on PATH in the runtime. + 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) + cfg.enable_relay(output_dir=str(relay_dir), config=self._relay_config(relay_dir)) + cfg.runtime.artifacts = str(artifacts_dir) + cfg.environment.artifacts = str(artifacts_dir) + + return cfg + + def _eval_lock_profiles( + self, + profile_cls: type[FabricProfileConfig], + *, + workspace_dir: Path, + evidence_dir: Path, + ) -> list[FabricProfileConfig]: + # ``_compose_config`` composes the evaluator's per-task settings onto the config, but Fabric + # applies caller-supplied profiles OVER the config (last-wins), so a caller profile could + # otherwise override them. Re-assert the evaluator-owned settings here as trailing overlays — + # applied after the caller profiles — so the per-task workspace (isolation + ``workspace`` + # evidence integrity), the model under evaluation, and the trajectory artifact location stay + # authoritative and non-overridable. + overlays = [ + profile_cls.from_mapping( + {"name": _WORKSPACE_PROFILE_NAME, "environment": {"workspace": str(workspace_dir)}} + ) + ] + if self._model: + provider = self._model.split("/", maxsplit=1)[0] if "/" in self._model else "openai" + overlays.append( + profile_cls.from_mapping( + {"name": _MODEL_PROFILE_NAME, "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. - # + if self._capture_trajectory: + artifacts_dir = str(evidence_dir / _ARTIFACTS_SUBDIR) + overlays.append( + profile_cls.from_mapping( + { + "name": _ARTIFACTS_PROFILE_NAME, + "runtime": {"artifacts": artifacts_dir}, + "environment": {"artifacts": artifacts_dir}, + } + ) + ) + return overlays + + def _relay_config(self, relay_dir: Path) -> dict[str, Any]: # 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. + # lazily since nemo-relay, like nemo-fabric, is an optional native dependency. try: from nemo_relay.observability import ( # ty: ignore[unresolved-import] AtifConfig, @@ -333,7 +408,6 @@ def _trajectory_profile( 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( @@ -351,33 +425,7 @@ def _trajectory_profile( ), ) ) - 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 _workspace_profile(self, profile_cls: type[FabricProfileConfig], *, workspace_dir: Path) -> FabricProfileConfig: - # Point the harness at this task's staged workspace via ``environment.workspace`` (the codex-cli - # adapter resolves its cwd from it). Set as a final profile overlay, the same mechanism the - # trajectory profile uses for ``environment.artifacts``. - return profile_cls.from_mapping( - { - "name": _WORKSPACE_PROFILE_NAME, - "description": "Run the harness in the per-task evaluation workspace seeded with the task inputs.", - "environment": {"workspace": str(workspace_dir)}, - } - ) + return {"version": 1, "components": [observability.to_dict()]} def _evidence_dir(self, index: int, task: AgentEvalTask, config: AgentEvalRunConfig) -> Path: root = self._work_root @@ -406,6 +454,18 @@ def _fabric_input(task: AgentEvalTask, seeded_files: Sequence[str] = ()) -> str: return "\n".join(lines) + "\n" +def _normalize_output(output: RunOutput | JsonValue) -> JsonValue: + """Unwrap a Fabric ``RunResult.output`` into the plain JSON value the trial response stores. + + Newer Fabric wraps output in a ``RunOutput`` (the RunOutput response contract), which is a + ``Mapping``; copy it into a plain dict (equivalent to its ``to_mapping()``). Raw/older JSON outputs + are already JSON values and pass through unchanged. + """ + if isinstance(output, Mapping): + return dict(output) + return output + + def _extract_output_text(output: object) -> str | None: """Pull the user-visible message out of a Fabric ``RunResult.output`` (JSON-shaped). 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 index 39870b6020..64af6626a1 100644 --- a/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_integration.py +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_integration.py @@ -14,6 +14,7 @@ from __future__ import annotations +import copy import importlib.util import json import os @@ -68,11 +69,45 @@ def _task() -> AgentEvalTask: # --- hermetic: fake nemo_fabric so CI exercises the runner+evaluator+metric+evidence chain ---------- +class _FakeEnvironment: + def __init__(self, *, provider: str = "local", workspace: str | None = None, artifacts: str | None = None) -> None: + self.provider = provider + self.workspace = workspace + self.artifacts = artifacts + + +class _FakeRuntimeCfg: + def __init__(self, artifacts: str | None = None) -> None: + self.artifacts = artifacts + + class _FakeConfig: + """Stand-in for nemo_fabric.FabricConfig supporting the config-first helpers the runtime uses.""" + + def __init__(self) -> None: + self.environment: _FakeEnvironment | None = None + self.runtime = _FakeRuntimeCfg() + self.models: dict[str, Any] = {} + self.relay: dict[str, Any] | None = None + @classmethod def from_mapping(cls, mapping: dict[str, Any]) -> _FakeConfig: return cls() + def model_copy(self, *, deep: bool = False) -> _FakeConfig: + clone = _FakeConfig() + clone.environment = copy.deepcopy(self.environment) + clone.runtime = _FakeRuntimeCfg(self.runtime.artifacts) + clone.models = copy.deepcopy(self.models) + clone.relay = copy.deepcopy(self.relay) + return clone + + def enable_relay( + self, *, project: str | None = None, output_dir: str | None = None, config: Any = None + ) -> _FakeConfig: + self.relay = {"project": project, "output_dir": output_dir, "config": config} + return self + class _FakeProfile: def __init__(self, **kwargs: Any) -> None: @@ -124,19 +159,20 @@ async def test_fabric_runner_eval_exposes_trajectory_to_metric(tmp_path: Path, m (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 - + # Fabric is a plain reusable facade (not an async context manager). async def run(self, agent: Any, **kwargs: Any) -> _FakeResult: return _FakeResult(artifacts) + class _FakeRunRequest: + def __init__(self, **kwargs: Any) -> None: + self.__dict__.update(kwargs) + module = types.ModuleType("nemo_fabric") - module.FabricClient = _FakeClient # type: ignore[attr-defined] + module.Fabric = _FakeClient # type: ignore[attr-defined] module.FabricConfig = _FakeConfig # type: ignore[attr-defined] module.FabricProfileConfig = _FakeProfile # type: ignore[attr-defined] + module.EnvironmentConfig = _FakeEnvironment # type: ignore[attr-defined] + module.RunRequest = _FakeRunRequest # 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 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 index cc5d97b759..6f9afb82a2 100644 --- a/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_runtime.py +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_runtime.py @@ -5,9 +5,11 @@ from __future__ import annotations +import copy import json import sys import types +from collections.abc import Iterator, Mapping from pathlib import Path from typing import Any @@ -17,14 +19,48 @@ from nemo_evaluator_sdk.values.evidence import EVIDENCE_FORMAT_ATIF, EVIDENCE_TRACE +class _FakeEnvironment: + """Stand-in for nemo_fabric.EnvironmentConfig (the runtime sets workspace/provider/artifacts).""" + + def __init__(self, *, provider: str = "local", workspace: str | None = None, artifacts: str | None = None) -> None: + self.provider = provider + self.workspace = workspace + self.artifacts = artifacts + + +class _FakeRuntimeCfg: + def __init__(self, artifacts: str | None = None) -> None: + self.artifacts = artifacts + + class _FakeConfig: + """Stand-in for nemo_fabric.FabricConfig with the config-first helpers the runtime composes onto.""" + def __init__(self, mapping: dict[str, Any]) -> None: self.mapping = mapping + self.environment: _FakeEnvironment | None = None + self.runtime = _FakeRuntimeCfg() + self.models: dict[str, Any] = dict(mapping.get("models", {})) + self.relay: dict[str, Any] | None = None # records enable_relay(...) @classmethod def from_mapping(cls, mapping: dict[str, Any]) -> _FakeConfig: return cls(mapping) + def model_copy(self, *, deep: bool = False) -> _FakeConfig: + clone = _FakeConfig(self.mapping) + clone.environment = copy.deepcopy(self.environment) + clone.runtime = _FakeRuntimeCfg(self.runtime.artifacts) + clone.models = copy.deepcopy(self.models) + clone.relay = copy.deepcopy(self.relay) + return clone + + def enable_relay( + self, *, project: str | None = None, output_dir: str | None = None, config: Any = None + ) -> _FakeConfig: + self.relay = {"project": project, "output_dir": output_dir, "config": config} + return self + class _FakeProfile: def __init__(self, *, name: str | None = None, models: Any = None, mapping: Any = None) -> None: @@ -37,6 +73,14 @@ def from_mapping(cls, mapping: dict[str, Any]) -> _FakeProfile: return cls(name=mapping.get("name"), mapping=mapping) +class _FakeRunRequest: + """Stand-in for nemo_fabric.RunRequest (Fabric.run folds input + request id into it).""" + + def __init__(self, *, input: Any = None, request_id: str | None = None) -> None: + self.input = input + self.request_id = request_id + + class _FakeRelayConfig: """Stand-in for nemo_relay.observability's typed config objects (AtifConfig/AtofConfig/...).""" @@ -119,23 +163,20 @@ 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: + # Fabric is a plain reusable facade (not an async context manager). 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.Fabric = _FakeClient # type: ignore[attr-defined] module.FabricConfig = _FakeConfig # type: ignore[attr-defined] module.FabricProfileConfig = _FakeProfile # type: ignore[attr-defined] + module.EnvironmentConfig = _FakeEnvironment # type: ignore[attr-defined] + module.RunRequest = _FakeRunRequest # 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); @@ -190,14 +231,12 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: 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" + # Config-first: the model is set on the config's default model and relay (ATIF trajectory) is + # enabled on the config, rather than layered as profile overlays. + composed = client_cls.recorded[0]["agent"] + assert composed.models["default"] == {"provider": "openai", "model": "openai/gpt-5.4"} + assert composed.relay is not None # capture_trajectory defaults on -> enable_relay(...) called + assert client_cls.recorded[0]["request"].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" @@ -226,10 +265,60 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: assert trace.ref == str(tmp_path / "trajectory.atif.json") -def _workspace_from_profiles(profiles: list[Any]) -> Path: - """Pull the staged workspace path out of the ``eval_workspace`` profile overlay.""" - profile = next(p for p in profiles if p.name == fabric_runtime._WORKSPACE_PROFILE_NAME) - return Path(profile.mapping["environment"]["workspace"]) +def _workspace_from_config(config: Any) -> Path: + """Pull the staged workspace path out of the composed per-task config.""" + return Path(config.environment.workspace) + + +def _resolve_like_fabric(config: Any, profiles: list[Any], section: str, key: str) -> Any: + """Mirror Fabric's resolver: start from the config, then apply each profile as a winning overlay in + order (last wins). Used to assert what value actually reaches the harness for a config/profile key. + """ + if section == "environment": + value = getattr(config.environment, key, None) if config.environment is not None else None + elif section == "models": + value = config.models.get(key) + else: # pragma: no cover - only the two sections above are exercised + raise ValueError(section) + for profile in profiles: + overlay = getattr(profile, "mapping", None) + if isinstance(overlay, Mapping) and isinstance(overlay.get(section), Mapping): + if overlay[section].get(key) is not None: + value = overlay[section][key] + return value + + +@pytest.mark.asyncio +async def test_caller_profiles_cannot_override_evaluator_owned_settings( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Fabric applies caller-supplied profiles over the config (last-wins), so the evaluator's per-task + # workspace (isolation + `workspace` evidence integrity) and model-under-eval must remain the final, + # authoritative layer. A caller profile that sets these must NOT win. + caller_profile = { + "name": "caller", + "environment": {"workspace": "/caller/hijacked-workspace"}, + "models": {"default": {"provider": "openai", "model": "caller/rogue-model"}}, + } + + 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, model="openai/gpt-5.4", work_root=tmp_path / "fabric", profiles=[caller_profile] + ) + + await runtime.run_tasks([_TASK]) + + config = client_cls.recorded[0]["agent"] + profiles = client_cls.recorded[0]["profiles"] + eval_workspace = config.environment.workspace # the per-task dir the evaluator composed + eval_model = config.models["default"] + + # After Fabric applies the caller profile, the evaluator's workspace + model must still win. + assert _resolve_like_fabric(config, profiles, "environment", "workspace") == eval_workspace + assert _resolve_like_fabric(config, profiles, "models", "default") == eval_model @pytest.mark.asyncio @@ -244,7 +333,7 @@ async def test_fabric_runtime_seeds_workspace_and_exposes_workspace_evidence( def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: # The harness runs in the staged workspace; simulate an edit it leaves behind. - workspace = _workspace_from_profiles(kwargs["profiles"]) + workspace = _workspace_from_config(agent) (workspace / "result.txt").write_text("done", encoding="utf-8") return _FakeResult(status="succeeded", output={"response": "ok"}) @@ -255,10 +344,10 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: trial = trials[0] assert trial.status == "completed" - # A dedicated workspace profile overlay carries environment.workspace (the harness's cwd). - profiles = client_cls.recorded[0]["profiles"] - assert fabric_runtime._WORKSPACE_PROFILE_NAME in [p.name for p in profiles] - workspace = _workspace_from_profiles(profiles) + # The composed config carries environment.workspace (the harness's cwd) with provider=local. + composed = client_cls.recorded[0]["agent"] + assert composed.environment.provider == "local" + workspace = _workspace_from_config(composed) # The seed file is staged and the agent's edit is present in the same dir. assert (workspace / "calc.py").read_text(encoding="utf-8") == "value = 1\n" assert (workspace / "result.txt").read_text(encoding="utf-8") == "done" @@ -268,7 +357,7 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: assert workspace_evidence.kind == "filesystem" assert workspace_evidence.ref == str(workspace) # Seed-file contents are listed by name in the input, not dumped inline (they are already on disk). - harness_input = client_cls.recorded[0]["input"] + harness_input = client_cls.recorded[0]["request"].input assert "calc.py" in harness_input assert "value = 1" not in harness_input @@ -287,9 +376,7 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: trials = await runtime.run_tasks([_TASK]) # _TASK has no 'files' input - profiles = client_cls.recorded[0]["profiles"] - assert fabric_runtime._WORKSPACE_PROFILE_NAME in [p.name for p in profiles] - workspace = _workspace_from_profiles(profiles) + workspace = _workspace_from_config(client_cls.recorded[0]["agent"]) assert workspace.is_dir() assert trials[0].evidence is not None assert trials[0].evidence.descriptors["workspace"].ref == str(workspace) @@ -317,7 +404,7 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: @pytest.mark.asyncio -async def test_fabric_runtime_capture_trajectory_false_skips_relay_overlay( +async def test_fabric_runtime_capture_trajectory_false_skips_relay( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: @@ -328,8 +415,7 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: await runtime.run_tasks([_TASK]) - names = [p.name for p in client_cls.recorded[0]["profiles"]] - assert "eval_trajectory" not in names + assert client_cls.recorded[0]["agent"].relay is None @pytest.mark.asyncio @@ -440,3 +526,34 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: assert result.outputs[0].name == "agent_phase_success" assert result.outputs[0].value is True + + +@pytest.mark.asyncio +async def test_fabric_runtime_normalizes_runoutput_response(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # Newer Fabric wraps RunResult.output in a RunOutput Mapping (not a plain JSON value); the runtime + # must copy it into a plain dict so it round-trips through the trial's JsonValue response field. + class _FakeRunOutput(Mapping): + def __init__(self, data: dict[str, Any]) -> None: + self._data = dict(data) + + def __getitem__(self, key: str) -> Any: + return self._data[key] + + def __iter__(self) -> Iterator[str]: + return iter(self._data) + + def __len__(self) -> int: + return len(self._data) + + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output=_FakeRunOutput({"response": "PONG", "returncode": 0})) + + _install_fake_fabric(monkeypatch, handler) + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, work_root=tmp_path / "fabric") + + trial = (await runtime.run_tasks([_TASK]))[0] + + assert trial.status == "completed" + assert trial.output is not None + assert trial.output.output_text == "PONG" # extracted from the normalized mapping + assert trial.output.response == {"response": "PONG", "returncode": 0} # plain dict, not RunOutput diff --git a/script/dev-install-fabric.sh b/script/dev-install-fabric.sh index cb20509872..6c966fc324 100755 --- a/script/dev-install-fabric.sh +++ b/script/dev-install-fabric.sh @@ -53,7 +53,7 @@ if [ ! -d "$FABRIC_REPO" ]; then 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__)" +"$VENV_PY" -c "import nemo_fabric; from nemo_fabric import Fabric, 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. 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 index c0cd728cfd..10cd7d7a5c 100644 --- 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 @@ -7,7 +7,11 @@ 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. +is applied as the config's default model, mirroring Fabric's own Harbor integration. + +Per-task settings (workspace, model, trajectory capture) are composed directly onto +a copy of the supplied config via the SDK's config helpers (``model_copy`` + +``enable_relay`` + ``environment``), rather than layered as profile overlays. Every task runs in its own fresh workspace: the runtime seeds it from ``inputs['files']`` (a no-op when there are none), runs the harness in it (via @@ -38,6 +42,7 @@ CandidateEvidence, EvidenceDescriptor, ) +from pydantic import JsonValue if TYPE_CHECKING: # Annotations use nemo_fabric's real types (single source of truth). nemo_fabric is an optional @@ -45,9 +50,10 @@ # 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, + Fabric, FabricConfig, FabricProfileConfig, + RunOutput, RunResult, ) @@ -66,18 +72,18 @@ # Per-task workspace: where seed files are staged and where the harness reads/writes. We # create it, point Fabric's ``environment.workspace`` at it, and expose it as ``workspace`` evidence. _WORKSPACE_SUBDIR = "workspace" -_WORKSPACE_PROFILE_NAME = "eval_workspace" # Evidence key + descriptor kind for the staged workspace, consumed by the # workspace-reading metrics. _WORKSPACE_EVIDENCE_KEY = "workspace" _WORKSPACE_EVIDENCE_KIND = "filesystem" -# Trajectory profile identity + the file-exporter output names we choose (Relay accepts these as inputs). -_TRAJECTORY_PROFILE_NAME = "eval_trajectory" +# File-exporter output names we choose for the Relay ATIF/ATOF trajectory (Relay accepts these as inputs). _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" +# Names for the trailing overlays that re-assert the evaluator-owned per-task settings (see +# ``_eval_lock_profiles``): Fabric applies caller profiles over the config, so these must trail them. +_WORKSPACE_PROFILE_NAME = "eval_workspace" +_MODEL_PROFILE_NAME = "eval_model" +_ARTIFACTS_PROFILE_NAME = "eval_artifacts" # ``kind`` Fabric stamps on the promoted Relay ATIF artifact; used to surface it as trace evidence. _ATIF_ARTIFACT_KIND = "atif" @@ -121,69 +127,75 @@ async def run_tasks( 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] + from nemo_fabric import Fabric, 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) + # Fail fast (once) if trajectory capture is requested but the nemo-relay gateway isn't + # importable, rather than failing every task the same way inside the per-task guard. + if self._capture_trajectory: + try: + import nemo_relay.observability # noqa: F401 # ty: ignore[unresolved-import] + except ImportError as exc: + raise RuntimeError(_MISSING_RELAY_MSG) from exc + # Caller-supplied profile overlays pass through as-is; this runtime's per-task workspace, model, + # and trajectory settings are composed directly onto a copy of the config (config-first), not + # layered as profiles. + base_profiles = [FabricProfileConfig.from_mapping(profile) for profile in self._profiles] semaphore = asyncio.Semaphore(resolved_config.parallelism) - async with FabricClient() as client: + # ``Fabric`` (formerly ``FabricClient``) is a lightweight, reusable facade — not a lifecycle + # context manager — so it is created once and reused across tasks with no cleanup. + client = Fabric() - 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 - ) + async def run_one(index: int, task: AgentEvalTask) -> AgentEvalTrial: + async with semaphore: + return await self._run_task(client, agent_config, base_profiles, index, task, resolved_config) - return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) + return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) async def _run_task( self, - client: FabricClient, + client: Fabric, agent_config: FabricConfig, base_profiles: list[FabricProfileConfig], - profile_cls: type[FabricProfileConfig], index: int, task: AgentEvalTask, config: AgentEvalRunConfig, ) -> AgentEvalTrial: + # nemo_fabric is already imported+validated in ``run_tasks``; this is a cached sys.modules + # lookup, not a re-load, so the type is used where it's constructed instead of threaded down. + from nemo_fabric import FabricProfileConfig, RunRequest # ty: ignore[unresolved-import] + 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)) - # Every task runs in its own fresh workspace: seed any ``inputs['files']`` into it (a no-op when - # there are none), point the harness at it via ``environment.workspace``, and expose it as - # ``workspace`` filesystem evidence — a uniform per-task dir that maps cleanly onto a per-task - # container volume later. Seeding runs inside the guarded block so a bad seed (a path escaping - # the workspace, an unresolvable fileset) fails just this task, not the whole run; it is - # synchronous and may block (a fileset handler downloads), so it is offloaded off the shared - # event loop. + # there are none), point the harness at it, and expose it as ``workspace`` filesystem evidence — + # a uniform per-task dir that maps cleanly onto a per-task container volume later. Seeding runs + # inside the guarded block so a bad seed (a path escaping the workspace, an unresolvable fileset) + # fails just this task, not the whole run; it is synchronous and may block (a fileset handler + # downloads), so it is offloaded off the shared event loop. workspace_dir = evidence_dir / _WORKSPACE_SUBDIR workspace_dir.mkdir(parents=True, exist_ok=True) try: seeded_files = await asyncio.to_thread(seed_workspace, workspace_dir, task.inputs.get(SEED_FILES_INPUT_KEY)) - profiles.append(self._workspace_profile(profile_cls, workspace_dir=workspace_dir)) + task_config = self._compose_config(agent_config, evidence_dir, workspace_dir) + # Caller ``base_profiles`` are applied by Fabric over the config; the evaluator-owned + # settings are re-asserted as trailing overlays so they win over any caller profile. + lock_profiles = self._eval_lock_profiles( + FabricProfileConfig, workspace_dir=workspace_dir, evidence_dir=evidence_dir + ) result = await asyncio.wait_for( + # ``Fabric.run`` folds the per-invocation input + request id into a ``RunRequest``. client.run( - agent_config, - profiles=profiles, - input=_fabric_input(task, seeded_files), - request_id=task.id, + task_config, + profiles=[*base_profiles, *lock_profiles], base_dir=self._base_dir, + request=RunRequest(input=_fabric_input(task, seeded_files), request_id=task.id), ), timeout=self._timeout_s, ) @@ -214,13 +226,17 @@ def _to_trial( if result.status != "succeeded": return self._failed_trial(task, evidence_dir, _result_error(result), extra_metadata=base_metadata) + # Fabric wraps the output in a ``RunOutput`` mapping (RunOutput response contract, #52), + # which is not itself a JSON value; normalize it to a plain mapping so it round-trips through the + # trial's ``JsonValue``-typed response. + output = _normalize_output(result.output) 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, + output_text=_extract_output_text(output), + response=output, metadata={**base_metadata, "evidence_dir": str(evidence_dir)}, ), evidence=self._evidence(result, result_path, workspace_dir), @@ -297,31 +313,90 @@ def _failed_trial( }, ) - def _build_profiles(self, profile_cls: type[FabricProfileConfig]) -> list[FabricProfileConfig]: - profiles = [profile_cls.from_mapping(profile) for profile in self._profiles] + def _compose_config( + self, + agent_config: FabricConfig, + evidence_dir: Path, + workspace_dir: Path, + ) -> FabricConfig: + # nemo_fabric is already imported+validated in ``run_tasks``; this is a cached sys.modules + # lookup, not a re-load, so the type is used where it's constructed instead of threaded down. + from nemo_fabric import EnvironmentConfig # ty: ignore[unresolved-import] + + # Config-first composition (the SDK's recommended in-memory pattern): copy the base config and + # apply this task's workspace, model, and trajectory settings directly onto it, rather than + # layering FabricProfileConfig overlays. + cfg = agent_config.model_copy(deep=True) + + # Point the harness at this task's staged workspace (the codex-cli adapter resolves its cwd from + # it). ``provider="local"`` is required by the native planner. Any config-supplied + # environment.workspace is overridden per task. + environment = cfg.environment or EnvironmentConfig(provider="local") + environment.provider = environment.provider or "local" + environment.workspace = str(workspace_dir) + cfg.environment = environment + + # Apply the model as the config's default (mirrors nemo_fabric.integrations.harbor). 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}}, + cfg.models["default"] = {"provider": provider, "model": self._model} + + if self._capture_trajectory: + # Enable Relay's ATIF/ATOF file exporter under this task's durable evidence dir, and pin the + # Fabric artifact root so the promoted ``trajectory-*.atif.json`` persists. Requires the + # ``nemo-relay`` gateway on PATH in the runtime. + 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) + cfg.enable_relay(output_dir=str(relay_dir), config=self._relay_config(relay_dir)) + cfg.runtime.artifacts = str(artifacts_dir) + cfg.environment.artifacts = str(artifacts_dir) + + return cfg + + def _eval_lock_profiles( + self, + profile_cls: type[FabricProfileConfig], + *, + workspace_dir: Path, + evidence_dir: Path, + ) -> list[FabricProfileConfig]: + # ``_compose_config`` composes the evaluator's per-task settings onto the config, but Fabric + # applies caller-supplied profiles OVER the config (last-wins), so a caller profile could + # otherwise override them. Re-assert the evaluator-owned settings here as trailing overlays — + # applied after the caller profiles — so the per-task workspace (isolation + ``workspace`` + # evidence integrity), the model under evaluation, and the trajectory artifact location stay + # authoritative and non-overridable. + overlays = [ + profile_cls.from_mapping( + {"name": _WORKSPACE_PROFILE_NAME, "environment": {"workspace": str(workspace_dir)}} + ) + ] + if self._model: + provider = self._model.split("/", maxsplit=1)[0] if "/" in self._model else "openai" + overlays.append( + profile_cls.from_mapping( + {"name": _MODEL_PROFILE_NAME, "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. - # + if self._capture_trajectory: + artifacts_dir = str(evidence_dir / _ARTIFACTS_SUBDIR) + overlays.append( + profile_cls.from_mapping( + { + "name": _ARTIFACTS_PROFILE_NAME, + "runtime": {"artifacts": artifacts_dir}, + "environment": {"artifacts": artifacts_dir}, + } + ) + ) + return overlays + + def _relay_config(self, relay_dir: Path) -> dict[str, Any]: # 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. + # lazily since nemo-relay, like nemo-fabric, is an optional native dependency. try: from nemo_relay.observability import ( # ty: ignore[unresolved-import] AtifConfig, @@ -333,7 +408,6 @@ def _trajectory_profile( 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( @@ -351,33 +425,7 @@ def _trajectory_profile( ), ) ) - 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 _workspace_profile(self, profile_cls: type[FabricProfileConfig], *, workspace_dir: Path) -> FabricProfileConfig: - # Point the harness at this task's staged workspace via ``environment.workspace`` (the codex-cli - # adapter resolves its cwd from it). Set as a final profile overlay, the same mechanism the - # trajectory profile uses for ``environment.artifacts``. - return profile_cls.from_mapping( - { - "name": _WORKSPACE_PROFILE_NAME, - "description": "Run the harness in the per-task evaluation workspace seeded with the task inputs.", - "environment": {"workspace": str(workspace_dir)}, - } - ) + return {"version": 1, "components": [observability.to_dict()]} def _evidence_dir(self, index: int, task: AgentEvalTask, config: AgentEvalRunConfig) -> Path: root = self._work_root @@ -406,6 +454,18 @@ def _fabric_input(task: AgentEvalTask, seeded_files: Sequence[str] = ()) -> str: return "\n".join(lines) + "\n" +def _normalize_output(output: RunOutput | JsonValue) -> JsonValue: + """Unwrap a Fabric ``RunResult.output`` into the plain JSON value the trial response stores. + + Newer Fabric wraps output in a ``RunOutput`` (the RunOutput response contract), which is a + ``Mapping``; copy it into a plain dict (equivalent to its ``to_mapping()``). Raw/older JSON outputs + are already JSON values and pass through unchanged. + """ + if isinstance(output, Mapping): + return dict(output) + return output + + def _extract_output_text(output: object) -> str | None: """Pull the user-visible message out of a Fabric ``RunResult.output`` (JSON-shaped).