diff --git a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/container_runtime.py b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/container_runtime.py index 61d6e8d301..92cb5f0274 100644 --- a/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/container_runtime.py +++ b/packages/nemo_evaluator_sdk/src/nemo_evaluator_sdk/agent_eval/runtimes/fabric/container_runtime.py @@ -29,16 +29,29 @@ from __future__ import annotations import asyncio +import copy import json import logging import shlex +import shutil import tempfile from collections.abc import Mapping, Sequence -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING, Any, cast from nemo_evaluator_sdk.agent_eval.runtimes.fabric import _common from nemo_evaluator_sdk.agent_eval.runtimes.fabric.image import ensure_fabric_image +from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import ( + CODEX_SKILLS_DIR, + SKILL_MODE_CODEX_SKILLS_DIR, + AgentSkill, + SkillInjectionError, + SkillMode, + SkillProvenance, + SkillSet, + resolve_skill_mode, + stage_skills_seed, +) from nemo_evaluator_sdk.agent_eval.runtimes.sandbox.api import AsyncSandbox from nemo_evaluator_sdk.agent_eval.runtimes.sandbox.base import SandboxExecResult, SandboxProvider, SandboxSpec from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask @@ -67,6 +80,11 @@ # AgentEvalTask); until then it is an internal default rather than a runtime-construction knob. DEFAULT_FABRIC_TIMEOUT_S = 600 _RUNTIME_NAME = "fabric_container" +_MISSING_FABRIC_MSG = ( + "FabricContainerRuntime skill injection requires the `nemo-fabric` package (native NeMo Fabric SDK) " + "on the host to resolve how a skill reaches the selected adapter; the container otherwise runs Fabric " + "only inside the sandbox." +) # Fixed in-container layout. The runtime seeds ``/in`` (agent config, profiles, input), execs Fabric's # CLI, and reads the produced ``/out`` subtree back across the boundary. @@ -81,6 +99,13 @@ _AGENT_PATH = f"{_IN_DIR}/agent.yaml" _INPUT_PATH = f"{_IN_DIR}/input.txt" _WORKSPACE_PROFILE_NAME = "eval_workspace" +# In-sandbox root for a natively-injected skill bundle. It lives under ``/in`` (not ``/out``), so it is +# never part of the downloaded ``/out`` evidence — only codex-mode skills, which must sit in the workspace +# for the harness to self-discover them, need post-download cleanup. +_SKILLS_DIR = f"{_IN_DIR}/skills" +# Sentinel skill path attached only to probe Fabric's capability planner for the selected adapter's skills +# routing (mirrors the host runtime). Never staged and need not exist on disk. +_SKILL_PROBE_PATH = "nemo-eval-skill-capability-probe" class FabricContainerRuntime: @@ -94,6 +119,7 @@ def __init__( profiles: Sequence[FabricProfileConfig | Mapping[str, Any]] = (), secrets: Mapping[str, SecretRef] = {}, image: str | None = None, + skills: Sequence[AgentSkill] | None = None, ) -> None: # The Fabric agent is fully described by its ``FabricConfig`` (harness + model + runtime); it is # consumed structurally as a mapping to cross the sandbox boundary as JSON. @@ -109,6 +135,34 @@ def __init__( # Optional prebuilt image: the trial runs inside it, so it must contain the Fabric CLI + adapter. # None -> stock harness-agnostic image built on first run. self._image: str | None = image + # Optional agent skills injected per task (A/B: baseline vs. treated via ``with_skills``). How they + # reach the harness is resolved once per run (the adapter is constant across the taskset) in + # ``run_tasks``; only touched when a skill is set, so the no-skill path stays dependency-free. Names + # must be unique — each stages to its own ``/`` bundle, so a repeat would collide. + self._skill_set = SkillSet(tuple(skills or ())) + + def with_skills(self, skills: Sequence[AgentSkill]) -> FabricContainerRuntime: + """Return a copy of this runtime with ``skills`` *added* to its skill set; ``self`` is not modified. + + Mirrors :meth:`FabricAgentRuntime.with_skills`: additive and chainable + (``rt.with_skills([a]).with_skills([b])`` injects both), so an A/B eval derives a treated runtime + from a skill-free baseline (``baseline.with_skills(the_skills)``) and the arms differ in exactly the + injected skills. Names must be unique across the combined set (colliding ``/`` bundles), so + re-adding a present skill raises. A shallow copy suffices — the shared fields are immutable + config/paths/provider. (``run_tasks`` disposes the injected provider on completion, so an A/B run + over two arms should give each arm its own provider.) + """ + clone = copy.copy(self) + clone._skill_set = self._skill_set.with_skills(skills) + return clone + + def with_skill(self, skill: AgentSkill) -> FabricContainerRuntime: + """Return a copy of this runtime with ``skill`` *added*; ``self`` is not modified. + + Thin wrapper over :meth:`with_skills` for the single-skill case; equally chainable + (``rt.with_skill(a).with_skill(b)`` injects both). + """ + return self.with_skills([skill]) async def resolve_secrets(self, secret_resolver: SecretResolver) -> None: """Resolve declared ``SecretRef``\\ s to values, keyed by the env var each harness reads. @@ -134,10 +188,10 @@ async def run_tasks( resolved_config = config or AgentEvalRunConfig() semaphore = asyncio.Semaphore(resolved_config.parallelism) - async def run_one(index: int, task: AgentEvalTask) -> AgentEvalTrial: + async def run_one(index: int, task: AgentEvalTask, skill_mode: SkillMode | None) -> AgentEvalTrial: async with semaphore: logger.info("running task", extra={"index": index + 1, "task_id": task.id}) - result = await self._run_task(index, task, resolved_config) + result = await self._run_task(index, task, resolved_config, skill_mode) logger.info("task completed", extra={"index": index + 1, "task_id": task.id}) return result @@ -150,13 +204,26 @@ async def run_one(index: int, task: AgentEvalTask) -> AgentEvalTrial: if self._secrets and not self._secrets_resolved: # No orchestrator resolved our secrets (standalone run) — fall back to local env resolution. await self.resolve_secrets(LocalSecretResolver()) - return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) + # Resolve once (the adapter is constant across the taskset) how a skill reaches this harness, by + # probing Fabric's capability planner — the same authoritative routing the host runtime uses. + # Fail fast rather than silently run a skill-free trial mislabeled "with skill". Blocking pyo3 + # planning, so keep it off the shared event loop; only reached when a skill is set. + skill_mode = await asyncio.to_thread(self._resolve_skill_mode) if self._skill_set.skills else None + if self._skill_set.skills and skill_mode is None: + raise RuntimeError( + f"FabricContainerRuntime received one or more skills but adapter {self._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 skills." + ) + return await asyncio.gather(*(run_one(index, task, skill_mode) for index, task in enumerate(tasks))) finally: # Each sandbox tears itself down; the provider is shared across the batch, so its # process-wide resources are disposed once here, when the batch completes. await self._provider.aclose() - async def _run_task(self, index: int, task: AgentEvalTask, config: AgentEvalRunConfig) -> AgentEvalTrial: + async def _run_task( + self, index: int, task: AgentEvalTask, config: AgentEvalRunConfig, skill_mode: SkillMode | None + ) -> AgentEvalTrial: evidence_dir = self._evidence_dir(index, task, config) out_dir = evidence_dir / "out" evidence_dir.mkdir(parents=True, exist_ok=True) @@ -164,8 +231,9 @@ async def _run_task(self, index: int, task: AgentEvalTask, config: AgentEvalRunC # The whole per-task flow — framing input, seeding, exec, download, and parsing the result — is # guarded so any failure (bad seed, sandbox crash, unreadable result) fails only this task's # trial rather than aborting the gathered batch. + skill_provenances: list[SkillProvenance] = [] try: - seed_files, profile_paths = self._seed_files(task) + seed_files, profile_paths, skill_provenances = self._seed_files(task, skill_mode) spec = SandboxSpec( image=self._image, workdir=_WORKSPACE_DIR, env=dict(self._resolved_env), files=seed_files ) @@ -174,10 +242,61 @@ async def _run_task(self, index: int, task: AgentEvalTask, config: AgentEvalRunC await self._seed_workspace(sandbox, task) result = await sandbox.exec(self._fabric_command(profile_paths), timeout_s=DEFAULT_FABRIC_TIMEOUT_S) await sandbox.download_dir(_OUT_DIR, out_dir) - return self._to_trial(task, out_dir, evidence_dir, result) + # Codex self-injection seeds each bundle inside the workspace so the harness discovers it during + # the run; drop them from the downloaded evidence before the workspace is exposed (else the + # injected files read as agent output to workspace-reading metrics). Native staging lives under + # /in, which is never downloaded, so it never pollutes the evidence. + if skill_mode == SKILL_MODE_CODEX_SKILLS_DIR: + for provenance in skill_provenances: + await asyncio.to_thread(_remove_injected_bundle, out_dir / "workspace", provenance["location"]) + return self._to_trial(task, out_dir, evidence_dir, result, skill_provenances=skill_provenances) except Exception as exc: # noqa: BLE001 - a task failure must not abort the whole run - # Stamp runtime + image even on failures before _to_trial (startup/seeding/download). - return self._failed_trial(task, evidence_dir, exc, extra_metadata=self._base_metadata()) + # Stamp runtime + image + skills even on failures before _to_trial (startup/seeding/download). + return self._failed_trial( + task, evidence_dir, exc, extra_metadata={**self._base_metadata(), **_skill_metadata(skill_provenances)} + ) + + def _resolve_skill_mode(self) -> SkillMode | None: + """Ask Fabric how a skill would reach the selected harness, or ``None`` if it can't. + + Mirrors :meth:`FabricAgentRuntime._resolve_skill_mode`: 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 from the + capability plan. Querying the authoritative planner at runtime means any adapter that declares + native skills support — ours or an end-user's — is picked up without a hardcoded list. ``nemo_fabric`` + is imported lazily on the host (only when a skill is set), so the no-skill path never needs it. + """ + try: + from nemo_fabric import Fabric, FabricConfig, FabricProfileConfig # ty: ignore[unresolved-import] + except ImportError as exc: + raise RuntimeError(_MISSING_FABRIC_MSG) from exc + agent_config = FabricConfig.from_mapping(self._config) + base_profiles = [FabricProfileConfig.from_mapping(profile) for profile in self._profiles] + probe_config = agent_config.model_copy(deep=True) + probe_config.add_skill_path(_SKILL_PROBE_PATH) + plan = Fabric().plan(probe_config, profiles=base_profiles) + return resolve_skill_mode(capability_plan=plan.capability_plan, harness=plan.adapter.harness) + + def _adapter_id(self) -> str: + """The harness adapter id declared by the config mapping (for provenance + error messages).""" + harness = self._config.get("harness") + adapter_id = harness.get("adapter_id") if isinstance(harness, Mapping) else None + return str(adapter_id) if adapter_id is not None else "" + + 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 overlay has to re-list these + alongside the evaluated skill or the treated arm would silently drop preconfigured skills (see + ``stage_skills_seed``). Read from the raw config/profile mappings the runtime was given. + """ + 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 def _fabric_command(self, profile_paths: Sequence[str]) -> str: """The ``fabric run`` invocation: pre-create the /out dirs Fabric chdirs into, run, capture stdout.""" @@ -188,21 +307,45 @@ def _fabric_command(self, profile_paths: Sequence[str]) -> str: f"{run} > {shlex.quote(_RESULT_PATH)} 2> {shlex.quote(_FABRIC_STDERR)}" ) - def _seed_files(self, task: AgentEvalTask) -> tuple[dict[str, str], list[str]]: - """Return (files to seed into /in, profile paths for --profile). Configs are written as JSON, - which the Fabric CLI parses as YAML. Base profiles are followed by the per-task workspace overlay - and the trajectory profile (built as plain dicts — no host nemo_relay dependency).""" + def _seed_files( + self, task: AgentEvalTask, skill_mode: SkillMode | None + ) -> tuple[dict[str, str], list[str], list[SkillProvenance]]: + """Return (files to seed into the sandbox, profile paths for --profile, skill provenances). + + Configs are written as JSON, which the Fabric CLI parses as YAML. When skills are injected each + bundle is rendered into the seed set at the harness's in-sandbox discovery path (native: + ``/in/skills/``; codex: ``/.agents/skills/``), with at most ONE merged native + overlay listing every bundle. Profiles are ordered caller-first, then the native skill overlay (if + any), then the per-task workspace + trajectory overlays — which trail so the evaluator-owned + workspace/artifacts stay authoritative (mirroring the host runtime's overlay ordering). + """ files: dict[str, str] = { _AGENT_PATH: json.dumps(self._config), _INPUT_PATH: task.agent_prompt(), } + skill_profiles: list[dict[str, Any]] = [] + provenances: list[SkillProvenance] = [] + if self._skill_set.skills and skill_mode is not None: + if skill_mode == SKILL_MODE_CODEX_SKILLS_DIR: + _check_codex_skill_collision(self._skill_set.skills, task.inputs.get(SEED_FILES_INPUT_KEY) or {}) + seed = stage_skills_seed( + skills=self._skill_set.skills, + adapter_id=self._adapter_id(), + mode=skill_mode, + workspace_dir=_WORKSPACE_DIR, + skills_dir=_SKILLS_DIR, + existing_skill_paths=self._existing_skill_paths(), + ) + files.update(seed.files) + skill_profiles = seed.profiles + provenances = seed.provenances profile_paths: list[str] = [] - profiles = [*self._profiles, self._workspace_profile(), self._trajectory_profile()] + profiles = [*self._profiles, *skill_profiles, self._workspace_profile(), self._trajectory_profile()] for index, profile in enumerate(profiles): path = f"{_IN_DIR}/profile-{index}.yaml" files[path] = json.dumps(profile) profile_paths.append(path) - return files, profile_paths + return files, profile_paths, provenances @staticmethod def _workspace_profile() -> dict[str, Any]: @@ -241,9 +384,17 @@ def _base_metadata(self) -> dict[str, object]: return {"runtime": _RUNTIME_NAME, "image": self._image, "sandbox_provider": self._provider.name} def _to_trial( - self, task: AgentEvalTask, out_dir: Path, evidence_dir: Path, result: SandboxExecResult + self, + task: AgentEvalTask, + out_dir: Path, + evidence_dir: Path, + result: SandboxExecResult, + *, + skill_provenances: list[SkillProvenance] | None = None, ) -> AgentEvalTrial: - base_metadata = self._base_metadata() + # Skill provenance (name + content hash + injection mode) rides on every trial for the A/B diff: + # a ``skills`` list plus the historical lone ``skill`` field, matching the host FabricAgentRuntime. + base_metadata = {**self._base_metadata(), **_skill_metadata(skill_provenances or [])} # Gate on the exec outcome first: a timed-out or non-zero ``fabric run`` is untrustworthy even # when a stale/partial fabric_result.json is left behind (the shell ``>`` redirect truncates the @@ -342,6 +493,67 @@ def _to_mapping(config: FabricConfig | Mapping[str, Any]) -> dict[str, Any]: return dict(cast(Mapping[str, Any], source)) +def _skill_metadata(provenances: list[SkillProvenance]) -> dict[str, object]: + """Trial-metadata fields describing the injected skill set (the A/B provenance). + + ``skills`` is the full list of injected-skill provenances (empty = baseline). ``skill`` keeps the + historical single-provenance field — the lone provenance for a one-skill run, else ``None`` — so + single-skill consumers (e.g. ``SkillUsedMetric``) and existing trials/tests keep working unchanged. + Mirrors ``FabricAgentRuntime._skill_metadata``. + """ + return {"skill": provenances[0] if len(provenances) == 1 else None, "skills": provenances} + + +def _check_codex_skill_collision(skills: Sequence[AgentSkill], task_files: Mapping[str, object]) -> None: + """Raise if a task seed file targets the same bundle dir as a runtime-injected codex skill. + + ``.agents/skills/`` holds skills from two independent, equally valid sources: the runtime + ``skills`` parameter (the A/B knob — staged into the workspace before the sandbox starts) and + the task's own ``files`` inputs (skills the task definition always ships — uploaded after it + starts). Tasks are free to seed their own skills there; only writing the *same* + ``.agents/skills//`` from both sources is a conflict, since the task upload lands second + and would overwrite the injected bundle, leaving the stamped provenance hash describing content + the agent never saw. Fail that case rather than emit a silently mislabeled A/B trial. + """ + for skill in skills: + injected_bundle = PurePosixPath(CODEX_SKILLS_DIR) / skill.name + for rel_path in task_files: + seed = PurePosixPath(rel_path) + if seed == injected_bundle or injected_bundle in seed.parents: + raise SkillInjectionError( + f"task seed file {str(rel_path)!r} writes into {str(injected_bundle)!r}, which is " + f"also injected as the runtime skill {skill.name!r}; the task upload would overwrite " + "the injected bundle. Inject this skill via the runtime ``skills`` parameter or ship " + "it in the task's files, not both" + ) + + +def _remove_injected_bundle(workspace_dir: Path, location: str) -> None: + """Remove the Codex-injected skill subtree from a downloaded ``workspace`` dir and prune emptied parents. + + ``location`` is workspace-relative (``.agents/skills/``). Best-effort and mirrors the host + runtime's cleanup: 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. + """ + if not workspace_dir.is_dir(): + return + 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 _find_atif(relay_dir: Path) -> Path | None: # Relay nests the trajectory under a per-run subdir (relay/runtime-/trajectory-*.atif.json), # so search recursively rather than only relay's direct children. 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 47d31788b0..eb194b93e8 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 @@ -40,9 +40,10 @@ from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import ( SKILL_MODE_CODEX_SKILLS_DIR, AgentSkill, + SkillMode, SkillProvenance, + SkillSet, install_skills, - require_unique_skill_names, resolve_skill_mode, ) from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask @@ -139,8 +140,7 @@ def __init__( self._timeout_s = timeout_s self._capture_trajectory = capture_trajectory self._runtime_name = runtime_name - self._skills = list(skills or []) - require_unique_skill_names(self._skills) + self._skill_set = SkillSet(tuple(skills or ())) def with_skills(self, skills: Sequence[AgentSkill]) -> FabricAgentRuntime: """Return a copy of this runtime with ``skills`` *added* to its skill set; ``self`` is not modified. @@ -151,10 +151,8 @@ def with_skills(self, skills: Sequence[AgentSkill]) -> FabricAgentRuntime: set — two bundles claiming the same ``/`` would collide — so re-adding a skill already present raises. A shallow copy suffices — the shared fields are immutable config/paths. """ - combined = [*self._skills, *skills] - require_unique_skill_names(combined) clone = copy.copy(self) - clone._skills = combined + clone._skill_set = self._skill_set.with_skills(skills) return clone def with_skill(self, skill: AgentSkill) -> FabricAgentRuntime: @@ -205,8 +203,8 @@ async def run_tasks( # 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._skills: + skill_mode: SkillMode | None = None + if self._skill_set.skills: skill_mode = self._resolve_skill_mode(client, agent_config, base_profiles) if skill_mode is None: adapter_id = agent_config.harness.adapter_id @@ -231,7 +229,7 @@ def _resolve_skill_mode( client: Fabric, agent_config: FabricConfig, base_profiles: list[FabricProfileConfig], - ) -> str | None: + ) -> SkillMode | 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 @@ -269,7 +267,7 @@ async def _run_task( index: int, task: AgentEvalTask, config: AgentEvalRunConfig, - skill_mode: str | None, + skill_mode: SkillMode | None, ) -> AgentEvalTrial: # nemo_fabric is already imported+validated in ``run_tasks``; this is a cached sys.modules # lookup, not a re-load, so the types are used where they're constructed instead of threaded down. @@ -297,10 +295,10 @@ async def _run_task( # workspace and emits no overlay. One provenance per skill is stamped on the trial for the A/B # diff. Blocking file I/O, off the event loop. skill_profiles: list[FabricProfileConfig] = [] - if self._skills and skill_mode is not None: + if self._skill_set.skills and skill_mode is not None: installation = await asyncio.to_thread( install_skills, - skills=self._skills, + skills=self._skill_set.skills, adapter_id=agent_config.harness.adapter_id, mode=skill_mode, workspace_dir=workspace_dir, 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 index 975db81b68..81d9bb6f0f 100644 --- 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 @@ -41,7 +41,7 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass from pathlib import Path -from typing import TypedDict +from typing import Literal, TypedDict from pydantic import BaseModel, ConfigDict, Field, field_validator @@ -52,10 +52,15 @@ #: Name of the Fabric profile overlay that carries the native ``skills`` config. SKILL_PROFILE_NAME = "eval_skill" +#: How an injected skill reaches the selected harness (resolved from Fabric's capability plan). The two +#: runtimes thread this value from :func:`resolve_skill_mode` down to :func:`install_skill` / +#: :func:`stage_skills_seed`, so a mistyped mode is a type error rather than a silent no-op. +SkillMode = Literal["native", "codex_skills_dir"] + #: Skill reaches the harness via the native Fabric ``skills`` config (adapter accepts it). -SKILL_MODE_NATIVE = "native" +SKILL_MODE_NATIVE: SkillMode = "native" #: Skill is placed under ``/.agents/skills//`` for Codex to discover. -SKILL_MODE_CODEX_SKILLS_DIR = "codex_skills_dir" +SKILL_MODE_CODEX_SKILLS_DIR: SkillMode = "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]+)*$") @@ -120,7 +125,7 @@ class SkillProvenance(TypedDict): 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`). + mode: SkillMode #: 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). @@ -158,7 +163,7 @@ def native_skills_route(capability_plan: Mapping[str, object]) -> bool: ) -def resolve_skill_mode(*, capability_plan: Mapping[str, object], harness: str) -> str | None: +def resolve_skill_mode(*, capability_plan: Mapping[str, object], harness: str) -> SkillMode | 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 @@ -180,7 +185,7 @@ def install_skill( *, skill: AgentSkill, adapter_id: str, - mode: str, + mode: SkillMode, workspace_dir: Path, skill_stage_dir: Path, existing_skill_paths: Sequence[str] = (), @@ -259,11 +264,35 @@ def require_unique_skill_names(skills: Sequence[AgentSkill]) -> None: ) +@dataclass(frozen=True) +class SkillSet: + """Immutable, name-validated collection of :class:`AgentSkill`\\s shared by both Fabric runtimes. + + Centralizes the uniqueness check and clone-on-mutation pattern that + :class:`~...FabricAgentRuntime` and :class:`~...FabricContainerRuntime` would otherwise + duplicate: construction validates that skill names are unique; :meth:`with_skills` and + :meth:`with_skill` each return a new ``SkillSet`` without modifying ``self``. + """ + + skills: tuple[AgentSkill, ...] = () + + def __post_init__(self) -> None: + require_unique_skill_names(self.skills) + + def with_skills(self, skills: Sequence[AgentSkill]) -> SkillSet: + """Return a new ``SkillSet`` with ``skills`` appended; ``self`` is not modified.""" + return SkillSet((*self.skills, *skills)) + + def with_skill(self, skill: AgentSkill) -> SkillSet: + """Return a new ``SkillSet`` with ``skill`` appended; ``self`` is not modified.""" + return self.with_skills([skill]) + + def install_skills( *, skills: Sequence[AgentSkill], adapter_id: str, - mode: str, + mode: SkillMode, workspace_dir: Path, skill_stage_dir: Path, existing_skill_paths: Sequence[str] = (), @@ -287,6 +316,14 @@ def install_skills( staged_roots: list[Path] = [] try: for skill in skills: + # Register the target BEFORE staging: install_skill can raise after it has already written + # files (a copytree failing partway, an unreadable file while hashing), and a root recorded + # only on success would leave that partial bundle behind. A target that already exists is + # never registered — in codex mode that is a task-seeded file install_skill refuses to + # clobber, and rolling it back would delete task input this call did not create. + stage_root = _skill_stage_root(skill, mode, workspace_dir, skill_stage_dir) + if not stage_root.exists(): + staged_roots.append(stage_root) provenance = install_skill( skill=skill, adapter_id=adapter_id, @@ -296,9 +333,6 @@ def install_skills( existing_skill_paths=existing_skill_paths, ).provenance provenances.append(provenance) - # A native provenance ``location`` is the absolute staged root; a codex one is - # workspace-relative (``.agents/skills/``). Record it only after a successful stage. - staged_roots.append(_provenance_stage_root(provenance, workspace_dir)) except Exception: for root in staged_roots: shutil.rmtree(root, ignore_errors=True) @@ -319,13 +353,129 @@ def install_skills( return SkillsInstallation(profiles=profiles, provenances=provenances) -def _provenance_stage_root(provenance: SkillProvenance, workspace_dir: Path) -> Path: - """Absolute on-disk root of a staged bundle, for rollback. Native ``location`` is already absolute; - codex ``location`` is workspace-relative (``.agents/skills/``).""" - location = provenance["location"] - if provenance["mode"] == SKILL_MODE_CODEX_SKILLS_DIR: - return workspace_dir / location - return Path(location) +def _skill_stage_root(skill: AgentSkill, mode: SkillMode, workspace_dir: Path, skill_stage_dir: Path) -> Path: + """Absolute on-disk root :func:`install_skill` would stage ``skill`` into, computed before staging. + + Mirrors install_skill's per-mode placement so :func:`install_skills` can register a rollback target + up front (an unknown mode raises there, not here; the returned path is simply never created, and + rolling back a path that does not exist is a no-op).""" + if mode == SKILL_MODE_NATIVE: + return skill_stage_dir / skill.name + return workspace_dir / CODEX_SKILLS_DIR / skill.name + + +def _render_skill_seed( + *, skill: AgentSkill, adapter_id: str, mode: SkillMode, workspace_dir: str, skills_dir: str +) -> tuple[dict[str, str], SkillProvenance]: + """Render one skill bundle into an in-sandbox ``{path: text}`` seed map + its provenance. + + The per-skill core of :func:`stage_skills_seed` (the containerized counterpart of :func:`install_skill`, + which ``copytree``\\ s onto host disk): the container has no host workspace, so the bundle is read into + memory as UTF-8 text and keyed at the harness's in-sandbox discovery path — native: ``/ + /``; codex: ``/.agents/skills//``. The content hash is over the source bundle + (matching :func:`install_skill`). The caller merges these into one seed set and, for native mode, a + single ``skills`` overlay — so no per-skill overlay is emitted here. + """ + bundle = _read_text_bundle(skill.directory) + skill_hash = _hash_directory(skill.directory) + if mode == SKILL_MODE_NATIVE: + skill_root = f"{skills_dir.rstrip('/')}/{skill.name}" + files = {f"{skill_root}/{rel}": text for rel, text in bundle.items()} + return files, _provenance(skill, skill_hash, mode, adapter_id, skill_root) + + if mode == SKILL_MODE_CODEX_SKILLS_DIR: + skill_root = f"{workspace_dir.rstrip('/')}/{CODEX_SKILLS_DIR}/{skill.name}" + files = {f"{skill_root}/{rel}": text for rel, text in bundle.items()} + location = f"{CODEX_SKILLS_DIR}/{skill.name}" + return files, _provenance(skill, skill_hash, mode, adapter_id, location) + + raise SkillInjectionError(f"unknown skill injection mode {mode!r} for adapter {adapter_id!r}") + + +def _read_text_bundle(directory: Path) -> dict[str, str]: + """Read an agentskills bundle into a ``{posix_relpath: text}`` map (requires a top-level ``SKILL.md``). + + Every file is decoded as UTF-8: the containerized seed set (``SandboxSpec.files``) is text-only, so a + binary file (e.g. an image under ``assets/``) raises here rather than silently corrupting the staged + bundle — the host :func:`install_skill` path (OS-level ``copytree``) handles binary bundles instead. + """ + src = directory.expanduser() + if not (src / PRIMARY_SKILL_DOC).is_file(): + raise SkillInjectionError(f"skill directory {str(directory)!r} has no {PRIMARY_SKILL_DOC}") + bundle: dict[str, str] = {} + for path in sorted(candidate for candidate in src.rglob("*") if candidate.is_file()): + rel = path.relative_to(src).as_posix() + try: + bundle[rel] = path.read_text(encoding="utf-8") + except UnicodeDecodeError as exc: + raise SkillInjectionError( + f"skill file {rel!r} is not UTF-8 text; containerized skill injection (via the sandbox " + "seed set) supports text bundles only" + ) from exc + return bundle + + +@dataclass +class SkillsSeed: + """Result of rendering several skills into one sandbox seed set (see :func:`stage_skills_seed`). + + The plural, containerized sibling of :class:`SkillsInstallation`: + + * ``files`` — the merged ``{absolute_in_sandbox_path: text}`` seed map for every staged bundle. + * ``profiles`` — at most ONE merged native ``skills`` overlay listing every bundle (Fabric applies + ``skills.paths`` last-wins, so all must ride in a single overlay or all but the last are dropped); + the codex branch emits none. + * ``provenances`` — one entry per skill, in the given order, for the multi-skill A/B trial metadata. + """ + + files: dict[str, str] + profiles: list[dict[str, object]] + provenances: list[SkillProvenance] + + +def stage_skills_seed( + *, + skills: Sequence[AgentSkill], + adapter_id: str, + mode: SkillMode, + workspace_dir: str, + skills_dir: str, + existing_skill_paths: Sequence[str] = (), +) -> SkillsSeed: + """Render every skill in ``skills`` into one sandbox seed set + overlays for the container runtime. + + The plural, containerized sibling of :func:`install_skills`: renders each bundle (via + :func:`_render_skill_seed`) under its own ``/`` at the harness's in-sandbox discovery path, then, + for the native mode, merges the per-skill paths into a SINGLE ``skills`` overlay (Fabric applies profile + ``skills.paths`` last-wins, so one overlay per skill would silently drop all but the last). Pre-existing + ``existing_skill_paths`` are preserved ahead of the injected skills (same reason). Skill names must be + unique — their ``/`` bundles would otherwise collide. No on-disk rollback is needed (unlike + :func:`install_skills`): the seed set is an in-memory map, so a failure to render any skill just + discards the accumulated map and raises, leaving nothing staged. + """ + require_unique_skill_names(skills) + files: dict[str, str] = {} + provenances: list[SkillProvenance] = [] + for skill in skills: + rendered, provenance = _render_skill_seed( + skill=skill, adapter_id=adapter_id, mode=mode, workspace_dir=workspace_dir, skills_dir=skills_dir + ) + files.update(rendered) + provenances.append(provenance) + + profiles: list[dict[str, object]] = [] + if mode == SKILL_MODE_NATIVE and provenances: + # One merged overlay: pre-existing skills first, then each staged bundle (a native provenance's + # ``location`` is its absolute in-sandbox skill root), order-preserved and de-duplicated. + paths = list(dict.fromkeys([*existing_skill_paths, *(prov["location"] for prov in provenances)])) + profiles = [ + { + "name": SKILL_PROFILE_NAME, + "description": "Make the evaluation skills available via the native Fabric skills config.", + "skills": {"paths": paths}, + } + ] + return SkillsSeed(files=files, profiles=profiles, provenances=provenances) def _stage_bundle(directory: Path, skill_root: Path, *, reserved: bool) -> None: @@ -355,7 +505,7 @@ def _stage_bundle(directory: Path, skill_root: Path, *, reserved: bool) -> None: shutil.copytree(src, skill_root) -def _provenance(skill: AgentSkill, skill_hash: str, mode: str, adapter_id: str, location: str) -> SkillProvenance: +def _provenance(skill: AgentSkill, skill_hash: str, mode: SkillMode, adapter_id: str, location: str) -> SkillProvenance: return { "name": skill.name, "hash": skill_hash, diff --git a/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_container_runtime.py b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_container_runtime.py index a20d0e31d6..5778c45b40 100644 --- a/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_container_runtime.py +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_container_runtime.py @@ -11,6 +11,8 @@ from __future__ import annotations import json +import sys +import types from collections.abc import Sequence from pathlib import Path @@ -326,6 +328,380 @@ def test_trajectory_profile_built_from_relay_types() -> None: assert cfg["atof"]["filename"] == crt._common.ATOF_FILENAME +# -------------------------------------------------------------------------------------------------- +# Agent-skill injection (containerized) — mirrors the host-runtime skill tests in test_fabric_runtime.py. +# -------------------------------------------------------------------------------------------------- + +# Adapters the fake planner reports as accepting the native Fabric ``skills`` config. ``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", "acme.custom.native"} +_KNOWN_HARNESSES = ("hermes", "codex", "claude") +_CODEX_CONFIG = {"metadata": {"name": "eval"}, "harness": {"adapter_id": "nvidia.fabric.codex.cli"}} + + +def _harness_name(adapter_id: str) -> str: + return next((harness for harness in _KNOWN_HARNESSES if harness in adapter_id), "custom") + + +class _FakeHarness: + def __init__(self, adapter_id: str) -> None: + self.adapter_id = adapter_id + + +class _FakeConfig: + """Minimal stand-in for nemo_fabric.FabricConfig — only what ``_resolve_skill_mode`` touches.""" + + def __init__(self, mapping: dict[str, object]) -> None: + self.mapping = mapping + harness = mapping.get("harness", {}) + self.harness = _FakeHarness(harness.get("adapter_id", "") if isinstance(harness, dict) else "") + self.skill_paths: list[str] = [] + + @classmethod + def from_mapping(cls, mapping: dict[str, object]) -> _FakeConfig: + return cls(mapping) + + def model_copy(self, *, deep: bool = False) -> _FakeConfig: + clone = _FakeConfig(self.mapping) + clone.skill_paths = list(self.skill_paths) + return clone + + def add_skill_path(self, path: object) -> None: + self.skill_paths.append(str(path)) + + +class _FakeProfile: + def __init__(self, mapping: dict[str, object]) -> None: + self.mapping = mapping + self.name = mapping.get("name") + + @classmethod + def from_mapping(cls, mapping: dict[str, object]) -> _FakeProfile: + return cls(mapping) + + +class _FakeAdapterInfo: + def __init__(self, harness: str) -> None: + self.harness = harness + + +class _FakePlan: + def __init__(self, *, capability_plan: dict[str, object], harness: str) -> None: + self.capability_plan = capability_plan + self.adapter = _FakeAdapterInfo(harness) + + +class _FakeFabric: + planned: list[dict[str, object]] = [] + + def plan(self, agent: object, *, profiles: object = None, base_dir: object = None) -> _FakePlan: + # Mirror Fabric's planner: a ``skills`` route appears only when a skill path is attached, and it + # routes ``harness_native`` iff the selected adapter accepts native skills. + _FakeFabric.planned.append({"agent": agent, "profiles": profiles}) + 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)) + + +def _install_fake_fabric(monkeypatch: pytest.MonkeyPatch) -> type[_FakeFabric]: + """Inject a fake ``nemo_fabric`` module (the runtime imports it lazily only to plan skills routing).""" + _FakeFabric.planned = [] + module = types.ModuleType("nemo_fabric") + module.Fabric = _FakeFabric # type: ignore[attr-defined] + module.FabricConfig = _FakeConfig # type: ignore[attr-defined] + module.FabricProfileConfig = _FakeProfile # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "nemo_fabric", module) + return _FakeFabric + + +def _skill_bundle(base: Path, *, name: str = "code-review", extra: dict[str, str] | None = None) -> Path: + """Write a minimal agentskills bundle under ``base//`` and return its path.""" + root = base / name + root.mkdir(parents=True, exist_ok=True) + (root / "SKILL.md").write_text(f"---\nname: {name}\ndescription: d\n---\n\nBe thorough.\n", 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 _seeded_profiles(provider: _FakeProvider) -> dict[str, dict[str, object]]: + """The profile overlays the runtime seeded into /in, keyed by their ``name``.""" + profiles: dict[str, dict[str, object]] = {} + for key, value in provider.seeded.items(): + if key.startswith("/in/profile-"): + profile = json.loads(value) + profiles[profile["name"]] = profile + return profiles + + +async def test_native_skill_seeds_bundle_into_seed_set_with_overlay( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + fabric = _install_fake_fabric(monkeypatch) + skill = AgentSkill.from_directory(_skill_bundle(tmp_path / "src", extra={"references/r.md": "material"})) + provider = _FakeProvider() # module _CONFIG is the hermes.sdk adapter -> native routing + (trial,) = await _run(_runtime(provider, skills=[skill]), [_task()], tmp_path) + + assert trial.status == AgentEvalTrialStatus.COMPLETED + # The mode is resolved by probing Fabric's capability planner (with a probe skill path attached). + assert fabric.planned and fabric.planned[0]["agent"].skill_paths + # The bundle is rendered INTO the sandbox seed set at the native in-/in discovery path (not /out, so it + # never lands in the downloaded workspace evidence). + assert provider.seeded["/in/skills/code-review/SKILL.md"].startswith("---") + assert provider.seeded["/in/skills/code-review/references/r.md"] == "material" + # A native `skills` overlay points at the staged bundle dir; the eval workspace/trajectory overlays trail. + overlay = _seeded_profiles(provider)["eval_skill"] + assert overlay["skills"]["paths"][-1] == "/in/skills/code-review" + # Provenance is stamped into trial metadata for the A/B diff. + prov = trial.metadata["skill"] + assert prov["name"] == "code-review" and prov["mode"] == "native" and prov["hash"] + assert prov["location"] == "/in/skills/code-review" + + +async def test_native_skill_preserves_preconfigured_skill_paths( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Fabric applies profile skills.paths last-wins, so the overlay must re-list config- and profile-declared + # skills (order-preserved) ahead of the evaluated skill, or the treated arm would drop them. + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + _install_fake_fabric(monkeypatch) + config = {**_CONFIG, "skills": {"paths": ["/pre/existing-a"]}} + skill = AgentSkill.from_directory(_skill_bundle(tmp_path / "src")) + provider = _FakeProvider() + runtime = FabricContainerRuntime( + config, # type: ignore[arg-type] + provider=provider, + profiles=[{"name": "caller", "skills": {"paths": ["/pre/existing-b"]}}], + skills=[skill], + ) + await runtime.run_tasks([_task()], AgentEvalRunConfig(output_dir=tmp_path)) + + paths = _seeded_profiles(provider)["eval_skill"]["skills"]["paths"] + assert paths[:2] == ["/pre/existing-a", "/pre/existing-b"] + assert paths[-1] == "/in/skills/code-review" + + +async def test_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. + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + _install_fake_fabric(monkeypatch) + custom = {"metadata": {"name": "eval"}, "harness": {"adapter_id": "acme.custom.native"}} + skill = AgentSkill.from_directory(_skill_bundle(tmp_path / "src")) + provider = _FakeProvider() + runtime = FabricContainerRuntime(custom, provider=provider, skills=[skill]) # type: ignore[arg-type] + (trial,) = await runtime.run_tasks([_task()], AgentEvalRunConfig(output_dir=tmp_path)) + + assert "eval_skill" in _seeded_profiles(provider) + assert trial.metadata["skill"]["mode"] == "native" + + +async def test_codex_skill_seeds_workspace_and_is_excluded_from_evidence( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + class _CodexWorkspaceProvider(_FakeProvider): + # Simulate the codex-seeded bundle landing in the workspace that gets downloaded as /out evidence. + async def download_dir(self, handle: SandboxHandle, source_dir: str, target_dir: Path) -> None: + await super().download_dir(handle, source_dir, target_dir) + skill_md = target_dir / "workspace" / ".agents" / "skills" / "code-review" / "SKILL.md" + skill_md.parent.mkdir(parents=True, exist_ok=True) + skill_md.write_text("---\nname: code-review\n---\n", encoding="utf-8") + + _install_fake_fabric(monkeypatch) + skill = AgentSkill.from_directory(_skill_bundle(tmp_path / "src")) + provider = _CodexWorkspaceProvider() + runtime = FabricContainerRuntime(_CODEX_CONFIG, provider=provider, skills=[skill]) # type: ignore[arg-type] + (trial,) = await runtime.run_tasks([_task()], AgentEvalRunConfig(output_dir=tmp_path)) + + # Codex discovers agentskills from .agents/skills/ in its working dir, so the bundle is seeded there in + # the workspace (not /in), for the harness to self-discover during the run. + assert provider.seeded["/out/workspace/.agents/skills/code-review/SKILL.md"].startswith("---") + # No native overlay: placement in the workspace is the delivery mechanism. + assert "eval_skill" not in _seeded_profiles(provider) + prov = trial.metadata["skill"] + assert prov["mode"] == "codex_skills_dir" + assert prov["location"] == ".agents/skills/code-review" + # ...then removed from the downloaded evidence (with its emptied .agents parents) before the workspace is + # exposed, so the injected files don't read as agent output to workspace-reading metrics. + workspace = Path(trial.evidence.require("workspace").ref) # type: ignore[arg-type] + assert not (workspace / ".agents").exists() + + +async def test_skill_on_unsupported_adapter_fails_fast(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + _install_fake_fabric(monkeypatch) + unsupported = {"metadata": {"name": "eval"}, "harness": {"adapter_id": "some.other.adapter"}} + skill = AgentSkill.from_directory(_skill_bundle(tmp_path / "src", name="s")) + runtime = FabricContainerRuntime(unsupported, provider=_FakeProvider(), skills=[skill]) # type: ignore[arg-type] + + with pytest.raises(RuntimeError, match="no known skill-injection strategy"): + await runtime.run_tasks([_task()], AgentEvalRunConfig(output_dir=tmp_path)) + + +async def test_no_skill_leaves_metadata_none_and_skips_planner(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + fabric = _install_fake_fabric(monkeypatch) + provider = _FakeProvider() + (trial,) = await _run(_runtime(provider), [_task()], tmp_path) + + assert trial.metadata["skill"] is None and trial.metadata["skills"] == [] + # No skill -> no planner probe (the no-skill path must not import nemo_fabric or pay for a plan()). + assert fabric.planned == [] + # And nothing is seeded under a skills discovery path. + assert not any("/skills/" in key or "/.agents/" in key for key in provider.seeded) + + +async def test_multiple_native_skills_each_staged_with_one_merged_overlay( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # A set of skills: each bundle stages under its own /in/skills//, and all ride in ONE merged + # `eval_skill` overlay (Fabric applies skills.paths last-wins, so a per-skill overlay would drop all + # but the last). Trial metadata carries one provenance per skill; the lone `skill` field is None. + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + _install_fake_fabric(monkeypatch) + skills = [ + AgentSkill.from_directory(_skill_bundle(tmp_path / "a", name="docx")), + AgentSkill.from_directory(_skill_bundle(tmp_path / "b", name="pptx")), + ] + provider = _FakeProvider() # hermes.sdk -> native + (trial,) = await _run(_runtime(provider, skills=skills), [_task()], tmp_path) + + assert provider.seeded["/in/skills/docx/SKILL.md"].startswith("---") + assert provider.seeded["/in/skills/pptx/SKILL.md"].startswith("---") + # Exactly one merged overlay listing both bundle roots, in order. + overlay = _seeded_profiles(provider)["eval_skill"] + assert overlay["skills"]["paths"] == ["/in/skills/docx", "/in/skills/pptx"] + # One provenance per skill; the historical lone `skill` field is None for a multi-skill run. + names = [prov["name"] for prov in trial.metadata["skills"]] + assert names == ["docx", "pptx"] + assert trial.metadata["skill"] is None + + +async def test_multiple_codex_skills_all_removed_from_evidence(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + class _CodexWorkspaceProvider(_FakeProvider): + # Simulate every codex-seeded bundle landing in the downloaded /out workspace. + async def download_dir(self, handle: SandboxHandle, source_dir: str, target_dir: Path) -> None: + await super().download_dir(handle, source_dir, target_dir) + for name in ("docx", "pptx"): + md = target_dir / "workspace" / ".agents" / "skills" / name / "SKILL.md" + md.parent.mkdir(parents=True, exist_ok=True) + md.write_text("---\n---\n", encoding="utf-8") + + _install_fake_fabric(monkeypatch) + skills = [ + AgentSkill.from_directory(_skill_bundle(tmp_path / "a", name="docx")), + AgentSkill.from_directory(_skill_bundle(tmp_path / "b", name="pptx")), + ] + provider = _CodexWorkspaceProvider() + runtime = FabricContainerRuntime(_CODEX_CONFIG, provider=provider, skills=skills) # type: ignore[arg-type] + (trial,) = await runtime.run_tasks([_task()], AgentEvalRunConfig(output_dir=tmp_path)) + + # Both bundles seeded under the codex discovery dir, no overlay, and every one scrubbed from evidence. + assert provider.seeded["/out/workspace/.agents/skills/docx/SKILL.md"].startswith("---") + assert provider.seeded["/out/workspace/.agents/skills/pptx/SKILL.md"].startswith("---") + assert "eval_skill" not in _seeded_profiles(provider) + assert [prov["name"] for prov in trial.metadata["skills"]] == ["docx", "pptx"] + workspace = Path(trial.evidence.require("workspace").ref) # type: ignore[arg-type] + assert not (workspace / ".agents").exists() + + +def test_duplicate_skill_names_rejected_at_construction_and_with_skills() -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill, SkillInjectionError + + a = AgentSkill(name="dup", directory=Path("/skills/a")) + b = AgentSkill(name="dup", directory=Path("/skills/b")) + # Two bundles claiming the same / would collide — rejected up front, before any task runs. + with pytest.raises(SkillInjectionError, match="duplicate skill name"): + _runtime(_FakeProvider(), skills=[a, b]) + with pytest.raises(SkillInjectionError, match="duplicate skill name"): + _runtime(_FakeProvider(), skills=[a]).with_skills([b]) + + +def test_with_skills_is_additive_and_independent() -> None: + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + base = _runtime(_FakeProvider()) + a = AgentSkill(name="docx", directory=Path("/skills/docx")) + b = AgentSkill(name="pptx", directory=Path("/skills/pptx")) + + # with_skill is a thin, additive, chainable wrapper over with_skills; the original is untouched. + chained = base.with_skill(a).with_skill(b) + assert chained is not base + assert base._skill_set.skills == () + assert chained._skill_set.skills == (a, b) + assert base.with_skills([a, b])._skill_set.skills == (a, b) + + +async def test_same_skill_from_both_injection_and_task_files_fails_task( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Seeding a skill via task files is legitimate; doing it for a skill the runtime ALSO injects is + # not. The task upload lands after the pre-start seed, so it would overwrite the injected bundle + # and leave the stamped provenance hash describing content the agent never saw. + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + _install_fake_fabric(monkeypatch) + skill = AgentSkill.from_directory(_skill_bundle(tmp_path / "src")) + provider = _FakeProvider() + runtime = FabricContainerRuntime(_CODEX_CONFIG, provider=provider, skills=[skill]) # type: ignore[arg-type] + task = AgentEvalTask( + id="collision", + intent="...", + inputs={ + "instruction": "Do something.", + "files": {".agents/skills/code-review/SKILL.md": "# override"}, + }, + ) + (trial,) = await runtime.run_tasks([task], AgentEvalRunConfig(output_dir=tmp_path)) + + assert trial.status == AgentEvalTrialStatus.FAILED + error = json.loads(Path(trial.evidence.require("error").ref).read_text()) # type: ignore[arg-type] + assert error["error_type"] == "SkillInjectionError" + assert "also injected as the runtime skill 'code-review'" in error["error"] + + +async def test_task_seeded_skill_coexists_with_a_different_injected_skill( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Both mechanisms may populate .agents/skills/ in the same run: the A/B-injected skill and a + # skill the task definition always ships as a file input. They only conflict when they target + # the same / bundle, so different names go through their separate paths untouched. + from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import AgentSkill + + _install_fake_fabric(monkeypatch) + skill = AgentSkill.from_directory(_skill_bundle(tmp_path / "src", name="code-review")) + provider = _FakeProvider() + runtime = FabricContainerRuntime(_CODEX_CONFIG, provider=provider, skills=[skill]) # type: ignore[arg-type] + task = AgentEvalTask( + id="coexist", + intent="...", + inputs={ + "instruction": "Do something.", + # A skill the task always ships as a file input (different name, not A/B-injected). + "files": {".agents/skills/style-guide/SKILL.md": "---\nname: style-guide\n---\n"}, + }, + ) + (trial,) = await runtime.run_tasks([task], AgentEvalRunConfig(output_dir=tmp_path)) + + assert trial.status == AgentEvalTrialStatus.COMPLETED + assert "/out/workspace/.agents/skills/code-review/SKILL.md" in provider.seeded + assert any(target == "/out/workspace" for _, target in provider.uploaded_dirs) + + async def test_sandbox_exception_is_isolated_per_task(tmp_path: Path) -> None: # A sandbox that blows up mid-run must yield a FAILED trial per task, not abort the gather. class _BrokenProvider(_FakeProvider): 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 34b77e94fb..f51d674bb7 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 @@ -819,8 +819,8 @@ def test_with_skill_returns_independent_copy() -> None: # A new instance is returned; the original is untouched. assert treated is not base - assert base._skills == [] - assert treated._skills == [skill] + assert base._skill_set.skills == () + assert treated._skill_set.skills == (skill,) def test_with_skills_returns_independent_copy_of_the_set() -> None: @@ -836,8 +836,8 @@ def test_with_skills_returns_independent_copy_of_the_set() -> None: # A new instance carries the added set; the original is untouched. assert treated is not base - assert base._skills == [] - assert treated._skills == skills + assert base._skill_set.skills == () + assert treated._skill_set.skills == tuple(skills) def test_with_skill_is_additive_and_chainable() -> None: @@ -851,12 +851,12 @@ def test_with_skill_is_additive_and_chainable() -> None: chained = base.with_skill(a).with_skill(b) # Both skills are present, in order; each intermediate runtime is left untouched. - assert chained._skills == [a, b] - assert base._skills == [] - assert base.with_skill(a)._skills == [a] + assert chained._skill_set.skills == (a, b) + assert base._skill_set.skills == () + assert base.with_skill(a)._skill_set.skills == (a,) # with_skills extends the same way (equivalent to chaining the single-skill calls). - assert base.with_skills([a, b])._skills == [a, b] - assert base.with_skill(a).with_skills([b])._skills == [a, b] + assert base.with_skills([a, b])._skill_set.skills == (a, b) + assert base.with_skill(a).with_skills([b])._skill_set.skills == (a, b) def test_with_skills_rejects_duplicate_names() -> None: 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 index 6e0353cda0..23e449f6d0 100644 --- a/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_skills.py +++ b/packages/nemo_evaluator_sdk/tests/agent_eval/test_fabric_skills.py @@ -8,6 +8,7 @@ from pathlib import Path import pytest +from nemo_evaluator_sdk.agent_eval.runtimes.fabric import skills as skills_module from nemo_evaluator_sdk.agent_eval.runtimes.fabric.skills import ( CODEX_SKILLS_DIR, SKILL_MODE_CODEX_SKILLS_DIR, @@ -289,6 +290,39 @@ def test_install_skills_rolls_back_staged_bundles_on_failure(tmp_path: Path) -> assert not (stage_dir / "good").exists() +def test_install_skills_rolls_back_the_bundle_that_failed_mid_stage( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # install_skill can raise AFTER writing files (a copytree failing partway, an unreadable file while + # hashing). That skill's own partially staged bundle must be rolled back too, not just the bundles + # from earlier iterations. Hashing runs once the bundle is fully copied, so failing it there puts + # real files on disk before the error. + good = AgentSkill.from_directory(_make_bundle(tmp_path / "src", name="good")) + late = AgentSkill.from_directory(_make_bundle(tmp_path / "src2", name="late")) + stage_dir = tmp_path / "stage" + + real_hash = skills_module._hash_directory + + def _fail_on_late(directory: Path) -> str: + if directory.name == "late": + raise OSError("unreadable file in bundle") + return real_hash(directory) + + monkeypatch.setattr(skills_module, "_hash_directory", _fail_on_late) + + with pytest.raises(OSError): + install_skills( + skills=[good, late], + adapter_id="nvidia.fabric.hermes.sdk", + mode=SKILL_MODE_NATIVE, + workspace_dir=tmp_path / "ws", + skill_stage_dir=stage_dir, + ) + + assert not (stage_dir / "good").exists() # earlier bundle rolled back, as before + assert not (stage_dir / "late").exists() # ...and so is the one that failed after staging files + + def test_install_skills_rollback_never_deletes_preexisting_seed_file(tmp_path: Path) -> None: # Codex reserved-path collision: the second skill's target already holds a task-seeded file. Rollback # must remove only the bundle THIS call staged (the first skill), never the pre-existing seed file. diff --git a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/container_runtime.py b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/container_runtime.py index b89c2b69f5..f93c043f51 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/container_runtime.py +++ b/sdk/python/nemo-platform/src/nemo_platform/beta/evaluator/agent_eval/runtimes/fabric/container_runtime.py @@ -29,16 +29,29 @@ from __future__ import annotations import asyncio +import copy import json import logging import shlex +import shutil import tempfile from collections.abc import Mapping, Sequence -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING, Any, cast from nemo_platform.beta.evaluator.agent_eval.runtimes.fabric import _common from nemo_platform.beta.evaluator.agent_eval.runtimes.fabric.image import ensure_fabric_image +from nemo_platform.beta.evaluator.agent_eval.runtimes.fabric.skills import ( + CODEX_SKILLS_DIR, + SKILL_MODE_CODEX_SKILLS_DIR, + AgentSkill, + SkillInjectionError, + SkillMode, + SkillProvenance, + SkillSet, + resolve_skill_mode, + stage_skills_seed, +) from nemo_platform.beta.evaluator.agent_eval.runtimes.sandbox.api import AsyncSandbox from nemo_platform.beta.evaluator.agent_eval.runtimes.sandbox.base import SandboxExecResult, SandboxProvider, SandboxSpec from nemo_platform.beta.evaluator.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask @@ -67,6 +80,11 @@ # AgentEvalTask); until then it is an internal default rather than a runtime-construction knob. DEFAULT_FABRIC_TIMEOUT_S = 600 _RUNTIME_NAME = "fabric_container" +_MISSING_FABRIC_MSG = ( + "FabricContainerRuntime skill injection requires the `nemo-fabric` package (native NeMo Fabric SDK) " + "on the host to resolve how a skill reaches the selected adapter; the container otherwise runs Fabric " + "only inside the sandbox." +) # Fixed in-container layout. The runtime seeds ``/in`` (agent config, profiles, input), execs Fabric's # CLI, and reads the produced ``/out`` subtree back across the boundary. @@ -81,6 +99,13 @@ _AGENT_PATH = f"{_IN_DIR}/agent.yaml" _INPUT_PATH = f"{_IN_DIR}/input.txt" _WORKSPACE_PROFILE_NAME = "eval_workspace" +# In-sandbox root for a natively-injected skill bundle. It lives under ``/in`` (not ``/out``), so it is +# never part of the downloaded ``/out`` evidence — only codex-mode skills, which must sit in the workspace +# for the harness to self-discover them, need post-download cleanup. +_SKILLS_DIR = f"{_IN_DIR}/skills" +# Sentinel skill path attached only to probe Fabric's capability planner for the selected adapter's skills +# routing (mirrors the host runtime). Never staged and need not exist on disk. +_SKILL_PROBE_PATH = "nemo-eval-skill-capability-probe" class FabricContainerRuntime: @@ -94,6 +119,7 @@ def __init__( profiles: Sequence[FabricProfileConfig | Mapping[str, Any]] = (), secrets: Mapping[str, SecretRef] = {}, image: str | None = None, + skills: Sequence[AgentSkill] | None = None, ) -> None: # The Fabric agent is fully described by its ``FabricConfig`` (harness + model + runtime); it is # consumed structurally as a mapping to cross the sandbox boundary as JSON. @@ -109,6 +135,34 @@ def __init__( # Optional prebuilt image: the trial runs inside it, so it must contain the Fabric CLI + adapter. # None -> stock harness-agnostic image built on first run. self._image: str | None = image + # Optional agent skills injected per task (A/B: baseline vs. treated via ``with_skills``). How they + # reach the harness is resolved once per run (the adapter is constant across the taskset) in + # ``run_tasks``; only touched when a skill is set, so the no-skill path stays dependency-free. Names + # must be unique — each stages to its own ``/`` bundle, so a repeat would collide. + self._skill_set = SkillSet(tuple(skills or ())) + + def with_skills(self, skills: Sequence[AgentSkill]) -> FabricContainerRuntime: + """Return a copy of this runtime with ``skills`` *added* to its skill set; ``self`` is not modified. + + Mirrors :meth:`FabricAgentRuntime.with_skills`: additive and chainable + (``rt.with_skills([a]).with_skills([b])`` injects both), so an A/B eval derives a treated runtime + from a skill-free baseline (``baseline.with_skills(the_skills)``) and the arms differ in exactly the + injected skills. Names must be unique across the combined set (colliding ``/`` bundles), so + re-adding a present skill raises. A shallow copy suffices — the shared fields are immutable + config/paths/provider. (``run_tasks`` disposes the injected provider on completion, so an A/B run + over two arms should give each arm its own provider.) + """ + clone = copy.copy(self) + clone._skill_set = self._skill_set.with_skills(skills) + return clone + + def with_skill(self, skill: AgentSkill) -> FabricContainerRuntime: + """Return a copy of this runtime with ``skill`` *added*; ``self`` is not modified. + + Thin wrapper over :meth:`with_skills` for the single-skill case; equally chainable + (``rt.with_skill(a).with_skill(b)`` injects both). + """ + return self.with_skills([skill]) async def resolve_secrets(self, secret_resolver: SecretResolver) -> None: """Resolve declared ``SecretRef``\\ s to values, keyed by the env var each harness reads. @@ -134,10 +188,10 @@ async def run_tasks( resolved_config = config or AgentEvalRunConfig() semaphore = asyncio.Semaphore(resolved_config.parallelism) - async def run_one(index: int, task: AgentEvalTask) -> AgentEvalTrial: + async def run_one(index: int, task: AgentEvalTask, skill_mode: SkillMode | None) -> AgentEvalTrial: async with semaphore: logger.info("running task", extra={"index": index + 1, "task_id": task.id}) - result = await self._run_task(index, task, resolved_config) + result = await self._run_task(index, task, resolved_config, skill_mode) logger.info("task completed", extra={"index": index + 1, "task_id": task.id}) return result @@ -150,13 +204,26 @@ async def run_one(index: int, task: AgentEvalTask) -> AgentEvalTrial: if self._secrets and not self._secrets_resolved: # No orchestrator resolved our secrets (standalone run) — fall back to local env resolution. await self.resolve_secrets(LocalSecretResolver()) - return await asyncio.gather(*(run_one(index, task) for index, task in enumerate(tasks))) + # Resolve once (the adapter is constant across the taskset) how a skill reaches this harness, by + # probing Fabric's capability planner — the same authoritative routing the host runtime uses. + # Fail fast rather than silently run a skill-free trial mislabeled "with skill". Blocking pyo3 + # planning, so keep it off the shared event loop; only reached when a skill is set. + skill_mode = await asyncio.to_thread(self._resolve_skill_mode) if self._skill_set.skills else None + if self._skill_set.skills and skill_mode is None: + raise RuntimeError( + f"FabricContainerRuntime received one or more skills but adapter {self._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 skills." + ) + return await asyncio.gather(*(run_one(index, task, skill_mode) for index, task in enumerate(tasks))) finally: # Each sandbox tears itself down; the provider is shared across the batch, so its # process-wide resources are disposed once here, when the batch completes. await self._provider.aclose() - async def _run_task(self, index: int, task: AgentEvalTask, config: AgentEvalRunConfig) -> AgentEvalTrial: + async def _run_task( + self, index: int, task: AgentEvalTask, config: AgentEvalRunConfig, skill_mode: SkillMode | None + ) -> AgentEvalTrial: evidence_dir = self._evidence_dir(index, task, config) out_dir = evidence_dir / "out" evidence_dir.mkdir(parents=True, exist_ok=True) @@ -164,8 +231,9 @@ async def _run_task(self, index: int, task: AgentEvalTask, config: AgentEvalRunC # The whole per-task flow — framing input, seeding, exec, download, and parsing the result — is # guarded so any failure (bad seed, sandbox crash, unreadable result) fails only this task's # trial rather than aborting the gathered batch. + skill_provenances: list[SkillProvenance] = [] try: - seed_files, profile_paths = self._seed_files(task) + seed_files, profile_paths, skill_provenances = self._seed_files(task, skill_mode) spec = SandboxSpec( image=self._image, workdir=_WORKSPACE_DIR, env=dict(self._resolved_env), files=seed_files ) @@ -174,10 +242,61 @@ async def _run_task(self, index: int, task: AgentEvalTask, config: AgentEvalRunC await self._seed_workspace(sandbox, task) result = await sandbox.exec(self._fabric_command(profile_paths), timeout_s=DEFAULT_FABRIC_TIMEOUT_S) await sandbox.download_dir(_OUT_DIR, out_dir) - return self._to_trial(task, out_dir, evidence_dir, result) + # Codex self-injection seeds each bundle inside the workspace so the harness discovers it during + # the run; drop them from the downloaded evidence before the workspace is exposed (else the + # injected files read as agent output to workspace-reading metrics). Native staging lives under + # /in, which is never downloaded, so it never pollutes the evidence. + if skill_mode == SKILL_MODE_CODEX_SKILLS_DIR: + for provenance in skill_provenances: + await asyncio.to_thread(_remove_injected_bundle, out_dir / "workspace", provenance["location"]) + return self._to_trial(task, out_dir, evidence_dir, result, skill_provenances=skill_provenances) except Exception as exc: # noqa: BLE001 - a task failure must not abort the whole run - # Stamp runtime + image even on failures before _to_trial (startup/seeding/download). - return self._failed_trial(task, evidence_dir, exc, extra_metadata=self._base_metadata()) + # Stamp runtime + image + skills even on failures before _to_trial (startup/seeding/download). + return self._failed_trial( + task, evidence_dir, exc, extra_metadata={**self._base_metadata(), **_skill_metadata(skill_provenances)} + ) + + def _resolve_skill_mode(self) -> SkillMode | None: + """Ask Fabric how a skill would reach the selected harness, or ``None`` if it can't. + + Mirrors :meth:`FabricAgentRuntime._resolve_skill_mode`: 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 from the + capability plan. Querying the authoritative planner at runtime means any adapter that declares + native skills support — ours or an end-user's — is picked up without a hardcoded list. ``nemo_fabric`` + is imported lazily on the host (only when a skill is set), so the no-skill path never needs it. + """ + try: + from nemo_fabric import Fabric, FabricConfig, FabricProfileConfig # ty: ignore[unresolved-import] + except ImportError as exc: + raise RuntimeError(_MISSING_FABRIC_MSG) from exc + agent_config = FabricConfig.from_mapping(self._config) + base_profiles = [FabricProfileConfig.from_mapping(profile) for profile in self._profiles] + probe_config = agent_config.model_copy(deep=True) + probe_config.add_skill_path(_SKILL_PROBE_PATH) + plan = Fabric().plan(probe_config, profiles=base_profiles) + return resolve_skill_mode(capability_plan=plan.capability_plan, harness=plan.adapter.harness) + + def _adapter_id(self) -> str: + """The harness adapter id declared by the config mapping (for provenance + error messages).""" + harness = self._config.get("harness") + adapter_id = harness.get("adapter_id") if isinstance(harness, Mapping) else None + return str(adapter_id) if adapter_id is not None else "" + + 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 overlay has to re-list these + alongside the evaluated skill or the treated arm would silently drop preconfigured skills (see + ``stage_skills_seed``). Read from the raw config/profile mappings the runtime was given. + """ + 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 def _fabric_command(self, profile_paths: Sequence[str]) -> str: """The ``fabric run`` invocation: pre-create the /out dirs Fabric chdirs into, run, capture stdout.""" @@ -188,21 +307,45 @@ def _fabric_command(self, profile_paths: Sequence[str]) -> str: f"{run} > {shlex.quote(_RESULT_PATH)} 2> {shlex.quote(_FABRIC_STDERR)}" ) - def _seed_files(self, task: AgentEvalTask) -> tuple[dict[str, str], list[str]]: - """Return (files to seed into /in, profile paths for --profile). Configs are written as JSON, - which the Fabric CLI parses as YAML. Base profiles are followed by the per-task workspace overlay - and the trajectory profile (built as plain dicts — no host nemo_relay dependency).""" + def _seed_files( + self, task: AgentEvalTask, skill_mode: SkillMode | None + ) -> tuple[dict[str, str], list[str], list[SkillProvenance]]: + """Return (files to seed into the sandbox, profile paths for --profile, skill provenances). + + Configs are written as JSON, which the Fabric CLI parses as YAML. When skills are injected each + bundle is rendered into the seed set at the harness's in-sandbox discovery path (native: + ``/in/skills/``; codex: ``/.agents/skills/``), with at most ONE merged native + overlay listing every bundle. Profiles are ordered caller-first, then the native skill overlay (if + any), then the per-task workspace + trajectory overlays — which trail so the evaluator-owned + workspace/artifacts stay authoritative (mirroring the host runtime's overlay ordering). + """ files: dict[str, str] = { _AGENT_PATH: json.dumps(self._config), _INPUT_PATH: task.agent_prompt(), } + skill_profiles: list[dict[str, Any]] = [] + provenances: list[SkillProvenance] = [] + if self._skill_set.skills and skill_mode is not None: + if skill_mode == SKILL_MODE_CODEX_SKILLS_DIR: + _check_codex_skill_collision(self._skill_set.skills, task.inputs.get(SEED_FILES_INPUT_KEY) or {}) + seed = stage_skills_seed( + skills=self._skill_set.skills, + adapter_id=self._adapter_id(), + mode=skill_mode, + workspace_dir=_WORKSPACE_DIR, + skills_dir=_SKILLS_DIR, + existing_skill_paths=self._existing_skill_paths(), + ) + files.update(seed.files) + skill_profiles = seed.profiles + provenances = seed.provenances profile_paths: list[str] = [] - profiles = [*self._profiles, self._workspace_profile(), self._trajectory_profile()] + profiles = [*self._profiles, *skill_profiles, self._workspace_profile(), self._trajectory_profile()] for index, profile in enumerate(profiles): path = f"{_IN_DIR}/profile-{index}.yaml" files[path] = json.dumps(profile) profile_paths.append(path) - return files, profile_paths + return files, profile_paths, provenances @staticmethod def _workspace_profile() -> dict[str, Any]: @@ -241,9 +384,17 @@ def _base_metadata(self) -> dict[str, object]: return {"runtime": _RUNTIME_NAME, "image": self._image, "sandbox_provider": self._provider.name} def _to_trial( - self, task: AgentEvalTask, out_dir: Path, evidence_dir: Path, result: SandboxExecResult + self, + task: AgentEvalTask, + out_dir: Path, + evidence_dir: Path, + result: SandboxExecResult, + *, + skill_provenances: list[SkillProvenance] | None = None, ) -> AgentEvalTrial: - base_metadata = self._base_metadata() + # Skill provenance (name + content hash + injection mode) rides on every trial for the A/B diff: + # a ``skills`` list plus the historical lone ``skill`` field, matching the host FabricAgentRuntime. + base_metadata = {**self._base_metadata(), **_skill_metadata(skill_provenances or [])} # Gate on the exec outcome first: a timed-out or non-zero ``fabric run`` is untrustworthy even # when a stale/partial fabric_result.json is left behind (the shell ``>`` redirect truncates the @@ -342,6 +493,67 @@ def _to_mapping(config: FabricConfig | Mapping[str, Any]) -> dict[str, Any]: return dict(cast(Mapping[str, Any], source)) +def _skill_metadata(provenances: list[SkillProvenance]) -> dict[str, object]: + """Trial-metadata fields describing the injected skill set (the A/B provenance). + + ``skills`` is the full list of injected-skill provenances (empty = baseline). ``skill`` keeps the + historical single-provenance field — the lone provenance for a one-skill run, else ``None`` — so + single-skill consumers (e.g. ``SkillUsedMetric``) and existing trials/tests keep working unchanged. + Mirrors ``FabricAgentRuntime._skill_metadata``. + """ + return {"skill": provenances[0] if len(provenances) == 1 else None, "skills": provenances} + + +def _check_codex_skill_collision(skills: Sequence[AgentSkill], task_files: Mapping[str, object]) -> None: + """Raise if a task seed file targets the same bundle dir as a runtime-injected codex skill. + + ``.agents/skills/`` holds skills from two independent, equally valid sources: the runtime + ``skills`` parameter (the A/B knob — staged into the workspace before the sandbox starts) and + the task's own ``files`` inputs (skills the task definition always ships — uploaded after it + starts). Tasks are free to seed their own skills there; only writing the *same* + ``.agents/skills//`` from both sources is a conflict, since the task upload lands second + and would overwrite the injected bundle, leaving the stamped provenance hash describing content + the agent never saw. Fail that case rather than emit a silently mislabeled A/B trial. + """ + for skill in skills: + injected_bundle = PurePosixPath(CODEX_SKILLS_DIR) / skill.name + for rel_path in task_files: + seed = PurePosixPath(rel_path) + if seed == injected_bundle or injected_bundle in seed.parents: + raise SkillInjectionError( + f"task seed file {str(rel_path)!r} writes into {str(injected_bundle)!r}, which is " + f"also injected as the runtime skill {skill.name!r}; the task upload would overwrite " + "the injected bundle. Inject this skill via the runtime ``skills`` parameter or ship " + "it in the task's files, not both" + ) + + +def _remove_injected_bundle(workspace_dir: Path, location: str) -> None: + """Remove the Codex-injected skill subtree from a downloaded ``workspace`` dir and prune emptied parents. + + ``location`` is workspace-relative (``.agents/skills/``). Best-effort and mirrors the host + runtime's cleanup: 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. + """ + if not workspace_dir.is_dir(): + return + 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 _find_atif(relay_dir: Path) -> Path | None: # Relay nests the trajectory under a per-run subdir (relay/runtime-/trajectory-*.atif.json), # so search recursively rather than only relay's direct children. 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 e8c5fae220..c345aadc46 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 @@ -40,9 +40,10 @@ from nemo_platform.beta.evaluator.agent_eval.runtimes.fabric.skills import ( SKILL_MODE_CODEX_SKILLS_DIR, AgentSkill, + SkillMode, SkillProvenance, + SkillSet, install_skills, - require_unique_skill_names, resolve_skill_mode, ) from nemo_platform.beta.evaluator.agent_eval.tasks import AgentEvalRunConfig, AgentEvalTask @@ -139,8 +140,7 @@ def __init__( self._timeout_s = timeout_s self._capture_trajectory = capture_trajectory self._runtime_name = runtime_name - self._skills = list(skills or []) - require_unique_skill_names(self._skills) + self._skill_set = SkillSet(tuple(skills or ())) def with_skills(self, skills: Sequence[AgentSkill]) -> FabricAgentRuntime: """Return a copy of this runtime with ``skills`` *added* to its skill set; ``self`` is not modified. @@ -151,10 +151,8 @@ def with_skills(self, skills: Sequence[AgentSkill]) -> FabricAgentRuntime: set — two bundles claiming the same ``/`` would collide — so re-adding a skill already present raises. A shallow copy suffices — the shared fields are immutable config/paths. """ - combined = [*self._skills, *skills] - require_unique_skill_names(combined) clone = copy.copy(self) - clone._skills = combined + clone._skill_set = self._skill_set.with_skills(skills) return clone def with_skill(self, skill: AgentSkill) -> FabricAgentRuntime: @@ -205,8 +203,8 @@ async def run_tasks( # 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._skills: + skill_mode: SkillMode | None = None + if self._skill_set.skills: skill_mode = self._resolve_skill_mode(client, agent_config, base_profiles) if skill_mode is None: adapter_id = agent_config.harness.adapter_id @@ -231,7 +229,7 @@ def _resolve_skill_mode( client: Fabric, agent_config: FabricConfig, base_profiles: list[FabricProfileConfig], - ) -> str | None: + ) -> SkillMode | 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 @@ -269,7 +267,7 @@ async def _run_task( index: int, task: AgentEvalTask, config: AgentEvalRunConfig, - skill_mode: str | None, + skill_mode: SkillMode | None, ) -> AgentEvalTrial: # nemo_fabric is already imported+validated in ``run_tasks``; this is a cached sys.modules # lookup, not a re-load, so the types are used where they're constructed instead of threaded down. @@ -297,10 +295,10 @@ async def _run_task( # workspace and emits no overlay. One provenance per skill is stamped on the trial for the A/B # diff. Blocking file I/O, off the event loop. skill_profiles: list[FabricProfileConfig] = [] - if self._skills and skill_mode is not None: + if self._skill_set.skills and skill_mode is not None: installation = await asyncio.to_thread( install_skills, - skills=self._skills, + skills=self._skill_set.skills, adapter_id=agent_config.harness.adapter_id, mode=skill_mode, workspace_dir=workspace_dir, 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 index 975db81b68..81d9bb6f0f 100644 --- 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 @@ -41,7 +41,7 @@ from collections.abc import Mapping, Sequence from dataclasses import dataclass from pathlib import Path -from typing import TypedDict +from typing import Literal, TypedDict from pydantic import BaseModel, ConfigDict, Field, field_validator @@ -52,10 +52,15 @@ #: Name of the Fabric profile overlay that carries the native ``skills`` config. SKILL_PROFILE_NAME = "eval_skill" +#: How an injected skill reaches the selected harness (resolved from Fabric's capability plan). The two +#: runtimes thread this value from :func:`resolve_skill_mode` down to :func:`install_skill` / +#: :func:`stage_skills_seed`, so a mistyped mode is a type error rather than a silent no-op. +SkillMode = Literal["native", "codex_skills_dir"] + #: Skill reaches the harness via the native Fabric ``skills`` config (adapter accepts it). -SKILL_MODE_NATIVE = "native" +SKILL_MODE_NATIVE: SkillMode = "native" #: Skill is placed under ``/.agents/skills//`` for Codex to discover. -SKILL_MODE_CODEX_SKILLS_DIR = "codex_skills_dir" +SKILL_MODE_CODEX_SKILLS_DIR: SkillMode = "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]+)*$") @@ -120,7 +125,7 @@ class SkillProvenance(TypedDict): 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`). + mode: SkillMode #: 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). @@ -158,7 +163,7 @@ def native_skills_route(capability_plan: Mapping[str, object]) -> bool: ) -def resolve_skill_mode(*, capability_plan: Mapping[str, object], harness: str) -> str | None: +def resolve_skill_mode(*, capability_plan: Mapping[str, object], harness: str) -> SkillMode | 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 @@ -180,7 +185,7 @@ def install_skill( *, skill: AgentSkill, adapter_id: str, - mode: str, + mode: SkillMode, workspace_dir: Path, skill_stage_dir: Path, existing_skill_paths: Sequence[str] = (), @@ -259,11 +264,35 @@ def require_unique_skill_names(skills: Sequence[AgentSkill]) -> None: ) +@dataclass(frozen=True) +class SkillSet: + """Immutable, name-validated collection of :class:`AgentSkill`\\s shared by both Fabric runtimes. + + Centralizes the uniqueness check and clone-on-mutation pattern that + :class:`~...FabricAgentRuntime` and :class:`~...FabricContainerRuntime` would otherwise + duplicate: construction validates that skill names are unique; :meth:`with_skills` and + :meth:`with_skill` each return a new ``SkillSet`` without modifying ``self``. + """ + + skills: tuple[AgentSkill, ...] = () + + def __post_init__(self) -> None: + require_unique_skill_names(self.skills) + + def with_skills(self, skills: Sequence[AgentSkill]) -> SkillSet: + """Return a new ``SkillSet`` with ``skills`` appended; ``self`` is not modified.""" + return SkillSet((*self.skills, *skills)) + + def with_skill(self, skill: AgentSkill) -> SkillSet: + """Return a new ``SkillSet`` with ``skill`` appended; ``self`` is not modified.""" + return self.with_skills([skill]) + + def install_skills( *, skills: Sequence[AgentSkill], adapter_id: str, - mode: str, + mode: SkillMode, workspace_dir: Path, skill_stage_dir: Path, existing_skill_paths: Sequence[str] = (), @@ -287,6 +316,14 @@ def install_skills( staged_roots: list[Path] = [] try: for skill in skills: + # Register the target BEFORE staging: install_skill can raise after it has already written + # files (a copytree failing partway, an unreadable file while hashing), and a root recorded + # only on success would leave that partial bundle behind. A target that already exists is + # never registered — in codex mode that is a task-seeded file install_skill refuses to + # clobber, and rolling it back would delete task input this call did not create. + stage_root = _skill_stage_root(skill, mode, workspace_dir, skill_stage_dir) + if not stage_root.exists(): + staged_roots.append(stage_root) provenance = install_skill( skill=skill, adapter_id=adapter_id, @@ -296,9 +333,6 @@ def install_skills( existing_skill_paths=existing_skill_paths, ).provenance provenances.append(provenance) - # A native provenance ``location`` is the absolute staged root; a codex one is - # workspace-relative (``.agents/skills/``). Record it only after a successful stage. - staged_roots.append(_provenance_stage_root(provenance, workspace_dir)) except Exception: for root in staged_roots: shutil.rmtree(root, ignore_errors=True) @@ -319,13 +353,129 @@ def install_skills( return SkillsInstallation(profiles=profiles, provenances=provenances) -def _provenance_stage_root(provenance: SkillProvenance, workspace_dir: Path) -> Path: - """Absolute on-disk root of a staged bundle, for rollback. Native ``location`` is already absolute; - codex ``location`` is workspace-relative (``.agents/skills/``).""" - location = provenance["location"] - if provenance["mode"] == SKILL_MODE_CODEX_SKILLS_DIR: - return workspace_dir / location - return Path(location) +def _skill_stage_root(skill: AgentSkill, mode: SkillMode, workspace_dir: Path, skill_stage_dir: Path) -> Path: + """Absolute on-disk root :func:`install_skill` would stage ``skill`` into, computed before staging. + + Mirrors install_skill's per-mode placement so :func:`install_skills` can register a rollback target + up front (an unknown mode raises there, not here; the returned path is simply never created, and + rolling back a path that does not exist is a no-op).""" + if mode == SKILL_MODE_NATIVE: + return skill_stage_dir / skill.name + return workspace_dir / CODEX_SKILLS_DIR / skill.name + + +def _render_skill_seed( + *, skill: AgentSkill, adapter_id: str, mode: SkillMode, workspace_dir: str, skills_dir: str +) -> tuple[dict[str, str], SkillProvenance]: + """Render one skill bundle into an in-sandbox ``{path: text}`` seed map + its provenance. + + The per-skill core of :func:`stage_skills_seed` (the containerized counterpart of :func:`install_skill`, + which ``copytree``\\ s onto host disk): the container has no host workspace, so the bundle is read into + memory as UTF-8 text and keyed at the harness's in-sandbox discovery path — native: ``/ + /``; codex: ``/.agents/skills//``. The content hash is over the source bundle + (matching :func:`install_skill`). The caller merges these into one seed set and, for native mode, a + single ``skills`` overlay — so no per-skill overlay is emitted here. + """ + bundle = _read_text_bundle(skill.directory) + skill_hash = _hash_directory(skill.directory) + if mode == SKILL_MODE_NATIVE: + skill_root = f"{skills_dir.rstrip('/')}/{skill.name}" + files = {f"{skill_root}/{rel}": text for rel, text in bundle.items()} + return files, _provenance(skill, skill_hash, mode, adapter_id, skill_root) + + if mode == SKILL_MODE_CODEX_SKILLS_DIR: + skill_root = f"{workspace_dir.rstrip('/')}/{CODEX_SKILLS_DIR}/{skill.name}" + files = {f"{skill_root}/{rel}": text for rel, text in bundle.items()} + location = f"{CODEX_SKILLS_DIR}/{skill.name}" + return files, _provenance(skill, skill_hash, mode, adapter_id, location) + + raise SkillInjectionError(f"unknown skill injection mode {mode!r} for adapter {adapter_id!r}") + + +def _read_text_bundle(directory: Path) -> dict[str, str]: + """Read an agentskills bundle into a ``{posix_relpath: text}`` map (requires a top-level ``SKILL.md``). + + Every file is decoded as UTF-8: the containerized seed set (``SandboxSpec.files``) is text-only, so a + binary file (e.g. an image under ``assets/``) raises here rather than silently corrupting the staged + bundle — the host :func:`install_skill` path (OS-level ``copytree``) handles binary bundles instead. + """ + src = directory.expanduser() + if not (src / PRIMARY_SKILL_DOC).is_file(): + raise SkillInjectionError(f"skill directory {str(directory)!r} has no {PRIMARY_SKILL_DOC}") + bundle: dict[str, str] = {} + for path in sorted(candidate for candidate in src.rglob("*") if candidate.is_file()): + rel = path.relative_to(src).as_posix() + try: + bundle[rel] = path.read_text(encoding="utf-8") + except UnicodeDecodeError as exc: + raise SkillInjectionError( + f"skill file {rel!r} is not UTF-8 text; containerized skill injection (via the sandbox " + "seed set) supports text bundles only" + ) from exc + return bundle + + +@dataclass +class SkillsSeed: + """Result of rendering several skills into one sandbox seed set (see :func:`stage_skills_seed`). + + The plural, containerized sibling of :class:`SkillsInstallation`: + + * ``files`` — the merged ``{absolute_in_sandbox_path: text}`` seed map for every staged bundle. + * ``profiles`` — at most ONE merged native ``skills`` overlay listing every bundle (Fabric applies + ``skills.paths`` last-wins, so all must ride in a single overlay or all but the last are dropped); + the codex branch emits none. + * ``provenances`` — one entry per skill, in the given order, for the multi-skill A/B trial metadata. + """ + + files: dict[str, str] + profiles: list[dict[str, object]] + provenances: list[SkillProvenance] + + +def stage_skills_seed( + *, + skills: Sequence[AgentSkill], + adapter_id: str, + mode: SkillMode, + workspace_dir: str, + skills_dir: str, + existing_skill_paths: Sequence[str] = (), +) -> SkillsSeed: + """Render every skill in ``skills`` into one sandbox seed set + overlays for the container runtime. + + The plural, containerized sibling of :func:`install_skills`: renders each bundle (via + :func:`_render_skill_seed`) under its own ``/`` at the harness's in-sandbox discovery path, then, + for the native mode, merges the per-skill paths into a SINGLE ``skills`` overlay (Fabric applies profile + ``skills.paths`` last-wins, so one overlay per skill would silently drop all but the last). Pre-existing + ``existing_skill_paths`` are preserved ahead of the injected skills (same reason). Skill names must be + unique — their ``/`` bundles would otherwise collide. No on-disk rollback is needed (unlike + :func:`install_skills`): the seed set is an in-memory map, so a failure to render any skill just + discards the accumulated map and raises, leaving nothing staged. + """ + require_unique_skill_names(skills) + files: dict[str, str] = {} + provenances: list[SkillProvenance] = [] + for skill in skills: + rendered, provenance = _render_skill_seed( + skill=skill, adapter_id=adapter_id, mode=mode, workspace_dir=workspace_dir, skills_dir=skills_dir + ) + files.update(rendered) + provenances.append(provenance) + + profiles: list[dict[str, object]] = [] + if mode == SKILL_MODE_NATIVE and provenances: + # One merged overlay: pre-existing skills first, then each staged bundle (a native provenance's + # ``location`` is its absolute in-sandbox skill root), order-preserved and de-duplicated. + paths = list(dict.fromkeys([*existing_skill_paths, *(prov["location"] for prov in provenances)])) + profiles = [ + { + "name": SKILL_PROFILE_NAME, + "description": "Make the evaluation skills available via the native Fabric skills config.", + "skills": {"paths": paths}, + } + ] + return SkillsSeed(files=files, profiles=profiles, provenances=provenances) def _stage_bundle(directory: Path, skill_root: Path, *, reserved: bool) -> None: @@ -355,7 +505,7 @@ def _stage_bundle(directory: Path, skill_root: Path, *, reserved: bool) -> None: shutil.copytree(src, skill_root) -def _provenance(skill: AgentSkill, skill_hash: str, mode: str, adapter_id: str, location: str) -> SkillProvenance: +def _provenance(skill: AgentSkill, skill_hash: str, mode: SkillMode, adapter_id: str, location: str) -> SkillProvenance: return { "name": skill.name, "hash": skill_hash,