From 93f637c4ef0d19a431fc0a7b0487f6cd645076e0 Mon Sep 17 00:00:00 2001 From: Sandy Chapman Date: Tue, 14 Jul 2026 10:15:47 -0300 Subject: [PATCH] feat(evaluator-sdk): agent-skill injection for Fabric agent-eval MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add agent-skill (agentskills.io) injection to FabricAgentRuntime so an A/B eval can score the same taskset with and without a skill: build one runtime with `skill=None` and one via `with_skill(skill)` over the same tasks, then diff the scores. Each run gets its own run-id evidence subtree so a baseline and a skilled variant sharing a work_root don't collide, and a SkillUsedMetric scores whether the agent actually engaged the injected skill. How the skill reaches the harness is decided by querying Fabric's own capability planner at runtime rather than a hardcoded adapter list: the runtime plans a copy of the config with a sentinel skill path attached and reads the resulting RunPlan.capability_plan routes. A `skills` route targeting `harness_native` means the adapter accepts the native Fabric `skills` config, so the bundle is handed over natively; otherwise a codex harness falls back to the `.agents/skills/` self-discovery convention, and anything else fails fast rather than run a skill-free trial mislabeled "with skill". Driving the decision from Fabric means it tracks whatever the installed adapters declare — including end-user adapters the platform doesn't ship (e.g. the `claude` adapter) — instead of a list that silently goes stale. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Sandy Chapman --- .../examples/skill_eval/.gitignore | 1 + .../examples/skill_eval/README.md | 107 ++++++ .../examples/skill_eval/run_skill_eval.py | 324 ++++++++++++++++++ .../skills/supercool-guidelines/SKILL.md | 27 ++ .../nemo_evaluator_sdk/agent_eval/metrics.py | 102 +++++- .../agent_eval/runtimes/fabric/runtime.py | 183 +++++++++- .../agent_eval/runtimes/fabric/skills.py | 272 +++++++++++++++ .../tests/agent_eval/test_fabric_runtime.py | 240 ++++++++++++- .../tests/agent_eval/test_fabric_skills.py | 267 +++++++++++++++ .../agent_eval/test_skill_used_metric.py | 72 ++++ .../beta/evaluator/agent_eval/metrics.py | 102 +++++- .../agent_eval/runtimes/fabric/runtime.py | 183 +++++++++- .../agent_eval/runtimes/fabric/skills.py | 272 +++++++++++++++ 13 files changed, 2123 insertions(+), 29 deletions(-) create mode 100644 packages/nemo_evaluator_sdk/examples/skill_eval/.gitignore create mode 100644 packages/nemo_evaluator_sdk/examples/skill_eval/README.md create mode 100644 packages/nemo_evaluator_sdk/examples/skill_eval/run_skill_eval.py create mode 100644 packages/nemo_evaluator_sdk/examples/skill_eval/skills/supercool-guidelines/SKILL.md create mode 100644 packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/skills.py create mode 100644 packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_skills.py create mode 100644 packages/nemo_evaluator_sdk/tests/agent_eval/test_skill_used_metric.py create mode 100644 sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/skills.py diff --git a/packages/nemo_evaluator_sdk/examples/skill_eval/.gitignore b/packages/nemo_evaluator_sdk/examples/skill_eval/.gitignore new file mode 100644 index 0000000000..5fbffdb86f --- /dev/null +++ b/packages/nemo_evaluator_sdk/examples/skill_eval/.gitignore @@ -0,0 +1 @@ +skill-eval-output/ diff --git a/packages/nemo_evaluator_sdk/examples/skill_eval/README.md b/packages/nemo_evaluator_sdk/examples/skill_eval/README.md new file mode 100644 index 0000000000..defe07f22c --- /dev/null +++ b/packages/nemo_evaluator_sdk/examples/skill_eval/README.md @@ -0,0 +1,107 @@ +# skill_eval — A/B evaluation of an injected agent skill + +Runs one taskset twice through the Fabric agent-eval runtime — once **without** a +skill (baseline) and once **with** an injected [agentskills.io](https://agentskills.io) +skill (treated, via `runtime.with_skill(skill)`) — then compares the scores. The +two arms differ in *exactly* the skill, so the delta is attributable to it. + +## The setup: the skill is *required* to pass + +Each task asks the agent to *"write a Python function that ..., following the +Supercool Coding Guidelines."* Those guidelines live **only** in the injected +`supercool-guidelines` skill and are **not inferable from the prompt**: + +- function names must start with `supercool_`; +- a positional-index parameter must be named `enieme` (French for "nth"). + +So the `follows_guidelines` metric can pass **only when the agent actually uses +the skill** — giving a clean, measured baseline-vs-treated difference rather than +a fuzzy "did it help". + +## What it demonstrates + +- Injecting an agentskills bundle into `FabricAgentRuntime` and running a + confound-free A/B with `with_skill(...)` (baseline vs. treated as distinct + `run_id`s). +- Task-authored guidelines metrics whose `follows_guidelines` output is + skill-dependent by construction, scored off the **parsed function signature** + (AST) so a convention only *mentioned* in prose doesn't count. Each task gets the + metric matching its parameters: `GuidelinesMetric` (index tasks) checks for the + `enieme` positional-index name — the unguessable discriminator — while + `GcdGuidelinesMetric` (gcd, no index param) checks its parameters are French + words. `supercool_prefix` is weakly guessable and reported on its own. +- `SkillUsedMetric` — `skill_present` / `skill_used` surface whether the agent + used the injected skill. + +The bundled skill is `skills/supercool-guidelines/` (a spec-compliant `SKILL.md`). + +## Run it + +**Prerequisites** — this example imports `nemo_evaluator_sdk` and drives the native +Fabric stack, so run it from the project virtualenv (a bare repo-root `python` +won't have the workspace on its import path): + +- `make bootstrap-python` — creates `.venv` and `uv sync --all-packages`, which + installs the workspace packages (including `nemo_evaluator_sdk`); +- `script/dev-install-fabric.sh` — the native `nemo-fabric` + Hermes SDK adapter + + `nemo-relay` gateway (not in the lockfile, so installed separately); +- `NVIDIA_API_KEY` for an account **provisioned for** `MODEL` in `run_skill_eval.py`. + +Then, from the repo root, run with the venv interpreter: + +```bash +NVIDIA_API_KEY=... ADAPTER_PYTHON="$(pwd)/.venv/bin/python" \ + .venv/bin/python -m packages.nemo_evaluator_sdk.examples.skill_eval.run_skill_eval +``` + +`ADAPTER_PYTHON` is required whenever the `python3` on your `PATH` is not this venv +(common on macOS/Homebrew, pyenv, etc.): the Fabric Hermes adapter runs as a +subprocess and otherwise falls back to a bare `python3` off `PATH`, which won't +have `nemo_fabric_adapters` installed (`python_adapter_exit_nonzero` / +`ModuleNotFoundError`). + +Each arm writes a run bundle under `skill-eval-output//`, with per-task +Fabric evidence under `evidence/fabric//`. If a trial fails (bad model +id, missing credential, harness crash), the run prints a `⚠️ N trial(s) FAILED` +block and exits non-zero rather than showing an empty-but-tidy table. + +Example output (with `nvidia/nemotron-3-super-120b-a12b`): + +```text +Harness: nvidia.fabric.hermes.sdk model: nvidia/nemotron-3-super-120b-a12b tasks: 2 +runs: baseline (baseline) vs treated (treated) + + metric.output baseline with-skill + agent_phase_success.agent_phase_success 2/2 2/2 + follows_guidelines.enieme_param 0/1 1/1 + follows_guidelines.follows_guidelines 0/2 2/2 + follows_guidelines.french_params 0/1 1/1 + follows_guidelines.supercool_prefix 0/2 2/2 + skill_used.skill_present 0/2 2/2 + skill_used.skill_used 0/2 2/2 +``` + +(`enieme_param` and `french_params` each total `/1` — they are the per-task checks, +emitted only by the index task and the gcd task respectively.) + +## Notes + +- `follows_guidelines` is the causal metric — it is what the skill directly + controls. `skill_used` is the *mechanism* signal; it detects the skill's staged + `location` in the trajectory and can under-report for the **Hermes** harness + (in-context loading), so treat `follows_guidelines` as the source of truth for + "did the skill take effect". See `SkillUsedMetric` for the detail. +- The model is the `MODEL` constant in `run_skill_eval.py`. It must be + **provisioned for your account** — some catalog-listed models return HTTP 404 + (`Function ... not found for account`). It also has to be capable enough to + *obey* the injected skill: weaker models read the guidelines but ignore them, + producing a flat A/B (a valid, if undramatic, result). + `nvidia/nemotron-3-super-120b-a12b` produces the lift shown above. +- The Hermes agent loop budget is `harness.settings.max_iterations` (set to 50 + here). The Fabric Hermes adapter defaults it to **1**, which starves any + multi-step task — leave it set. +- The guidelines metrics score the **parsed function signature** (`ast`, with a + `def`-regex fallback), scoping to the `supercool_`-named answer function so a + helper definition or a convention mentioned only in prose doesn't count. The gcd + `french_params` check uses a small illustrative French wordlist (`_FRENCH_PARAMS`) + — extend it if a run picks a French word not listed. diff --git a/packages/nemo_evaluator_sdk/examples/skill_eval/run_skill_eval.py b/packages/nemo_evaluator_sdk/examples/skill_eval/run_skill_eval.py new file mode 100644 index 0000000000..bcc7291ac5 --- /dev/null +++ b/packages/nemo_evaluator_sdk/examples/skill_eval/run_skill_eval.py @@ -0,0 +1,324 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""A/B *skill evaluation* over the Fabric agent-eval runtime. + +Runs one taskset twice through :class:`FabricAgentRuntime`: + +* **baseline** — no skill; +* **treated** — the same runtime with an injected `agentskills.io `_ skill + (``runtime.with_skill(skill)``), so the two arms differ in *exactly* the skill. + +Each arm is a separate run (distinct ``run_id``), scored with each task's metrics — including +``SkillUsedMetric``, whose ``skill_present`` / ``skill_used`` outputs surface whether the agent +actually used the injected skill. A ``skill_present=True, skill_used=False`` row is a failure to use +the skill; comparing the other metrics baseline-vs-treated shows whether the skill helped. + +Run as a module from the repository root. Needs the native NeMo Fabric stack and an ``NVIDIA_API_KEY`` +whose account is provisioned for ``MODEL``. If ``python3`` on ``PATH`` is not the interpreter running +this (e.g. a venv on macOS/Homebrew), also set ``ADAPTER_PYTHON`` to it — the Fabric Hermes adapter +spawns a subprocess and otherwise falls back to bare ``python3`` (which lacks ``nemo_fabric_adapters``):: + + NVIDIA_API_KEY=... ADAPTER_PYTHON="$(pwd)/.venv/bin/python" \\ + python -m packages.nemo_evaluator_sdk.examples.skill_eval.run_skill_eval +""" + +from __future__ import annotations + +import ast +import asyncio +import logging +import re +from pathlib import Path + +if __package__ in {None, ""}: + raise SystemExit( + "Run this example as a module from the repository root:\n" + " python -m packages.nemo_evaluator_sdk.examples.skill_eval.run_skill_eval" + ) + +from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator +from nemo_evaluator_sdk.agent_eval.metrics import AgentPhaseSuccessMetric, SkillUsedMetric +from nemo_evaluator_sdk.agent_eval.results import AgentEvalResult +from nemo_evaluator_sdk.agent_eval.runtimes.fabric.runtime import FabricAgentRuntime +from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill, SkillInjectionError +from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask +from nemo_evaluator_sdk.agent_eval.trials import AgentEvalTrialStatus +from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult + +# Model under evaluation. A mid-size NVIDIA Nemotron is capable enough to follow the injected skill's +# guidelines (so the A/B shows signal) without the latency of a frontier model. Served via +# ``integrate.api.nvidia.com`` (provider ``nvidia``), so it needs ``NVIDIA_API_KEY``. +MODEL = "nvidia/nemotron-3-super-120b-a12b" + + +# Score off the *parsed function signature*, not the whole reply text, so a convention merely mentioned +# in prose doesn't count. Primary path: AST-parse the reply's fenced code blocks (robust to bracketed +# annotations, multi-line signatures, ``*args``). Fallback: a ``def`` regex for code that doesn't parse +# cleanly (truncated / pseudo-code). +_CODE_BLOCK = re.compile(r"```(?:python|py)?\s*\n?(.*?)```", re.DOTALL) +_DEF_RE = re.compile(r"def\s+(?P\w+)\s*\((?P[^)]*)\)") +# French parameter names the gcd solution may legitimately use. The skill mandates French parameter +# names in general; gcd has no positional-index parameter, so its ``enieme`` rule does not apply here. +# Illustrative, not exhaustive — extend if a run uses a French word not listed. +_FRENCH_PARAMS = frozenset( + { + "premier", + "premiere", + "deuxieme", + "second", + "seconde", + "nombre", + "nombres", + "entier", + "entiers", + "valeur", + "valeurs", + "numero", + "chiffre", + "terme", + } +) + + +def _signatures_via_ast(text: str) -> list[tuple[str, list[str]]]: + """``(name, [param, ...])`` for every function defined in the reply's code blocks, via ``ast``.""" + functions: list[tuple[str, list[str]]] = [] + for block in _CODE_BLOCK.findall(text) or [text]: + try: + tree = ast.parse(block) + except SyntaxError: + continue + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + args = node.args + names = [arg.arg for arg in (*args.posonlyargs, *args.args, *args.kwonlyargs)] + if args.vararg: + names.append(args.vararg.arg) + if args.kwarg: + names.append(args.kwarg.arg) + functions.append((node.name, [name for name in names if name != "self"])) + return functions + + +def _param_names(raw: str) -> list[str]: + """Bare parameter names from a raw ``def`` parameter list — the regex fallback path.""" + params: list[str] = [] + for chunk in raw.split(","): + # "enieme: int = 0" / "*args" -> "enieme" / "args" + name = chunk.split(":", 1)[0].split("=", 1)[0].strip().lstrip("*").strip() + if name and name != "self": + params.append(name) + return params + + +def _scored_signature(text: str) -> tuple[str, list[str]]: + """Return the ``(name, [param, ...])`` of the function the guidelines govern. + + A reply often defines helper functions too, so scoring the first/last ``def`` is wrong. The skill + mandates the answer be named ``supercool_...``, so prefer the first ``supercool_``-named definition; + fall back to the first function defined, or ``("", [])`` if the reply defines none. + """ + signatures = _signatures_via_ast(text) + if not signatures: # code didn't parse (truncated / pseudo-code) — best-effort regex + signatures = [(m.group("name"), _param_names(m.group("params"))) for m in _DEF_RE.finditer(text)] + if not signatures: + return "", [] + for name, params in signatures: + if name.startswith("supercool_"): + return name, params + return signatures[0] + + +class GuidelinesMetric: + """Task-authored (positional-index tasks): does the solution follow the *Supercool Coding Guidelines*? + + Scored off the defined function's signature (so a convention only *mentioned* in prose does not + count): + + * ``supercool_prefix`` — the function name starts with ``supercool_`` (weakly guessable, reported + on its own); + * ``enieme_param`` — the positional-index parameter is named ``enieme`` (French for "nth"), the + unguessable skill-only signal; + * ``follows_guidelines`` — both. + + Use for a task whose function takes a positional index (e.g. "nth digit of pi"). Expected to pass + only in the treated (with-skill) arm. + """ + + @property + def type(self) -> str: + return "follows_guidelines" + + def output_spec(self) -> list[MetricOutputSpec]: + return [ + MetricOutputSpec.boolean("supercool_prefix"), + MetricOutputSpec.boolean("enieme_param"), + MetricOutputSpec.boolean("follows_guidelines"), + ] + + async def compute_scores(self, input: MetricInput) -> MetricResult: + name, params = _scored_signature(input.candidate.output_text or "") + prefix = name.startswith("supercool_") + enieme = "enieme" in params + return MetricResult( + outputs=[ + MetricOutput(name="supercool_prefix", value=prefix), + MetricOutput(name="enieme_param", value=enieme), + MetricOutput(name="follows_guidelines", value=prefix and enieme), + ] + ) + + +class GcdGuidelinesMetric: + """Task-authored (gcd): the *Supercool Coding Guidelines* for a task with **no** positional index. + + The skill's ``enieme`` rule is specific to a positional-index parameter, which gcd (two integers) + does not have — so this checks the skill's *general* parameter rule instead: names must be French. + + * ``supercool_prefix`` — the function name starts with ``supercool_``; + * ``french_params`` — every parameter name is a French word (:data:`_FRENCH_PARAMS`); baseline arms + use English names (``a``/``b``), so this is the skill-only signal; + * ``follows_guidelines`` — both. + """ + + @property + def type(self) -> str: + return "follows_guidelines" + + def output_spec(self) -> list[MetricOutputSpec]: + return [ + MetricOutputSpec.boolean("supercool_prefix"), + MetricOutputSpec.boolean("french_params"), + MetricOutputSpec.boolean("follows_guidelines"), + ] + + async def compute_scores(self, input: MetricInput) -> MetricResult: + name, params = _scored_signature(input.candidate.output_text or "") + prefix = name.startswith("supercool_") + french = bool(params) and all(param.lower() in _FRENCH_PARAMS for param in params) + return MetricResult( + outputs=[ + MetricOutput(name="supercool_prefix", value=prefix), + MetricOutput(name="french_params", value=french), + MetricOutput(name="follows_guidelines", value=prefix and french), + ] + ) + + +def skill_eval_tasks() -> list[AgentEvalTask]: + """Two "write a function following the Supercool Coding Guidelines" tasks. + + The guidelines (``supercool_`` prefix + French parameter names) live only in the injected + ``supercool-guidelines`` skill and are not inferable from the prompt, so ``follows_guidelines`` + should pass only in the treated (with-skill) arm — a measured, skill-dependent difference. Each task + gets the guidelines metric that matches its parameters: the index task checks for the ``enieme`` + positional-index name; gcd (no index param) checks that its parameters are French words. + """ + + def build(task_id: str, task: str, guidelines: GuidelinesMetric | GcdGuidelinesMetric) -> AgentEvalTask: + intent = f"Write a Python function that {task}, following the Supercool Coding Guidelines." + return AgentEvalTask( + id=task_id, + intent=intent, + inputs={"instruction": f"{intent} Include the complete function in your reply."}, + metrics=[AgentPhaseSuccessMetric(), SkillUsedMetric(), guidelines], + ) + + return [ + build("pi-digit", "returns the nth digit of pi", GuidelinesMetric()), + build("gcd", "returns the greatest common divisor of two integers", GcdGuidelinesMetric()), + ] + + +async def _main() -> int: + logging.basicConfig(level=logging.INFO, format="%(levelname)s:%(name)s:%(message)s") + logging.getLogger("httpx").setLevel(logging.WARNING) + + tasks = skill_eval_tasks() + current_dir = Path(__file__).resolve().parent + output_dir = current_dir / "skill-eval-output" + + fabric_config = { + "metadata": {"name": "skill-eval-hermes"}, + "harness": { + "adapter_id": "nvidia.fabric.hermes.sdk", + "resolution": "preinstalled", + "settings": {"max_iterations": 50}, + }, + "models": {"default": {"provider": "nvidia", "model": MODEL}}, + "runtime": {"mode": "oneshot", "transport": "library", "input_schema": "chat", "output_schema": "message"}, + } + + baseline_runtime = FabricAgentRuntime(config=fabric_config) + try: + # Load the bundled skill inside the guarded block so a packaging mistake (e.g. a missing + # SKILL.md) prints the friendly message instead of a traceback. + skill = AgentSkill.from_directory(current_dir / "skills" / "supercool-guidelines") + baseline = await AgentEvaluator().run( + tasks=tasks, + target=baseline_runtime, + config=AgentEvalRunConfig(run_id="baseline", output_dir=output_dir / "baseline", write_dashboard=False), + ) + treated = await AgentEvaluator().run( + tasks=tasks, + target=baseline_runtime.with_skill( + skill + ), # We include the skill in the treated arm, so the two runs differ in *exactly* the skill. + config=AgentEvalRunConfig(run_id="treated", output_dir=output_dir / "treated", write_dashboard=False), + ) + except SkillInjectionError as exc: + print(f"skill eval failed to load the bundled skill: {exc}") + return 1 + except RuntimeError as exc: + print(f"skill eval failed: {exc}") + print("This example needs the native NeMo Fabric stack and NVIDIA_API_KEY.") + return 1 + + # A failed trial produces no scorable output, so it would silently vanish from the tallies below and + # make the A/B look empty-but-fine. Surface failures loudly and treat any as a non-zero exit — a + # broken harness/model/credential must not read as "0/0, all good". + def failed_trials(result: AgentEvalResult) -> list[tuple[str, str]]: + failures: list[tuple[str, str]] = [] + for trial in result.trials: + if trial.status == AgentEvalTrialStatus.FAILED: + meta = trial.metadata or {} + reason = str(meta.get("error") or meta.get("error_type") or "unknown error") + failures.append((trial.task_id, reason)) + return failures + + all_failures = [("baseline", tid, err) for tid, err in failed_trials(baseline)] + all_failures += [("treated", tid, err) for tid, err in failed_trials(treated)] + if all_failures: + print(f"\n⚠️ {len(all_failures)} trial(s) FAILED — the A/B numbers below are unreliable:") + for arm, task_id, err in all_failures: + print(f" [{arm}] {task_id}: {err}") + + # Tally each boolean metric output (true/total) per arm, then print baseline vs. treated. + def rates(result: AgentEvalResult) -> dict[str, tuple[int, int]]: + counts: dict[str, tuple[int, int]] = {} + for score in result.scores: + for output in score.outputs: + if isinstance(output.value, bool): + key = f"{score.metric_type}.{output.name}" + true_count, total = counts.get(key, (0, 0)) + counts[key] = (true_count + int(output.value), total + 1) + return counts + + base, treat = rates(baseline), rates(treated) + print( + f"\nHarness: {fabric_config['harness']['adapter_id']} model: {MODEL} tasks: {baseline.summary.task_count}" + ) + print(f"runs: {baseline.run_id} (baseline) vs {treated.run_id} (treated)\n") + width = max((len(key) for key in set(base) | set(treat)), default=len("metric.output")) + print(f" {'metric.output'.ljust(width)} baseline with-skill") + for key in sorted(set(base) | set(treat)): + bt, bn = base.get(key, (0, 0)) + tt, tn = treat.get(key, (0, 0)) + print(f" {key.ljust(width)} {bt}/{bn} {tt}/{tn}") + print(f"\noutput: {output_dir}") + return 1 if all_failures else 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(_main())) diff --git a/packages/nemo_evaluator_sdk/examples/skill_eval/skills/supercool-guidelines/SKILL.md b/packages/nemo_evaluator_sdk/examples/skill_eval/skills/supercool-guidelines/SKILL.md new file mode 100644 index 0000000000..6236fc9ed9 --- /dev/null +++ b/packages/nemo_evaluator_sdk/examples/skill_eval/skills/supercool-guidelines/SKILL.md @@ -0,0 +1,27 @@ +--- +name: supercool-guidelines +description: The Supercool Coding Guidelines — mandatory naming conventions for every function you write. Use this whenever you are asked to write or edit a Python function. +--- + +# Supercool Coding Guidelines + +Every function you write MUST follow these conventions: + +1. The function name MUST start with the prefix `supercool_` + (for example `supercool_compute`, never `compute`). +2. Parameter names MUST be French words, written in ASCII (no accents). + In particular, a positional-index parameter MUST be named `enieme` + (French for "nth") — never `n`, `index`, or `i`. + +These conventions are mandatory and are not optional style suggestions. + +## Example + +```python +def supercool_racine(nombre): + return nombre ** 0.5 + + +def supercool_pi(enieme): + ... +``` diff --git a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/metrics.py b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/metrics.py index cf5512f8d2..3aa3b2e256 100644 --- a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/metrics.py +++ b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/metrics.py @@ -16,13 +16,22 @@ from __future__ import annotations +import json import logging from collections.abc import Mapping from typing import Any from nemo_evaluator_sdk.agent_eval.trials import EVIDENCE_FINAL_STATE -from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult -from pydantic import BaseModel, ConfigDict +from nemo_evaluator_sdk.metrics.protocol import ( + CandidateOutput, + MetricInput, + MetricOutput, + MetricOutputSpec, + MetricResult, +) +from nemo_evaluator_sdk.values.atif import Trajectory +from nemo_evaluator_sdk.values.evidence import EVIDENCE_TRACE +from pydantic import BaseModel, ConfigDict, ValidationError logger = logging.getLogger(__name__) @@ -105,6 +114,75 @@ async def compute_scores(self, input: MetricInput) -> MetricResult: return MetricResult(outputs=[MetricOutput(name=self._output_name, value=present)]) +class SkillUsedMetric: + """Emit ``skill_present`` and ``skill_used`` so an eval can flag a failure to use an injected skill. + + * ``skill_present`` — ``True`` when a skill was injected into the trial. Reads the provenance a + skill-aware runtime stamps onto candidate metadata under the ``"skill"`` key + (``{"name", "hash", "mode", "adapter_id", "location", ...}``, see ``fabric.skills.SkillProvenance``); + baseline trials carry none. + * ``skill_used`` — best-effort ``True`` when the agent referenced the injected skill in its ATIF + trajectory. It matches the skill's staged ``location`` (a specific, low-false-positive path + signal — e.g. a read of ``.agents/skills//SKILL.md``) against tool-call names/arguments, + step messages, reasoning, and observations. A bare skill-*name* match is intentionally NOT + counted (the name commonly appears in the task prompt), so ``skill_present=True, skill_used=False`` + flags a *likely* failure to use the skill. + + Limitation: an absent trajectory reference cannot fully distinguish "not used" from "used without + leaving a filesystem trace" — strongest for codex-style filesystem discovery, weaker for in-context + skill loading. Authoritative usage detection via harness skill-activation events is a follow-up. + With no skill present, both outputs are ``False``. + """ + + metric_type: str = "skill_used" + OUTPUT_PRESENT: str = "skill_present" + OUTPUT_USED: str = "skill_used" + # Metadata key skill-aware runtimes stamp the provenance under (matches the fabric runtime). + _METADATA_KEY: str = "skill" + + def __init__(self, *, trace_evidence: str = EVIDENCE_TRACE) -> None: + self._trace_evidence = trace_evidence + + @property + def type(self) -> str: + return self.metric_type + + def output_spec(self) -> list[MetricOutputSpec]: + return [ + MetricOutputSpec.boolean(self.OUTPUT_PRESENT), + MetricOutputSpec.boolean(self.OUTPUT_USED), + ] + + async def compute_scores(self, input: MetricInput) -> MetricResult: + provenance = input.candidate.metadata.get(self._METADATA_KEY) + present = isinstance(provenance, Mapping) and bool(provenance) + used = await self._skill_used(input.candidate, provenance) if present else False + return MetricResult( + outputs=[ + MetricOutput(name=self.OUTPUT_PRESENT, value=present), + MetricOutput(name=self.OUTPUT_USED, value=used), + ] + ) + + async def _skill_used(self, candidate: CandidateOutput, provenance: Mapping[str, Any]) -> bool: + location = provenance.get("location") + if not isinstance(location, str) or not location: + return False + evidence = candidate.evidence + if evidence is None or evidence.get(self._trace_evidence) is None: + return False + try: + trajectory = await (await evidence.trace(self._trace_evidence)).trace() + except (KeyError, ValueError, ValidationError, OSError) as exc: + # Best-effort: a missing/malformed/invalid trajectory must score skill_used=False, not raise. + # ValidationError covers Trajectory.model_validate; OSError covers the underlying file read. + logger.warning( + "SkillUsedMetric scored skill_used=False: could not read trace %r: %s", self._trace_evidence, exc + ) + return False + return _trajectory_references(trajectory, location) + + class TrialMeasurements(BaseModel): """Numeric measurements projected from trial metadata. @@ -146,6 +224,26 @@ def from_metadata(cls, metadata: Mapping[str, Any] | None) -> TrialMeasurements: ) +def _trajectory_references(trajectory: Trajectory, needle: str) -> bool: + """Whether ``needle`` appears anywhere an agent action could reference the skill. + + Scans each step's message, reasoning, tool calls (name + arguments), and observation results. + """ + for step in trajectory.steps: + if needle in step.message or (step.reasoning_content is not None and needle in step.reasoning_content): + return True + for call in step.tool_calls or []: + if needle in call.function_name: + return True + if call.arguments is not None and needle in json.dumps(call.arguments, default=str): + return True + if step.observation is not None: + for result in step.observation.results: + if result.content is not None and needle in json.dumps(result.content, default=str): + return True + return False + + def _as_int(value: Any) -> int | None: # bool is an int subclass; never treat True/False as a token count. if isinstance(value, bool): 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 5d21d3fc9c..9fcc32cd47 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 @@ -28,11 +28,22 @@ from __future__ import annotations import asyncio +import copy import json +import shutil from collections.abc import Mapping, Sequence +from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Any - +from uuid import uuid4 + +from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import ( + SKILL_MODE_CODEX_SKILLS_DIR, + AgentSkill, + SkillProvenance, + install_skill, + resolve_skill_mode, +) 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.agent_eval.workspace_seeds import SEED_FILES_INPUT_KEY, seed_workspace @@ -72,6 +83,14 @@ # 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" +# Per-task skill staging dir (native injection): the skill's files are resolved here and a per-task +# ``skills`` profile overlay points Fabric at it. For codex self-injection the skill lands in the +# workspace instead (no overlay). +_SKILL_SUBDIR = "skill" +# Sentinel skill path attached only to probe Fabric's capability planner for the selected adapter's +# skills routing (see ``_resolve_skill_mode``). Never staged and need not exist on disk — the planner +# just reports how it would route a skill for this adapter. +_SKILL_PROBE_PATH = "nemo-eval-skill-capability-probe" # Evidence key + descriptor kind for the staged workspace, consumed by the # workspace-reading metrics. _WORKSPACE_EVIDENCE_KEY = "workspace" @@ -79,13 +98,13 @@ # 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" +# ``kind`` Fabric stamps on the promoted Relay ATIF artifact; used to surface it as trace evidence. +_ATIF_ARTIFACT_KIND = "atif" # 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" class FabricAgentRuntime: @@ -109,6 +128,7 @@ def __init__( timeout_s: int = DEFAULT_FABRIC_TIMEOUT_S, capture_trajectory: bool = True, runtime_name: str = _RUNTIME_NAME, + skill: AgentSkill | None = None, ) -> None: self._config = config self._profiles = list(profiles or []) @@ -118,6 +138,19 @@ def __init__( self._timeout_s = timeout_s self._capture_trajectory = capture_trajectory self._runtime_name = runtime_name + self._skill = skill + + def with_skill(self, skill: AgentSkill | None) -> FabricAgentRuntime: + """Return a copy of this runtime with the skill replaced; ``self`` is not modified. + + Lets an A/B eval run the same taskset with and without a skill by deriving both runtimes from + one configured instance (baseline = ``with_skill(None)``, treated = ``with_skill(the_skill)``), + so they differ in exactly the skill and nothing else. A shallow copy suffices — the shared + fields are immutable config/paths. + """ + clone = copy.copy(self) + clone._skill = skill + return clone async def run_tasks( self, @@ -132,6 +165,11 @@ async def run_tasks( raise RuntimeError(_MISSING_FABRIC_MSG) from exc resolved_config = config or AgentEvalRunConfig() + # Assign a run id once per run so two runs (e.g. an A/B baseline vs. skilled variant) written + # under the same work_root/output_dir land in distinct, non-colliding evidence trees. Callers + # that set run_id keep their identifier. + if resolved_config.run_id is None: + resolved_config = resolved_config.model_copy(update={"run_id": _new_run_id()}) agent_config = FabricConfig.from_mapping(self._config) # 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. @@ -144,18 +182,72 @@ async def run_tasks( # 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) # ``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() + # Resolve once how a skill would reach this harness (the adapter is constant across tasks) by + # asking Fabric's own capability planner, so any adapter that declares native skills support — ours + # or an end-user's — is picked up automatically instead of via a hardcoded allow-list. Fail fast + # rather than silently run a skill-free trial mislabeled as "with skill", which would corrupt an + # A/B comparison. Only touched when a skill is set, so the no-skill path is unaffected. + skill_mode: str | None = None + if self._skill is not None: + skill_mode = self._resolve_skill_mode(client, agent_config, base_profiles) + if skill_mode is None: + adapter_id = agent_config.harness.adapter_id + raise RuntimeError( + f"FabricAgentRuntime received a skill but adapter {adapter_id!r} has no known " + "skill-injection strategy: Fabric does not route skills to it natively and it is not a " + "codex harness. Use a skills-native or codex harness, or drop the skill." + ) + + semaphore = asyncio.Semaphore(resolved_config.parallelism) + 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 self._run_task( + client, agent_config, base_profiles, index, task, resolved_config, skill_mode + ) return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) + def _resolve_skill_mode( + self, + client: Fabric, + agent_config: FabricConfig, + base_profiles: list[FabricProfileConfig], + ) -> str | None: + """Ask Fabric how a skill would reach the selected harness, or ``None`` if it can't. + + Probes Fabric's capability planner: plan a copy of the config with a sentinel skill path attached + (it need not exist on disk) and read how the adapter routes skills. Querying the authoritative + source at runtime means adapters that declare native skills support — ours or an end-user's — are + detected without a hardcoded list. See :func:`~...skills.resolve_skill_mode`. + """ + probe_config = agent_config.model_copy(deep=True) + probe_config.add_skill_path(_SKILL_PROBE_PATH) + plan = client.plan(probe_config, profiles=base_profiles, base_dir=self._base_dir) + return resolve_skill_mode(capability_plan=plan.capability_plan, harness=plan.adapter.harness) + + def _existing_skill_paths(self) -> list[str]: + """Skill paths the base config/profiles already declare (union, order-preserved). + + Fabric applies profile ``skills.paths`` last-wins, so the native skill overlay has to re-list + these alongside the evaluated skill or the treated arm would silently drop them (see + ``install_skill``). Read from the raw config/profile mappings the runtime was given, so it covers + both config- and profile-declared skills without a Fabric round-trip. + """ + paths: list[str] = [] + for section in (self._config, *self._profiles): + skills = section.get("skills") if isinstance(section, Mapping) else None + declared = skills.get("paths") if isinstance(skills, Mapping) else None + for path in declared or []: + if isinstance(path, str) and path not in paths: + paths.append(path) + return paths + async def _run_task( self, client: Fabric, @@ -164,9 +256,10 @@ async def _run_task( index: int, task: AgentEvalTask, config: AgentEvalRunConfig, + skill_mode: str | None, ) -> 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. + # lookup, not a re-load, so the types are used where they're constructed instead of threaded down. from nemo_fabric import FabricProfileConfig, RunRequest # ty: ignore[unresolved-import] evidence_dir = self._evidence_dir(index, task, config) @@ -180,10 +273,29 @@ async def _run_task( # downloads), so it is offloaded off the shared event loop. workspace_dir = evidence_dir / _WORKSPACE_SUBDIR workspace_dir.mkdir(parents=True, exist_ok=True) + skill_provenance: SkillProvenance | None = None try: # Stage seed files into the workspace for their on-disk side effect; the prompt is the task # instruction only, so the returned paths are unused. await asyncio.to_thread(seed_workspace, workspace_dir, task.inputs.get(SEED_FILES_INPUT_KEY)) + + # Inject the skill (if any) for this task. A native harness gets a per-task ``skills`` profile + # overlay; codex self-injection stages the bundle into the workspace and emits no overlay. + # Provenance is stamped on the trial for the A/B diff. Blocking file I/O, off the event loop. + skill_profiles: list[FabricProfileConfig] = [] + if self._skill is not None and skill_mode is not None: + installation = await asyncio.to_thread( + install_skill, + skill=self._skill, + adapter_id=agent_config.harness.adapter_id, + mode=skill_mode, + workspace_dir=workspace_dir, + skill_stage_dir=(evidence_dir / _SKILL_SUBDIR).resolve(), + existing_skill_paths=self._existing_skill_paths(), + ) + skill_provenance = installation.provenance + skill_profiles = [FabricProfileConfig.from_mapping(p) for p in installation.profiles] + 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. @@ -195,21 +307,35 @@ async def _run_task( # ``Fabric.run`` folds the per-invocation input + request id into a ``RunRequest``. client.run( task_config, - profiles=[*base_profiles, *lock_profiles], + # Caller profiles, then the native skill overlay, then the evaluator lock overlays; + # the lock overlays trail so the per-task workspace/model/artifacts stay authoritative. + profiles=[*base_profiles, *skill_profiles, *lock_profiles], base_dir=self._base_dir, request=RunRequest(input=task.agent_prompt(), request_id=task.id), ), timeout=self._timeout_s, ) except TimeoutError as exc: - return self._failed_trial(task, evidence_dir, exc) + return self._failed_trial(task, evidence_dir, exc, extra_metadata={"skill": skill_provenance}) 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._failed_trial(task, evidence_dir, exc, extra_metadata={"skill": skill_provenance}) - return self._to_trial(task, result, evidence_dir, workspace_dir) + # Codex self-injection staged the bundle *inside* the workspace so the harness could discover it. + # Now that the run is done (and captured in the trajectory), remove it before the workspace is + # exposed as filesystem evidence — otherwise the injected files read as agent output and skew + # workspace-reading metrics (a treated run with no agent-created files would look non-empty). + if skill_mode == SKILL_MODE_CODEX_SKILLS_DIR and skill_provenance is not None: + await asyncio.to_thread(_remove_injected_bundle, workspace_dir, skill_provenance["location"]) + return self._to_trial(task, result, evidence_dir, workspace_dir, skill_provenance=skill_provenance) def _to_trial( - self, task: AgentEvalTask, result: RunResult, evidence_dir: Path, workspace_dir: Path + self, + task: AgentEvalTask, + result: RunResult, + evidence_dir: Path, + workspace_dir: Path, + *, + skill_provenance: SkillProvenance | None = None, ) -> AgentEvalTrial: # Persist the full normalized Fabric result so graders (and debugging) can see the raw # envelope, and expose it as an evidence descriptor. @@ -223,6 +349,8 @@ def _to_trial( "adapter_kind": result.adapter_kind, "invocation_id": result.invocation_id, "agent_model": self._model, + # Skill provenance (name + content hash + injection mode) for the A/B diff; None baseline. + "skill": skill_provenance, } if result.status != "succeeded": @@ -433,9 +561,35 @@ def _evidence_dir(self, index: int, task: AgentEvalTask, config: AgentEvalRunCon root = self._work_root if root is None: root = (config.output_dir or Path.cwd()) / "evidence" / "fabric" + # The run id isolates this run's evidence from other runs sharing the same root (A/B baseline + # vs. skilled); run_tasks always populates it, so the fallback only guards a direct call. + run_id = config.run_id or _new_run_id() 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 + return Path(root) / _safe_path_name(run_id) / task_dir + + +def _remove_injected_bundle(workspace_dir: Path, location: str) -> None: + """Remove the Codex-injected skill subtree from ``workspace_dir`` and prune emptied parents. + + ``location`` is workspace-relative (``.agents/skills/``). Best-effort: the skill was already + captured in the run's trajectory, so SkillUsedMetric (which reads the trace, not the workspace) is + unaffected, and any filesystem error here must not fail an otherwise-successful trial. + """ + workspace_root = workspace_dir.resolve() + injected = (workspace_dir / location).resolve() + # Guard against a location escaping the workspace (defensive; provenance is evaluator-authored). + if workspace_root not in injected.parents or not injected.exists(): + return + shutil.rmtree(injected, ignore_errors=True) + # Prune now-empty reserved parents (``.agents/skills``, ``.agents``) but never the workspace itself. + parent = injected.parent + while parent != workspace_root and parent.is_dir(): + try: + parent.rmdir() # only succeeds while empty + except OSError: + break + parent = parent.parent def _normalize_output(output: RunOutput | JsonValue) -> JsonValue: @@ -477,3 +631,8 @@ def _result_error(result: RunResult) -> Mapping[str, Any]: def _safe_path_name(value: str) -> str: return "".join(char if char.isalnum() or char in "._-" else "-" for char in value).strip(".-")[:120] + + +def _new_run_id() -> str: + timestamp = datetime.now(UTC).strftime("%Y%m%d%H%M%S%f") + return f"fabric-{timestamp}-{uuid4().hex[:8]}" diff --git a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/skills.py b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/skills.py new file mode 100644 index 0000000000..ae834b8d3d --- /dev/null +++ b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/skills.py @@ -0,0 +1,272 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Agent-skill injection for the Fabric agent-eval runtimes (PROTOTYPE). + +An *agent skill* is a directory following the `agentskills.io `_ +spec: a folder named ``/`` containing a required ``SKILL.md`` (YAML frontmatter with ``name`` + +``description``, then instructions) plus optional ``scripts/`` / ``references/`` / ``assets/``. We make +that bundle available to the harness before it runs a task so an A/B eval can score the same taskset +with and without the skill. The skill is a runtime-level knob: build one runtime with ``skill=None`` +and one with ``skill=`` over the same tasks, then diff the scores. + +An :class:`AgentSkill` points at a local skill directory; staging is an OS-level ``copytree`` (file +contents never pass through Python memory). The plugin resolves a platform fileset to a local +directory and constructs an ``AgentSkill`` from it — the SDK has no fileset concept of its own. + +How the skill reaches the harness depends on the selected Fabric adapter, and which mode applies is +decided by *querying Fabric's own capability planner at runtime* (:func:`resolve_skill_mode` over a +``RunPlan.capability_plan``), not a hardcoded adapter list — so it tracks whatever the installed +adapters declare, including end-user adapters we don't ship: + +* **Native** (:data:`SKILL_MODE_NATIVE`): the adapter advertises ``accepts: ["skills", ...]`` (the + Hermes/Claude adapters do), so Fabric's planner routes skills to ``harness_native``. We stage the + bundle into an isolated ``/`` dir and hand Fabric a ``skills.paths`` profile overlay; the + adapter loads it (Hermes → harness ``skills.external_dirs``). +* **Codex skills dir** (:data:`SKILL_MODE_CODEX_SKILLS_DIR`): the Fabric ``codex`` adapter only + ``accepts: ["models"]`` (planner routes skills ``unsupported``), but the Codex CLI itself discovers + agentskills bundles from ``.agents/skills/`` in its working directory. So we place the bundle at + ``/.agents/skills//`` and let Codex discover it — same discoverable-skill semantics + as native (cross-harness A/B is apples-to-apples), no Fabric adapter change needed. + +If an adapter neither routes skills natively nor is a Codex harness, :func:`resolve_skill_mode` returns +``None`` and the runtime fails fast rather than silently running a skill-free trial. +""" + +from __future__ import annotations + +import hashlib +import re +import shutil +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import TypedDict + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +#: Required entry document of an agentskills bundle. +PRIMARY_SKILL_DOC = "SKILL.md" +#: Directory Codex scans (relative to its working dir) for agentskills bundles. +CODEX_SKILLS_DIR = ".agents/skills" +#: Name of the Fabric profile overlay that carries the native ``skills`` config. +SKILL_PROFILE_NAME = "eval_skill" + +#: Skill reaches the harness via the native Fabric ``skills`` config (adapter accepts it). +SKILL_MODE_NATIVE = "native" +#: Skill is placed under ``/.agents/skills//`` for Codex to discover. +SKILL_MODE_CODEX_SKILLS_DIR = "codex_skills_dir" + +# agentskills.io name rule: 1-64 chars, lowercase alphanumeric + single interior hyphens. +_SKILL_NAME_RE = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$") +_MAX_NAME_LEN = 64 + +# Fabric capability-planner vocabulary (``RunPlan.capability_plan['routes']`` entries). A ``skills`` +# route with target ``harness_native`` means the selected adapter declared native skills support; the +# runtime plans a probe skill path and reads these to decide the injection mode (see resolve_skill_mode). +_SKILLS_ROUTE_KIND = "skills" +_SKILLS_TARGET_NATIVE = "harness_native" +# Fabric harness name of the Codex CLI adapter, which self-discovers ``.agents/skills/`` rather than +# accepting the native ``skills`` config. +_CODEX_HARNESS = "codex" + + +class SkillInjectionError(ValueError): + """A skill could not be resolved, staged, or wired into the selected harness. + + Subclasses ``ValueError`` so the runtime's per-task error handling still catches it and fails + only that task. + """ + + +class AgentSkill(BaseModel): + """An agentskills.io bundle (a local directory) to make available to the agent before a task. + + ``name`` must satisfy the agentskills naming rule and is used as the staged bundle's directory name + (spec: the name matches the directory name). ``directory`` is the local skill directory, which must + contain a top-level ``SKILL.md``. + """ + + model_config = ConfigDict(extra="forbid") + + name: str = Field(description="agentskills skill name; also the bundle directory name and provenance id.") + directory: Path = Field(description="Local agentskills bundle directory (a SKILL.md at its root).") + + @field_validator("name") + @classmethod + def _valid_name(cls, value: str) -> str: + if len(value) > _MAX_NAME_LEN or not _SKILL_NAME_RE.match(value): + raise ValueError( + f"skill name {value!r} must be 1-{_MAX_NAME_LEN} chars, lowercase alphanumeric with " + "single interior hyphens (agentskills.io naming rule)" + ) + return value + + @classmethod + def from_directory(cls, directory: str | Path, *, name: str | None = None) -> AgentSkill: + """Build a skill from an on-disk agentskills bundle. ``name`` defaults to the directory basename.""" + root = Path(directory).expanduser().resolve() + if not (root / PRIMARY_SKILL_DOC).is_file(): + raise SkillInjectionError(f"skill directory {str(directory)!r} has no {PRIMARY_SKILL_DOC}") + return cls(name=name or root.name, directory=root) + + +class SkillProvenance(TypedDict): + """Which skill was injected into a trial and how; stamped into trial metadata for the A/B diff. + + A plain (JSON-serializable) dict so it drops straight into trial metadata. ``None`` in that slot + means the baseline (no skill). + """ + + name: str #: The skill's agentskills name. + hash: str #: sha256 over the staged bundle — attributes a score delta to an exact skill version. + mode: str #: How it was injected (:data:`SKILL_MODE_NATIVE` / :data:`SKILL_MODE_CODEX_SKILLS_DIR`). + adapter_id: str #: The harness adapter the skill was wired into. + location: str #: Where the bundle was staged (absolute for native, workspace-relative for codex). + + +@dataclass +class SkillInstallation: + """Result of installing a skill for one task. + + ``profiles`` are Fabric profile-overlay mappings the runtime appends to its profile stack (the + native branch emits one ``skills`` overlay; the Codex branch emits none because placement in the + workspace is the delivery mechanism). ``provenance`` is stamped into trial metadata so the A/B + comparison is auditable. + """ + + profiles: list[dict[str, object]] + provenance: SkillProvenance + + +def native_skills_route(capability_plan: Mapping[str, object]) -> bool: + """Whether Fabric's capability planner routed skills to the harness natively. + + ``capability_plan`` is the ``RunPlan.capability_plan`` mapping from ``Fabric.plan(...)`` planned with + a skill path attached; its ``routes`` record each capability decision. A ``skills`` route with target + ``harness_native`` means the selected adapter declares ``accepts: ["skills", ...]`` and Fabric hands + the bundle to the harness itself. Any other outcome (``unsupported``, or no skills route) is False. + """ + routes = capability_plan.get("routes") + if not isinstance(routes, list): + return False + return any( + isinstance(route, Mapping) + and route.get("kind") == _SKILLS_ROUTE_KIND + and route.get("target") == _SKILLS_TARGET_NATIVE + for route in routes + ) + + +def resolve_skill_mode(*, capability_plan: Mapping[str, object], harness: str) -> str | None: + """Resolve how a skill would reach the selected harness, or ``None`` if it can't. + + Driven by Fabric's own capability routing (queried at runtime via ``Fabric.plan``) rather than a + hardcoded adapter list, so it tracks whatever the installed adapters declare — including end-user + adapters we don't ship: + + * skills route natively (:func:`native_skills_route`) -> :data:`SKILL_MODE_NATIVE`; + * else a Codex harness (self-discovers ``.agents/skills/``) -> :data:`SKILL_MODE_CODEX_SKILLS_DIR`; + * else ``None`` -> the runtime fails fast rather than run a skill-free trial labeled "with skill". + """ + if native_skills_route(capability_plan): + return SKILL_MODE_NATIVE + if harness.strip().lower() == _CODEX_HARNESS: + return SKILL_MODE_CODEX_SKILLS_DIR + return None + + +def install_skill( + *, + skill: AgentSkill, + adapter_id: str, + mode: str, + workspace_dir: Path, + skill_stage_dir: Path, + existing_skill_paths: Sequence[str] = (), +) -> SkillInstallation: + """Stage ``skill`` as a ``/`` bundle and wire it into the harness per ``mode``. + + Blocking file I/O — call via ``asyncio.to_thread`` from the async runtime. The bundle is always + namespaced under ``/`` so it never collides with task-seeded workspace-root files; the content + hash is computed over the staged bytes so provenance tracks the actual skill content. + + ``existing_skill_paths`` are the skill paths the base config/profiles already declare. Fabric applies + profile ``skills.paths`` last-wins, so the native overlay must re-list them alongside the evaluated + skill — otherwise the treated arm would silently drop every preconfigured skill and the A/B would + differ by more than the injected skill. + """ + if mode == SKILL_MODE_NATIVE: + skill_root = skill_stage_dir / skill.name + _stage_bundle(skill.directory, skill_root, reserved=False) + # Preserve the pre-existing skill paths (order-preserved, de-duplicated) and append the + # evaluated skill, so the last-wins overlay reproduces the baseline skill set plus this one. + paths = list(dict.fromkeys([*existing_skill_paths, str(skill_root)])) + overlay: dict[str, object] = { + "name": SKILL_PROFILE_NAME, + "description": "Make the evaluation skill available via the native Fabric skills config.", + "skills": {"paths": paths}, + } + return SkillInstallation( + profiles=[overlay], + provenance=_provenance(skill, _hash_directory(skill_root), mode, adapter_id, str(skill_root)), + ) + + if mode == SKILL_MODE_CODEX_SKILLS_DIR: + skill_root = workspace_dir / CODEX_SKILLS_DIR / skill.name + _stage_bundle(skill.directory, skill_root, reserved=True) + location = (Path(CODEX_SKILLS_DIR) / skill.name).as_posix() + return SkillInstallation( + profiles=[], + provenance=_provenance(skill, _hash_directory(skill_root), mode, adapter_id, location), + ) + + raise SkillInjectionError(f"unknown skill injection mode {mode!r} for adapter {adapter_id!r}") + + +def _stage_bundle(directory: Path, skill_root: Path, *, reserved: bool) -> None: + """Stage the skill ``directory`` as an *exact* copy at ``skill_root`` (the ``/`` bundle dir). + + The staged bundle must reflect exactly the supplied directory, so provenance and behaviour track the + real content. ``reserved`` picks the collision policy for the destination: + + * ``reserved=False`` — the evaluator-owned native stage dir: recreate it, so a reused run id can't + leave a file that was since removed from the source bundle surviving in the stage. + * ``reserved=True`` — the Codex workspace path (``.agents/skills/``): refuse to clobber + pre-existing content there, since it can only be a task-seeded file colliding with the reserved + skill path. + """ + src = directory.expanduser() + if not (src / PRIMARY_SKILL_DOC).is_file(): + raise SkillInjectionError(f"skill directory {str(directory)!r} has no {PRIMARY_SKILL_DOC}") + if skill_root.exists(): + if reserved: + raise SkillInjectionError( + f"cannot stage skill into reserved path {str(skill_root)!r}: it already exists " + "(a task-seeded file collides with the injected skill bundle)" + ) + shutil.rmtree(skill_root) # evaluator-owned: recreate so the stage is an exact copy + skill_root.parent.mkdir(parents=True, exist_ok=True) + # OS-level copy — file contents never pass through Python memory. + shutil.copytree(src, skill_root) + + +def _provenance(skill: AgentSkill, skill_hash: str, mode: str, adapter_id: str, location: str) -> SkillProvenance: + return { + "name": skill.name, + "hash": skill_hash, + "mode": mode, + "adapter_id": adapter_id, + "location": location, + } + + +def _hash_directory(directory: Path) -> str: + """Stable sha256 over a directory's file tree (sorted relpath + contents).""" + digest = hashlib.sha256() + for path in sorted(path for path in directory.rglob("*") if path.is_file()): + digest.update(path.relative_to(directory).as_posix().encode("utf-8")) + digest.update(b"\0") + digest.update(path.read_bytes()) + digest.update(b"\0") + return digest.hexdigest() 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 5805cc9425..26aaa14ae6 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 @@ -33,15 +33,24 @@ def __init__(self, artifacts: str | None = None) -> None: self.artifacts = artifacts +class _FakeHarness: + """Stand-in for nemo_fabric FabricConfig.harness — the skill path reads ``adapter_id`` off it.""" + + def __init__(self, adapter_id: str) -> None: + self.adapter_id = adapter_id + + 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.harness = _FakeHarness(mapping.get("harness", {}).get("adapter_id", "")) 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(...) + self.skill_paths: list[str] = [] # records add_skill_path(...) (the capability-plan probe uses it) @classmethod def from_mapping(cls, mapping: dict[str, Any]) -> _FakeConfig: @@ -53,8 +62,12 @@ def model_copy(self, *, deep: bool = False) -> _FakeConfig: clone.runtime = _FakeRuntimeCfg(self.runtime.artifacts) clone.models = copy.deepcopy(self.models) clone.relay = copy.deepcopy(self.relay) + clone.skill_paths = list(self.skill_paths) return clone + def add_skill_path(self, path: Any) -> None: + self.skill_paths.append(str(path)) + def enable_relay( self, *, project: str | None = None, output_dir: str | None = None, config: Any = None ) -> _FakeConfig: @@ -135,6 +148,38 @@ def __init__(self, kind: str, message: str) -> None: self.message = message +# Adapters the fake planner reports as accepting the native Fabric ``skills`` config, mirroring +# adapters/*/fabric-adapter.json. ``acme.custom.native`` stands in for an END-USER adapter the platform +# doesn't ship — the runtime learns it accepts skills purely from the plan, with no hardcoded list. +_NATIVE_SKILL_ADAPTERS = { + "nvidia.fabric.hermes.sdk", + "nvidia.fabric.hermes.cli", + "nvidia.fabric.claude", + "acme.custom.native", +} +_KNOWN_HARNESSES = ("hermes", "codex", "claude", "deepagents") + + +def _harness_name(adapter_id: str) -> str: + """Derive the Fabric harness name from an adapter id (mirrors what the planner reports).""" + return next((h for h in _KNOWN_HARNESSES if h in adapter_id), "custom") + + +class _FakeAdapterInfo: + """Stand-in for nemo_fabric AdapterInfo — the runtime reads ``harness`` off ``RunPlan.adapter``.""" + + def __init__(self, harness: str) -> None: + self.harness = harness + + +class _FakePlan: + """Stand-in for nemo_fabric RunPlan (from Fabric.plan): capability routing + selected adapter.""" + + def __init__(self, *, capability_plan: dict[str, Any], harness: str) -> None: + self.capability_plan = capability_plan + self.adapter = _FakeAdapterInfo(harness) + + class _FakeResult: def __init__( self, @@ -165,12 +210,26 @@ def _install_fake_fabric(monkeypatch: pytest.MonkeyPatch, handler: Any) -> type: class _FakeClient: # Fabric is a plain reusable facade (not an async context manager). recorded: list[dict[str, Any]] = [] + planned: list[dict[str, Any]] = [] async def run(self, agent: Any, **kwargs: Any) -> Any: _FakeClient.recorded.append({"agent": agent, **kwargs}) return handler(agent, kwargs) + def plan(self, agent: Any, *, profiles: Any = None, base_dir: Any = None) -> _FakePlan: + # Mirror Fabric's capability planner: a ``skills`` route appears only when a skill path is + # attached, and it routes ``harness_native`` iff the selected adapter accepts native skills. + _FakeClient.planned.append({"agent": agent, "profiles": profiles, "base_dir": base_dir}) + adapter_id = agent.harness.adapter_id + has_skill_path = bool(getattr(agent, "skill_paths", None)) + native = has_skill_path and adapter_id in _NATIVE_SKILL_ADAPTERS + routes = ( + [{"kind": "skills", "target": "harness_native" if native else "unsupported"}] if has_skill_path else [] + ) + return _FakePlan(capability_plan={"routes": routes}, harness=_harness_name(adapter_id)) + _FakeClient.recorded = [] + _FakeClient.planned = [] module = types.ModuleType("nemo_fabric") module.Fabric = _FakeClient # type: ignore[attr-defined] module.FabricConfig = _FakeConfig # type: ignore[attr-defined] @@ -229,7 +288,8 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: 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" + # Evidence lands under a per-run id subdir (isolates A/B baseline vs. skilled runs sharing a root). + result_file = next((tmp_path / "fabric").glob("*/000000-task-1/fabric_result.json")) assert json.loads(result_file.read_text(encoding="utf-8"))["status"] == "succeeded" # 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. @@ -583,3 +643,181 @@ def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: 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 + + +def _skill_bundle(base: Path, *, name: str = "code-review", body: str = "Be thorough.") -> Path: + """Write a minimal agentskills bundle under ``base/skills//`` and return its path.""" + root = base / "skills" / name + root.mkdir(parents=True, exist_ok=True) + (root / "SKILL.md").write_text(f"---\nname: {name}\ndescription: d\n---\n\n{body}\n", encoding="utf-8") + return root + + +_HERMES_CONFIG = {"metadata": {"name": "a"}, "harness": {"adapter_id": "nvidia.fabric.hermes.sdk"}} + + +@pytest.mark.asyncio +async def test_fabric_runtime_native_skill_adds_overlay_and_provenance( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output={"response": "ok"}) + + client_cls = _install_fake_fabric(monkeypatch, handler) + skill = AgentSkill.from_directory(_skill_bundle(tmp_path, body="# Code Review\n\nBe thorough.")) + runtime = fabric_runtime.FabricAgentRuntime(config=_HERMES_CONFIG, work_root=tmp_path / "fabric", skill=skill) + + trials = await runtime.run_tasks([_TASK]) + + # The mode is resolved by probing Fabric's capability planner (with a skill path attached), not a + # hardcoded adapter list. + assert client_cls.planned, "expected the runtime to query Fabric.plan for skills routing" + assert client_cls.planned[0]["agent"].skill_paths, "expected a probe skill path attached for planning" + # A native `skills` overlay reaches client.run pointing at the staged / skill dir. + profiles = client_cls.recorded[0]["profiles"] + skill_profile = next(p for p in profiles if p.name == "eval_skill") + assert skill_profile.mapping["skills"]["paths"][0].endswith("/code-review") + # Provenance is stamped into trial metadata for the A/B diff. + prov = trials[0].metadata["skill"] + assert prov["name"] == "code-review" + assert prov["mode"] == "native" + assert prov["hash"] + + +@pytest.mark.asyncio +async def test_fabric_runtime_native_skill_preserves_preconfigured_skills( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Regression: Fabric applies profile skills.paths last-wins, so the native overlay must re-list any + # skills the config/profiles already declare — otherwise the treated arm would drop them and the A/B + # would differ by more than the injected skill. + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output={"response": "ok"}) + + client_cls = _install_fake_fabric(monkeypatch, handler) + config = {**_HERMES_CONFIG, "skills": {"paths": ["/pre/existing-a"]}} + skill = AgentSkill.from_directory(_skill_bundle(tmp_path)) + runtime = fabric_runtime.FabricAgentRuntime( + config=config, + work_root=tmp_path / "fabric", + skill=skill, + profiles=[{"name": "caller", "skills": {"paths": ["/pre/existing-b"]}}], + ) + + await runtime.run_tasks([_TASK]) + + overlay = next(p for p in client_cls.recorded[0]["profiles"] if p.name == "eval_skill") + paths = overlay.mapping["skills"]["paths"] + # Config- and profile-declared skills are preserved, in order, ahead of the evaluated skill. + assert paths[:2] == ["/pre/existing-a", "/pre/existing-b"] + assert paths[-1].endswith("/code-review") + + +@pytest.mark.asyncio +async def test_fabric_runtime_native_skill_on_runtime_discovered_adapter( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # An end-user adapter the platform doesn't ship (harness "custom", not codex) still gets native + # injection purely because Fabric's planner routes its skills ``harness_native`` — nothing is + # hardcoded, so runtime capability discovery is what makes this work. + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output={"response": "ok"}) + + client_cls = _install_fake_fabric(monkeypatch, handler) + custom = {"metadata": {"name": "a"}, "harness": {"adapter_id": "acme.custom.native"}} + skill = AgentSkill.from_directory(_skill_bundle(tmp_path)) + runtime = fabric_runtime.FabricAgentRuntime(config=custom, work_root=tmp_path / "fabric", skill=skill) + + trials = await runtime.run_tasks([_TASK]) + + profiles = client_cls.recorded[0]["profiles"] + assert any(p.name == "eval_skill" for p in profiles) + assert trials[0].metadata["skill"]["mode"] == "native" + + +@pytest.mark.asyncio +async def test_fabric_runtime_codex_skill_staged_for_run_then_excluded_from_evidence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + seen: dict[str, bool] = {} + + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + # Codex discovers the bundle from .agents/skills// in its workspace *during* the run. + workspace = Path(agent.environment.workspace) + seen["present_during_run"] = (workspace / ".agents" / "skills" / "code-review" / "SKILL.md").is_file() + return _FakeResult(status="succeeded", output={"response": "ok"}) + + client_cls = _install_fake_fabric(monkeypatch, handler) + skill = AgentSkill.from_directory(_skill_bundle(tmp_path)) + runtime = fabric_runtime.FabricAgentRuntime(config=_CONFIG, work_root=tmp_path / "fabric", skill=skill) + + trials = await runtime.run_tasks([_TASK]) + + # Staged into the workspace so the harness could discover it during the run... + assert seen["present_during_run"] is True + # ...then removed (with its emptied .agents parents) before the workspace is exposed as evidence, so + # the injected files don't read as agent output to workspace-reading metrics. + workspace = next((tmp_path / "fabric").glob("*/000000-task-1/workspace")) + assert not (workspace / ".agents").exists() + # No skills overlay; provenance still records the codex injection. + names = [p.name for p in client_cls.recorded[0]["profiles"]] + assert "eval_skill" not in names + assert trials[0].metadata["skill"]["mode"] == "codex_skills_dir" + + +@pytest.mark.asyncio +async def test_fabric_runtime_skill_on_unsupported_adapter_fails_fast( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output={"response": "ok"}) + + _install_fake_fabric(monkeypatch, handler) + unsupported = {"metadata": {"name": "a"}, "harness": {"adapter_id": "some.other.adapter"}} + runtime = fabric_runtime.FabricAgentRuntime( + config=unsupported, + work_root=tmp_path / "fabric", + skill=AgentSkill.from_directory(_skill_bundle(tmp_path, name="s")), + ) + + with pytest.raises(RuntimeError, match="no known skill-injection strategy"): + await runtime.run_tasks([_TASK]) + + +@pytest.mark.asyncio +async def test_fabric_runtime_no_skill_leaves_metadata_none(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + def handler(agent: Any, kwargs: dict[str, Any]) -> _FakeResult: + return _FakeResult(status="succeeded", output={"response": "ok"}) + + client_cls = _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].metadata["skill"] is None + # No skill -> no planner probe (the no-skill path must not pay for a plan()). + assert client_cls.planned == [] + + +def test_with_skill_returns_independent_copy() -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + base = fabric_runtime.FabricAgentRuntime(config=_HERMES_CONFIG, model="m", work_root="/tmp/x") + skill = AgentSkill(name="s", directory=Path("/skills/s")) + + treated = base.with_skill(skill) + + # A new instance is returned; the original is untouched. + assert treated is not base + assert base._skill is None + assert treated._skill is skill diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_skills.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_skills.py new file mode 100644 index 0000000000..bf78c974f6 --- /dev/null +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_skills.py @@ -0,0 +1,267 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for agent-skill injection (pure; no nemo_fabric native stack required).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import ( + CODEX_SKILLS_DIR, + SKILL_MODE_CODEX_SKILLS_DIR, + SKILL_MODE_NATIVE, + SKILL_PROFILE_NAME, + AgentSkill, + SkillInjectionError, + install_skill, + native_skills_route, + resolve_skill_mode, +) + + +def _plan(*, native: bool | None) -> dict[str, object]: + """Build a ``RunPlan.capability_plan``-shaped mapping. ``native=None`` means no skills route at all.""" + if native is None: + return {"routes": []} + return {"routes": [{"kind": "skills", "target": "harness_native" if native else "unsupported"}]} + + +_SKILL_MD = "---\nname: code-review\ndescription: Review code thoroughly.\n---\n\nBe thorough." + + +def _make_bundle(base: Path, name: str = "code-review", extra: dict[str, str] | None = None) -> Path: + root = base / name + root.mkdir(parents=True) + (root / "SKILL.md").write_text(_SKILL_MD, encoding="utf-8") + for rel, content in (extra or {}).items(): + target = root / rel + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + return root + + +def test_from_directory_defaults_name_from_basename(tmp_path: Path) -> None: + skill = AgentSkill.from_directory(_make_bundle(tmp_path)) + assert skill.name == "code-review" + assert skill.directory == (tmp_path / "code-review").resolve() + + +def test_from_directory_requires_skill_md(tmp_path: Path) -> None: + src = tmp_path / "no-skill" + src.mkdir() + (src / "notes.md").write_text("hi", encoding="utf-8") + with pytest.raises(SkillInjectionError): + AgentSkill.from_directory(src) + + +@pytest.mark.parametrize("bad", ["Code-Review", "-pdf", "pdf-", "pdf--processing", "has space", ""]) +def test_invalid_names_rejected(bad: str) -> None: + with pytest.raises(ValueError): + AgentSkill(name=bad, directory=Path("/skills/x")) + + +@pytest.mark.parametrize( + ("capability_plan", "expected"), + [ + ({"routes": [{"kind": "skills", "target": "harness_native"}]}, True), + ({"routes": [{"kind": "skills", "target": "unsupported"}]}, False), + ({"routes": [{"kind": "tools", "target": "harness_native"}]}, False), # non-skills route ignored + ({"routes": []}, False), + ({}, False), # no routes key + ({"routes": "not-a-list"}, False), # defensive: malformed shape + ], +) +def test_native_skills_route(capability_plan: dict[str, object], expected: bool) -> None: + assert native_skills_route(capability_plan) is expected + + +@pytest.mark.parametrize( + ("capability_plan", "harness", "expected"), + [ + # Native routing wins regardless of harness name (e.g. Hermes, or an end-user adapter). + (_plan(native=True), "hermes", SKILL_MODE_NATIVE), + (_plan(native=True), "acme-custom", SKILL_MODE_NATIVE), + # Not native, but a codex harness -> self-discovered .agents/skills dir. + (_plan(native=False), "codex", SKILL_MODE_CODEX_SKILLS_DIR), + (_plan(native=None), "codex", SKILL_MODE_CODEX_SKILLS_DIR), + (_plan(native=False), "CODEX", SKILL_MODE_CODEX_SKILLS_DIR), # case-insensitive + # Neither native nor codex -> unsupported (runtime fails fast). + (_plan(native=False), "hermes", None), + (_plan(native=None), "some-other", None), + ], +) +def test_resolve_skill_mode(capability_plan: dict[str, object], harness: str, expected: str | None) -> None: + assert resolve_skill_mode(capability_plan=capability_plan, harness=harness) == expected + + +def test_install_native_stages_named_dir_and_overlay(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + stage = tmp_path / "stage" + workspace.mkdir() + + installation = install_skill( + skill=AgentSkill.from_directory(_make_bundle(tmp_path / "src")), + adapter_id="nvidia.fabric.hermes.sdk", + mode=SKILL_MODE_NATIVE, + workspace_dir=workspace, + skill_stage_dir=stage, + ) + + # Bundle is staged under an isolated / dir (spec: name matches dir), not the agent workspace. + skill_root = stage / "code-review" + assert (skill_root / "SKILL.md").is_file() + assert not (workspace / "SKILL.md").exists() + assert not (workspace / ".agents").exists() + + overlay = installation.profiles[0] + assert overlay["name"] == SKILL_PROFILE_NAME + assert overlay["skills"] == {"paths": [str(skill_root)]} + + prov = installation.provenance + assert prov["name"] == "code-review" + assert prov["mode"] == SKILL_MODE_NATIVE + assert prov["location"] == str(skill_root) + assert isinstance(prov["hash"], str) and prov["hash"] + + +def test_install_native_copies_directory_tree(tmp_path: Path) -> None: + src = _make_bundle(tmp_path / "src", extra={"references/ref.md": "material", "scripts/run.py": "print()"}) + stage = tmp_path / "stage" + + install_skill( + skill=AgentSkill.from_directory(src), + adapter_id="nvidia.fabric.hermes.sdk", + mode=SKILL_MODE_NATIVE, + workspace_dir=tmp_path / "workspace", + skill_stage_dir=stage, + ) + + base = stage / "code-review" + assert (base / "SKILL.md").is_file() + assert (base / "references" / "ref.md").read_text(encoding="utf-8") == "material" + assert (base / "scripts" / "run.py").read_text(encoding="utf-8") == "print()" + + +def test_install_codex_places_under_agents_skills(tmp_path: Path) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + + installation = install_skill( + skill=AgentSkill.from_directory(_make_bundle(tmp_path / "src")), + adapter_id="nvidia.fabric.codex.cli", + mode=SKILL_MODE_CODEX_SKILLS_DIR, + workspace_dir=workspace, + skill_stage_dir=tmp_path / "stage", + ) + + # Codex discovers agentskills bundles from .agents/skills/ in its working directory. + skill_md = workspace / ".agents" / "skills" / "code-review" / "SKILL.md" + assert "Be thorough." in skill_md.read_text(encoding="utf-8") + # No profile overlay: placement in the workspace is the delivery mechanism. + assert installation.profiles == [] + assert installation.provenance["mode"] == SKILL_MODE_CODEX_SKILLS_DIR + assert installation.provenance["location"] == f"{CODEX_SKILLS_DIR}/code-review" + + +def test_codex_bundle_does_not_collide_with_workspace_root(tmp_path: Path) -> None: + # A task-seeded workspace file at the root is untouched: the skill lives under .agents/skills/. + workspace = tmp_path / "workspace" + workspace.mkdir() + (workspace / "data.csv").write_text("task input", encoding="utf-8") + src = _make_bundle(tmp_path / "src", name="collide", extra={"data.csv": "skill payload"}) + + install_skill( + skill=AgentSkill.from_directory(src), + adapter_id="nvidia.fabric.codex.cli", + mode=SKILL_MODE_CODEX_SKILLS_DIR, + workspace_dir=workspace, + skill_stage_dir=tmp_path / "stage", + ) + + assert (workspace / "data.csv").read_text(encoding="utf-8") == "task input" + assert (workspace / ".agents" / "skills" / "collide" / "data.csv").read_text(encoding="utf-8") == "skill payload" + + +def test_install_native_preserves_existing_skill_paths(tmp_path: Path) -> None: + # Fabric applies profile skills.paths last-wins, so the overlay must carry the pre-existing paths + # (order-preserved) alongside the evaluated skill, or the treated arm would drop them. + installation = install_skill( + skill=AgentSkill.from_directory(_make_bundle(tmp_path / "src")), + adapter_id="nvidia.fabric.hermes.sdk", + mode=SKILL_MODE_NATIVE, + workspace_dir=tmp_path / "workspace", + skill_stage_dir=tmp_path / "stage", + existing_skill_paths=["/pre/a", "/pre/b", "/pre/a"], # duplicate is collapsed + ) + + paths = installation.profiles[0]["skills"]["paths"] + assert paths[:2] == ["/pre/a", "/pre/b"] + assert paths[-1] == str(tmp_path / "stage" / "code-review") + + +def test_install_native_recreates_stale_stage(tmp_path: Path) -> None: + # Re-staging into an existing stage (reused run id) must yield an *exact* copy of the source — a file + # since removed from the bundle must not survive. + stage = tmp_path / "stage" + install_skill( + skill=AgentSkill.from_directory(_make_bundle(tmp_path / "v1", extra={"old.md": "stale"})), + adapter_id="nvidia.fabric.hermes.sdk", + mode=SKILL_MODE_NATIVE, + workspace_dir=tmp_path / "workspace", + skill_stage_dir=stage, + ) + assert (stage / "code-review" / "old.md").exists() + + install_skill( + skill=AgentSkill.from_directory(_make_bundle(tmp_path / "v2")), # no old.md + adapter_id="nvidia.fabric.hermes.sdk", + mode=SKILL_MODE_NATIVE, + workspace_dir=tmp_path / "workspace", + skill_stage_dir=stage, + ) + assert (stage / "code-review" / "SKILL.md").is_file() + assert not (stage / "code-review" / "old.md").exists() # stale file recreated away + + +def test_install_codex_rejects_reserved_path_collision(tmp_path: Path) -> None: + # A task seed occupying the reserved Codex skill path must not be silently clobbered/merged. + workspace = tmp_path / "workspace" + reserved = workspace / CODEX_SKILLS_DIR / "code-review" + reserved.mkdir(parents=True) + (reserved / "task_seed.txt").write_text("task file at the reserved path", encoding="utf-8") + + with pytest.raises(SkillInjectionError, match="reserved path"): + install_skill( + skill=AgentSkill.from_directory(_make_bundle(tmp_path / "src")), + adapter_id="nvidia.fabric.codex.cli", + mode=SKILL_MODE_CODEX_SKILLS_DIR, + workspace_dir=workspace, + skill_stage_dir=tmp_path / "stage", + ) + + +def test_hash_is_content_sensitive(tmp_path: Path) -> None: + one = tmp_path / "one" / "code-review" + two = tmp_path / "two" / "code-review" + one.mkdir(parents=True) + two.mkdir(parents=True) + (one / "SKILL.md").write_text("one", encoding="utf-8") + (two / "SKILL.md").write_text("two", encoding="utf-8") + + a = install_skill( + skill=AgentSkill.from_directory(one), + adapter_id="nvidia.fabric.hermes.sdk", + mode=SKILL_MODE_NATIVE, + workspace_dir=tmp_path / "wa", + skill_stage_dir=tmp_path / "sa", + ) + b = install_skill( + skill=AgentSkill.from_directory(two), + adapter_id="nvidia.fabric.hermes.sdk", + mode=SKILL_MODE_NATIVE, + workspace_dir=tmp_path / "wb", + skill_stage_dir=tmp_path / "sb", + ) + assert a.provenance["hash"] != b.provenance["hash"] diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_skill_used_metric.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_skill_used_metric.py new file mode 100644 index 0000000000..40aad0a9a6 --- /dev/null +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_skill_used_metric.py @@ -0,0 +1,72 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for SkillUsedMetric (skill_present / skill_used).""" + +from __future__ import annotations + +from typing import Any + +import pytest +from nemo_evaluator_sdk.agent_eval.metrics import SkillUsedMetric +from nemo_evaluator_sdk.execution.samples import build_metric_input +from nemo_evaluator_sdk.values.evidence import EVIDENCE_TRACE, CandidateEvidence, EvidenceDescriptor + +_LOCATION = ".agents/skills/code-review" +_PROV = { + "name": "code-review", + "hash": "deadbeef", + "mode": "codex_skills_dir", + "adapter_id": "nvidia.fabric.codex.cli", + "location": _LOCATION, +} + + +def _atif(*, tool_path: str | None = None, message: str = "working") -> dict[str, Any]: + step: dict[str, Any] = {"source": "agent", "message": message} + if tool_path is not None: + step["tool_calls"] = [{"function_name": "read_file", "arguments": {"path": tool_path}}] + return {"schema_version": "ATIF-v1.7", "steps": [step]} + + +def _evidence(atif: dict[str, Any]) -> CandidateEvidence: + return CandidateEvidence(descriptors={EVIDENCE_TRACE: EvidenceDescriptor(kind="trace", format="atif", data=atif)}) + + +async def _score(sample: dict[str, Any]) -> dict[str, Any]: + result = await SkillUsedMetric().compute_scores(build_metric_input({}, sample, 0)) + return {output.name: output.value for output in result.outputs} + + +def test_output_spec_declares_two_booleans() -> None: + specs = SkillUsedMetric().output_spec() + assert [spec.name for spec in specs] == ["skill_present", "skill_used"] + + +@pytest.mark.asyncio +async def test_no_skill_present_both_false() -> None: + assert await _score({}) == {"skill_present": False, "skill_used": False} + + +@pytest.mark.asyncio +async def test_present_and_used_when_trajectory_reads_the_skill() -> None: + sample = {"skill": _PROV, "evidence": _evidence(_atif(tool_path=f"{_LOCATION}/SKILL.md"))} + assert await _score(sample) == {"skill_present": True, "skill_used": True} + + +@pytest.mark.asyncio +async def test_present_not_used_when_trajectory_ignores_the_skill() -> None: + sample = {"skill": _PROV, "evidence": _evidence(_atif(tool_path="README.md"))} + assert await _score(sample) == {"skill_present": True, "skill_used": False} + + +@pytest.mark.asyncio +async def test_present_not_used_without_a_trace() -> None: + assert await _score({"skill": _PROV}) == {"skill_present": True, "skill_used": False} + + +@pytest.mark.asyncio +async def test_bare_name_mention_is_not_counted_as_used() -> None: + # The skill *name* appears in a message, but not its staged location — not counted as used. + sample = {"skill": _PROV, "evidence": _evidence(_atif(message="starting the code-review task"))} + assert (await _score(sample))["skill_used"] is False diff --git a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/metrics.py b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/metrics.py index de60099921..01007cdefd 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/metrics.py +++ b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/metrics.py @@ -16,13 +16,22 @@ from __future__ import annotations +import json import logging from collections.abc import Mapping from typing import Any from nemo_platform.beta.evaluator.agent_eval.trials import EVIDENCE_FINAL_STATE -from nemo_platform.beta.evaluator.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult -from pydantic import BaseModel, ConfigDict +from nemo_platform.beta.evaluator.metrics.protocol import ( + CandidateOutput, + MetricInput, + MetricOutput, + MetricOutputSpec, + MetricResult, +) +from nemo_platform.beta.evaluator.values.atif import Trajectory +from nemo_platform.beta.evaluator.values.evidence import EVIDENCE_TRACE +from pydantic import BaseModel, ConfigDict, ValidationError logger = logging.getLogger(__name__) @@ -105,6 +114,75 @@ async def compute_scores(self, input: MetricInput) -> MetricResult: return MetricResult(outputs=[MetricOutput(name=self._output_name, value=present)]) +class SkillUsedMetric: + """Emit ``skill_present`` and ``skill_used`` so an eval can flag a failure to use an injected skill. + + * ``skill_present`` — ``True`` when a skill was injected into the trial. Reads the provenance a + skill-aware runtime stamps onto candidate metadata under the ``"skill"`` key + (``{"name", "hash", "mode", "adapter_id", "location", ...}``, see ``fabric.skills.SkillProvenance``); + baseline trials carry none. + * ``skill_used`` — best-effort ``True`` when the agent referenced the injected skill in its ATIF + trajectory. It matches the skill's staged ``location`` (a specific, low-false-positive path + signal — e.g. a read of ``.agents/skills//SKILL.md``) against tool-call names/arguments, + step messages, reasoning, and observations. A bare skill-*name* match is intentionally NOT + counted (the name commonly appears in the task prompt), so ``skill_present=True, skill_used=False`` + flags a *likely* failure to use the skill. + + Limitation: an absent trajectory reference cannot fully distinguish "not used" from "used without + leaving a filesystem trace" — strongest for codex-style filesystem discovery, weaker for in-context + skill loading. Authoritative usage detection via harness skill-activation events is a follow-up. + With no skill present, both outputs are ``False``. + """ + + metric_type: str = "skill_used" + OUTPUT_PRESENT: str = "skill_present" + OUTPUT_USED: str = "skill_used" + # Metadata key skill-aware runtimes stamp the provenance under (matches the fabric runtime). + _METADATA_KEY: str = "skill" + + def __init__(self, *, trace_evidence: str = EVIDENCE_TRACE) -> None: + self._trace_evidence = trace_evidence + + @property + def type(self) -> str: + return self.metric_type + + def output_spec(self) -> list[MetricOutputSpec]: + return [ + MetricOutputSpec.boolean(self.OUTPUT_PRESENT), + MetricOutputSpec.boolean(self.OUTPUT_USED), + ] + + async def compute_scores(self, input: MetricInput) -> MetricResult: + provenance = input.candidate.metadata.get(self._METADATA_KEY) + present = isinstance(provenance, Mapping) and bool(provenance) + used = await self._skill_used(input.candidate, provenance) if present else False + return MetricResult( + outputs=[ + MetricOutput(name=self.OUTPUT_PRESENT, value=present), + MetricOutput(name=self.OUTPUT_USED, value=used), + ] + ) + + async def _skill_used(self, candidate: CandidateOutput, provenance: Mapping[str, Any]) -> bool: + location = provenance.get("location") + if not isinstance(location, str) or not location: + return False + evidence = candidate.evidence + if evidence is None or evidence.get(self._trace_evidence) is None: + return False + try: + trajectory = await (await evidence.trace(self._trace_evidence)).trace() + except (KeyError, ValueError, ValidationError, OSError) as exc: + # Best-effort: a missing/malformed/invalid trajectory must score skill_used=False, not raise. + # ValidationError covers Trajectory.model_validate; OSError covers the underlying file read. + logger.warning( + "SkillUsedMetric scored skill_used=False: could not read trace %r: %s", self._trace_evidence, exc + ) + return False + return _trajectory_references(trajectory, location) + + class TrialMeasurements(BaseModel): """Numeric measurements projected from trial metadata. @@ -146,6 +224,26 @@ def from_metadata(cls, metadata: Mapping[str, Any] | None) -> TrialMeasurements: ) +def _trajectory_references(trajectory: Trajectory, needle: str) -> bool: + """Whether ``needle`` appears anywhere an agent action could reference the skill. + + Scans each step's message, reasoning, tool calls (name + arguments), and observation results. + """ + for step in trajectory.steps: + if needle in step.message or (step.reasoning_content is not None and needle in step.reasoning_content): + return True + for call in step.tool_calls or []: + if needle in call.function_name: + return True + if call.arguments is not None and needle in json.dumps(call.arguments, default=str): + return True + if step.observation is not None: + for result in step.observation.results: + if result.content is not None and needle in json.dumps(result.content, default=str): + return True + return False + + def _as_int(value: Any) -> int | None: # bool is an int subclass; never treat True/False as a token count. if isinstance(value, bool): 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 7e7f02468c..d596f2d6e3 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 @@ -28,11 +28,22 @@ from __future__ import annotations import asyncio +import copy import json +import shutil from collections.abc import Mapping, Sequence +from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Any - +from uuid import uuid4 + +from nemo_platform.beta.evaluator.agent_eval.runtimes.fabric.skills import ( + SKILL_MODE_CODEX_SKILLS_DIR, + AgentSkill, + SkillProvenance, + install_skill, + resolve_skill_mode, +) 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.agent_eval.workspace_seeds import SEED_FILES_INPUT_KEY, seed_workspace @@ -72,6 +83,14 @@ # 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" +# Per-task skill staging dir (native injection): the skill's files are resolved here and a per-task +# ``skills`` profile overlay points Fabric at it. For codex self-injection the skill lands in the +# workspace instead (no overlay). +_SKILL_SUBDIR = "skill" +# Sentinel skill path attached only to probe Fabric's capability planner for the selected adapter's +# skills routing (see ``_resolve_skill_mode``). Never staged and need not exist on disk — the planner +# just reports how it would route a skill for this adapter. +_SKILL_PROBE_PATH = "nemo-eval-skill-capability-probe" # Evidence key + descriptor kind for the staged workspace, consumed by the # workspace-reading metrics. _WORKSPACE_EVIDENCE_KEY = "workspace" @@ -79,13 +98,13 @@ # 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" +# ``kind`` Fabric stamps on the promoted Relay ATIF artifact; used to surface it as trace evidence. +_ATIF_ARTIFACT_KIND = "atif" # 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" class FabricAgentRuntime: @@ -109,6 +128,7 @@ def __init__( timeout_s: int = DEFAULT_FABRIC_TIMEOUT_S, capture_trajectory: bool = True, runtime_name: str = _RUNTIME_NAME, + skill: AgentSkill | None = None, ) -> None: self._config = config self._profiles = list(profiles or []) @@ -118,6 +138,19 @@ def __init__( self._timeout_s = timeout_s self._capture_trajectory = capture_trajectory self._runtime_name = runtime_name + self._skill = skill + + def with_skill(self, skill: AgentSkill | None) -> FabricAgentRuntime: + """Return a copy of this runtime with the skill replaced; ``self`` is not modified. + + Lets an A/B eval run the same taskset with and without a skill by deriving both runtimes from + one configured instance (baseline = ``with_skill(None)``, treated = ``with_skill(the_skill)``), + so they differ in exactly the skill and nothing else. A shallow copy suffices — the shared + fields are immutable config/paths. + """ + clone = copy.copy(self) + clone._skill = skill + return clone async def run_tasks( self, @@ -132,6 +165,11 @@ async def run_tasks( raise RuntimeError(_MISSING_FABRIC_MSG) from exc resolved_config = config or AgentEvalRunConfig() + # Assign a run id once per run so two runs (e.g. an A/B baseline vs. skilled variant) written + # under the same work_root/output_dir land in distinct, non-colliding evidence trees. Callers + # that set run_id keep their identifier. + if resolved_config.run_id is None: + resolved_config = resolved_config.model_copy(update={"run_id": _new_run_id()}) agent_config = FabricConfig.from_mapping(self._config) # 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. @@ -144,18 +182,72 @@ async def run_tasks( # 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) # ``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() + # Resolve once how a skill would reach this harness (the adapter is constant across tasks) by + # asking Fabric's own capability planner, so any adapter that declares native skills support — ours + # or an end-user's — is picked up automatically instead of via a hardcoded allow-list. Fail fast + # rather than silently run a skill-free trial mislabeled as "with skill", which would corrupt an + # A/B comparison. Only touched when a skill is set, so the no-skill path is unaffected. + skill_mode: str | None = None + if self._skill is not None: + skill_mode = self._resolve_skill_mode(client, agent_config, base_profiles) + if skill_mode is None: + adapter_id = agent_config.harness.adapter_id + raise RuntimeError( + f"FabricAgentRuntime received a skill but adapter {adapter_id!r} has no known " + "skill-injection strategy: Fabric does not route skills to it natively and it is not a " + "codex harness. Use a skills-native or codex harness, or drop the skill." + ) + + semaphore = asyncio.Semaphore(resolved_config.parallelism) + 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 self._run_task( + client, agent_config, base_profiles, index, task, resolved_config, skill_mode + ) return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) + def _resolve_skill_mode( + self, + client: Fabric, + agent_config: FabricConfig, + base_profiles: list[FabricProfileConfig], + ) -> str | None: + """Ask Fabric how a skill would reach the selected harness, or ``None`` if it can't. + + Probes Fabric's capability planner: plan a copy of the config with a sentinel skill path attached + (it need not exist on disk) and read how the adapter routes skills. Querying the authoritative + source at runtime means adapters that declare native skills support — ours or an end-user's — are + detected without a hardcoded list. See :func:`~...skills.resolve_skill_mode`. + """ + probe_config = agent_config.model_copy(deep=True) + probe_config.add_skill_path(_SKILL_PROBE_PATH) + plan = client.plan(probe_config, profiles=base_profiles, base_dir=self._base_dir) + return resolve_skill_mode(capability_plan=plan.capability_plan, harness=plan.adapter.harness) + + def _existing_skill_paths(self) -> list[str]: + """Skill paths the base config/profiles already declare (union, order-preserved). + + Fabric applies profile ``skills.paths`` last-wins, so the native skill overlay has to re-list + these alongside the evaluated skill or the treated arm would silently drop them (see + ``install_skill``). Read from the raw config/profile mappings the runtime was given, so it covers + both config- and profile-declared skills without a Fabric round-trip. + """ + paths: list[str] = [] + for section in (self._config, *self._profiles): + skills = section.get("skills") if isinstance(section, Mapping) else None + declared = skills.get("paths") if isinstance(skills, Mapping) else None + for path in declared or []: + if isinstance(path, str) and path not in paths: + paths.append(path) + return paths + async def _run_task( self, client: Fabric, @@ -164,9 +256,10 @@ async def _run_task( index: int, task: AgentEvalTask, config: AgentEvalRunConfig, + skill_mode: str | None, ) -> 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. + # lookup, not a re-load, so the types are used where they're constructed instead of threaded down. from nemo_fabric import FabricProfileConfig, RunRequest # ty: ignore[unresolved-import] evidence_dir = self._evidence_dir(index, task, config) @@ -180,10 +273,29 @@ async def _run_task( # downloads), so it is offloaded off the shared event loop. workspace_dir = evidence_dir / _WORKSPACE_SUBDIR workspace_dir.mkdir(parents=True, exist_ok=True) + skill_provenance: SkillProvenance | None = None try: # Stage seed files into the workspace for their on-disk side effect; the prompt is the task # instruction only, so the returned paths are unused. await asyncio.to_thread(seed_workspace, workspace_dir, task.inputs.get(SEED_FILES_INPUT_KEY)) + + # Inject the skill (if any) for this task. A native harness gets a per-task ``skills`` profile + # overlay; codex self-injection stages the bundle into the workspace and emits no overlay. + # Provenance is stamped on the trial for the A/B diff. Blocking file I/O, off the event loop. + skill_profiles: list[FabricProfileConfig] = [] + if self._skill is not None and skill_mode is not None: + installation = await asyncio.to_thread( + install_skill, + skill=self._skill, + adapter_id=agent_config.harness.adapter_id, + mode=skill_mode, + workspace_dir=workspace_dir, + skill_stage_dir=(evidence_dir / _SKILL_SUBDIR).resolve(), + existing_skill_paths=self._existing_skill_paths(), + ) + skill_provenance = installation.provenance + skill_profiles = [FabricProfileConfig.from_mapping(p) for p in installation.profiles] + 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. @@ -195,21 +307,35 @@ async def _run_task( # ``Fabric.run`` folds the per-invocation input + request id into a ``RunRequest``. client.run( task_config, - profiles=[*base_profiles, *lock_profiles], + # Caller profiles, then the native skill overlay, then the evaluator lock overlays; + # the lock overlays trail so the per-task workspace/model/artifacts stay authoritative. + profiles=[*base_profiles, *skill_profiles, *lock_profiles], base_dir=self._base_dir, request=RunRequest(input=task.agent_prompt(), request_id=task.id), ), timeout=self._timeout_s, ) except TimeoutError as exc: - return self._failed_trial(task, evidence_dir, exc) + return self._failed_trial(task, evidence_dir, exc, extra_metadata={"skill": skill_provenance}) 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._failed_trial(task, evidence_dir, exc, extra_metadata={"skill": skill_provenance}) - return self._to_trial(task, result, evidence_dir, workspace_dir) + # Codex self-injection staged the bundle *inside* the workspace so the harness could discover it. + # Now that the run is done (and captured in the trajectory), remove it before the workspace is + # exposed as filesystem evidence — otherwise the injected files read as agent output and skew + # workspace-reading metrics (a treated run with no agent-created files would look non-empty). + if skill_mode == SKILL_MODE_CODEX_SKILLS_DIR and skill_provenance is not None: + await asyncio.to_thread(_remove_injected_bundle, workspace_dir, skill_provenance["location"]) + return self._to_trial(task, result, evidence_dir, workspace_dir, skill_provenance=skill_provenance) def _to_trial( - self, task: AgentEvalTask, result: RunResult, evidence_dir: Path, workspace_dir: Path + self, + task: AgentEvalTask, + result: RunResult, + evidence_dir: Path, + workspace_dir: Path, + *, + skill_provenance: SkillProvenance | None = None, ) -> AgentEvalTrial: # Persist the full normalized Fabric result so graders (and debugging) can see the raw # envelope, and expose it as an evidence descriptor. @@ -223,6 +349,8 @@ def _to_trial( "adapter_kind": result.adapter_kind, "invocation_id": result.invocation_id, "agent_model": self._model, + # Skill provenance (name + content hash + injection mode) for the A/B diff; None baseline. + "skill": skill_provenance, } if result.status != "succeeded": @@ -433,9 +561,35 @@ def _evidence_dir(self, index: int, task: AgentEvalTask, config: AgentEvalRunCon root = self._work_root if root is None: root = (config.output_dir or Path.cwd()) / "evidence" / "fabric" + # The run id isolates this run's evidence from other runs sharing the same root (A/B baseline + # vs. skilled); run_tasks always populates it, so the fallback only guards a direct call. + run_id = config.run_id or _new_run_id() 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 + return Path(root) / _safe_path_name(run_id) / task_dir + + +def _remove_injected_bundle(workspace_dir: Path, location: str) -> None: + """Remove the Codex-injected skill subtree from ``workspace_dir`` and prune emptied parents. + + ``location`` is workspace-relative (``.agents/skills/``). Best-effort: the skill was already + captured in the run's trajectory, so SkillUsedMetric (which reads the trace, not the workspace) is + unaffected, and any filesystem error here must not fail an otherwise-successful trial. + """ + workspace_root = workspace_dir.resolve() + injected = (workspace_dir / location).resolve() + # Guard against a location escaping the workspace (defensive; provenance is evaluator-authored). + if workspace_root not in injected.parents or not injected.exists(): + return + shutil.rmtree(injected, ignore_errors=True) + # Prune now-empty reserved parents (``.agents/skills``, ``.agents``) but never the workspace itself. + parent = injected.parent + while parent != workspace_root and parent.is_dir(): + try: + parent.rmdir() # only succeeds while empty + except OSError: + break + parent = parent.parent def _normalize_output(output: RunOutput | JsonValue) -> JsonValue: @@ -477,3 +631,8 @@ def _result_error(result: RunResult) -> Mapping[str, Any]: def _safe_path_name(value: str) -> str: return "".join(char if char.isalnum() or char in "._-" else "-" for char in value).strip(".-")[:120] + + +def _new_run_id() -> str: + timestamp = datetime.now(UTC).strftime("%Y%m%d%H%M%S%f") + return f"fabric-{timestamp}-{uuid4().hex[:8]}" diff --git a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/skills.py b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/skills.py new file mode 100644 index 0000000000..ae834b8d3d --- /dev/null +++ b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/skills.py @@ -0,0 +1,272 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Agent-skill injection for the Fabric agent-eval runtimes (PROTOTYPE). + +An *agent skill* is a directory following the `agentskills.io `_ +spec: a folder named ``/`` containing a required ``SKILL.md`` (YAML frontmatter with ``name`` + +``description``, then instructions) plus optional ``scripts/`` / ``references/`` / ``assets/``. We make +that bundle available to the harness before it runs a task so an A/B eval can score the same taskset +with and without the skill. The skill is a runtime-level knob: build one runtime with ``skill=None`` +and one with ``skill=`` over the same tasks, then diff the scores. + +An :class:`AgentSkill` points at a local skill directory; staging is an OS-level ``copytree`` (file +contents never pass through Python memory). The plugin resolves a platform fileset to a local +directory and constructs an ``AgentSkill`` from it — the SDK has no fileset concept of its own. + +How the skill reaches the harness depends on the selected Fabric adapter, and which mode applies is +decided by *querying Fabric's own capability planner at runtime* (:func:`resolve_skill_mode` over a +``RunPlan.capability_plan``), not a hardcoded adapter list — so it tracks whatever the installed +adapters declare, including end-user adapters we don't ship: + +* **Native** (:data:`SKILL_MODE_NATIVE`): the adapter advertises ``accepts: ["skills", ...]`` (the + Hermes/Claude adapters do), so Fabric's planner routes skills to ``harness_native``. We stage the + bundle into an isolated ``/`` dir and hand Fabric a ``skills.paths`` profile overlay; the + adapter loads it (Hermes → harness ``skills.external_dirs``). +* **Codex skills dir** (:data:`SKILL_MODE_CODEX_SKILLS_DIR`): the Fabric ``codex`` adapter only + ``accepts: ["models"]`` (planner routes skills ``unsupported``), but the Codex CLI itself discovers + agentskills bundles from ``.agents/skills/`` in its working directory. So we place the bundle at + ``/.agents/skills//`` and let Codex discover it — same discoverable-skill semantics + as native (cross-harness A/B is apples-to-apples), no Fabric adapter change needed. + +If an adapter neither routes skills natively nor is a Codex harness, :func:`resolve_skill_mode` returns +``None`` and the runtime fails fast rather than silently running a skill-free trial. +""" + +from __future__ import annotations + +import hashlib +import re +import shutil +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import TypedDict + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +#: Required entry document of an agentskills bundle. +PRIMARY_SKILL_DOC = "SKILL.md" +#: Directory Codex scans (relative to its working dir) for agentskills bundles. +CODEX_SKILLS_DIR = ".agents/skills" +#: Name of the Fabric profile overlay that carries the native ``skills`` config. +SKILL_PROFILE_NAME = "eval_skill" + +#: Skill reaches the harness via the native Fabric ``skills`` config (adapter accepts it). +SKILL_MODE_NATIVE = "native" +#: Skill is placed under ``/.agents/skills//`` for Codex to discover. +SKILL_MODE_CODEX_SKILLS_DIR = "codex_skills_dir" + +# agentskills.io name rule: 1-64 chars, lowercase alphanumeric + single interior hyphens. +_SKILL_NAME_RE = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$") +_MAX_NAME_LEN = 64 + +# Fabric capability-planner vocabulary (``RunPlan.capability_plan['routes']`` entries). A ``skills`` +# route with target ``harness_native`` means the selected adapter declared native skills support; the +# runtime plans a probe skill path and reads these to decide the injection mode (see resolve_skill_mode). +_SKILLS_ROUTE_KIND = "skills" +_SKILLS_TARGET_NATIVE = "harness_native" +# Fabric harness name of the Codex CLI adapter, which self-discovers ``.agents/skills/`` rather than +# accepting the native ``skills`` config. +_CODEX_HARNESS = "codex" + + +class SkillInjectionError(ValueError): + """A skill could not be resolved, staged, or wired into the selected harness. + + Subclasses ``ValueError`` so the runtime's per-task error handling still catches it and fails + only that task. + """ + + +class AgentSkill(BaseModel): + """An agentskills.io bundle (a local directory) to make available to the agent before a task. + + ``name`` must satisfy the agentskills naming rule and is used as the staged bundle's directory name + (spec: the name matches the directory name). ``directory`` is the local skill directory, which must + contain a top-level ``SKILL.md``. + """ + + model_config = ConfigDict(extra="forbid") + + name: str = Field(description="agentskills skill name; also the bundle directory name and provenance id.") + directory: Path = Field(description="Local agentskills bundle directory (a SKILL.md at its root).") + + @field_validator("name") + @classmethod + def _valid_name(cls, value: str) -> str: + if len(value) > _MAX_NAME_LEN or not _SKILL_NAME_RE.match(value): + raise ValueError( + f"skill name {value!r} must be 1-{_MAX_NAME_LEN} chars, lowercase alphanumeric with " + "single interior hyphens (agentskills.io naming rule)" + ) + return value + + @classmethod + def from_directory(cls, directory: str | Path, *, name: str | None = None) -> AgentSkill: + """Build a skill from an on-disk agentskills bundle. ``name`` defaults to the directory basename.""" + root = Path(directory).expanduser().resolve() + if not (root / PRIMARY_SKILL_DOC).is_file(): + raise SkillInjectionError(f"skill directory {str(directory)!r} has no {PRIMARY_SKILL_DOC}") + return cls(name=name or root.name, directory=root) + + +class SkillProvenance(TypedDict): + """Which skill was injected into a trial and how; stamped into trial metadata for the A/B diff. + + A plain (JSON-serializable) dict so it drops straight into trial metadata. ``None`` in that slot + means the baseline (no skill). + """ + + name: str #: The skill's agentskills name. + hash: str #: sha256 over the staged bundle — attributes a score delta to an exact skill version. + mode: str #: How it was injected (:data:`SKILL_MODE_NATIVE` / :data:`SKILL_MODE_CODEX_SKILLS_DIR`). + adapter_id: str #: The harness adapter the skill was wired into. + location: str #: Where the bundle was staged (absolute for native, workspace-relative for codex). + + +@dataclass +class SkillInstallation: + """Result of installing a skill for one task. + + ``profiles`` are Fabric profile-overlay mappings the runtime appends to its profile stack (the + native branch emits one ``skills`` overlay; the Codex branch emits none because placement in the + workspace is the delivery mechanism). ``provenance`` is stamped into trial metadata so the A/B + comparison is auditable. + """ + + profiles: list[dict[str, object]] + provenance: SkillProvenance + + +def native_skills_route(capability_plan: Mapping[str, object]) -> bool: + """Whether Fabric's capability planner routed skills to the harness natively. + + ``capability_plan`` is the ``RunPlan.capability_plan`` mapping from ``Fabric.plan(...)`` planned with + a skill path attached; its ``routes`` record each capability decision. A ``skills`` route with target + ``harness_native`` means the selected adapter declares ``accepts: ["skills", ...]`` and Fabric hands + the bundle to the harness itself. Any other outcome (``unsupported``, or no skills route) is False. + """ + routes = capability_plan.get("routes") + if not isinstance(routes, list): + return False + return any( + isinstance(route, Mapping) + and route.get("kind") == _SKILLS_ROUTE_KIND + and route.get("target") == _SKILLS_TARGET_NATIVE + for route in routes + ) + + +def resolve_skill_mode(*, capability_plan: Mapping[str, object], harness: str) -> str | None: + """Resolve how a skill would reach the selected harness, or ``None`` if it can't. + + Driven by Fabric's own capability routing (queried at runtime via ``Fabric.plan``) rather than a + hardcoded adapter list, so it tracks whatever the installed adapters declare — including end-user + adapters we don't ship: + + * skills route natively (:func:`native_skills_route`) -> :data:`SKILL_MODE_NATIVE`; + * else a Codex harness (self-discovers ``.agents/skills/``) -> :data:`SKILL_MODE_CODEX_SKILLS_DIR`; + * else ``None`` -> the runtime fails fast rather than run a skill-free trial labeled "with skill". + """ + if native_skills_route(capability_plan): + return SKILL_MODE_NATIVE + if harness.strip().lower() == _CODEX_HARNESS: + return SKILL_MODE_CODEX_SKILLS_DIR + return None + + +def install_skill( + *, + skill: AgentSkill, + adapter_id: str, + mode: str, + workspace_dir: Path, + skill_stage_dir: Path, + existing_skill_paths: Sequence[str] = (), +) -> SkillInstallation: + """Stage ``skill`` as a ``/`` bundle and wire it into the harness per ``mode``. + + Blocking file I/O — call via ``asyncio.to_thread`` from the async runtime. The bundle is always + namespaced under ``/`` so it never collides with task-seeded workspace-root files; the content + hash is computed over the staged bytes so provenance tracks the actual skill content. + + ``existing_skill_paths`` are the skill paths the base config/profiles already declare. Fabric applies + profile ``skills.paths`` last-wins, so the native overlay must re-list them alongside the evaluated + skill — otherwise the treated arm would silently drop every preconfigured skill and the A/B would + differ by more than the injected skill. + """ + if mode == SKILL_MODE_NATIVE: + skill_root = skill_stage_dir / skill.name + _stage_bundle(skill.directory, skill_root, reserved=False) + # Preserve the pre-existing skill paths (order-preserved, de-duplicated) and append the + # evaluated skill, so the last-wins overlay reproduces the baseline skill set plus this one. + paths = list(dict.fromkeys([*existing_skill_paths, str(skill_root)])) + overlay: dict[str, object] = { + "name": SKILL_PROFILE_NAME, + "description": "Make the evaluation skill available via the native Fabric skills config.", + "skills": {"paths": paths}, + } + return SkillInstallation( + profiles=[overlay], + provenance=_provenance(skill, _hash_directory(skill_root), mode, adapter_id, str(skill_root)), + ) + + if mode == SKILL_MODE_CODEX_SKILLS_DIR: + skill_root = workspace_dir / CODEX_SKILLS_DIR / skill.name + _stage_bundle(skill.directory, skill_root, reserved=True) + location = (Path(CODEX_SKILLS_DIR) / skill.name).as_posix() + return SkillInstallation( + profiles=[], + provenance=_provenance(skill, _hash_directory(skill_root), mode, adapter_id, location), + ) + + raise SkillInjectionError(f"unknown skill injection mode {mode!r} for adapter {adapter_id!r}") + + +def _stage_bundle(directory: Path, skill_root: Path, *, reserved: bool) -> None: + """Stage the skill ``directory`` as an *exact* copy at ``skill_root`` (the ``/`` bundle dir). + + The staged bundle must reflect exactly the supplied directory, so provenance and behaviour track the + real content. ``reserved`` picks the collision policy for the destination: + + * ``reserved=False`` — the evaluator-owned native stage dir: recreate it, so a reused run id can't + leave a file that was since removed from the source bundle surviving in the stage. + * ``reserved=True`` — the Codex workspace path (``.agents/skills/``): refuse to clobber + pre-existing content there, since it can only be a task-seeded file colliding with the reserved + skill path. + """ + src = directory.expanduser() + if not (src / PRIMARY_SKILL_DOC).is_file(): + raise SkillInjectionError(f"skill directory {str(directory)!r} has no {PRIMARY_SKILL_DOC}") + if skill_root.exists(): + if reserved: + raise SkillInjectionError( + f"cannot stage skill into reserved path {str(skill_root)!r}: it already exists " + "(a task-seeded file collides with the injected skill bundle)" + ) + shutil.rmtree(skill_root) # evaluator-owned: recreate so the stage is an exact copy + skill_root.parent.mkdir(parents=True, exist_ok=True) + # OS-level copy — file contents never pass through Python memory. + shutil.copytree(src, skill_root) + + +def _provenance(skill: AgentSkill, skill_hash: str, mode: str, adapter_id: str, location: str) -> SkillProvenance: + return { + "name": skill.name, + "hash": skill_hash, + "mode": mode, + "adapter_id": adapter_id, + "location": location, + } + + +def _hash_directory(directory: Path) -> str: + """Stable sha256 over a directory's file tree (sorted relpath + contents).""" + digest = hashlib.sha256() + for path in sorted(path for path in directory.rglob("*") if path.is_file()): + digest.update(path.relative_to(directory).as_posix().encode("utf-8")) + digest.update(b"\0") + digest.update(path.read_bytes()) + digest.update(b"\0") + return digest.hexdigest()